Merge branch '5.0.x'

Author: spencergibb

spring-cloud-gateway-server-webflux/src/main/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilter.java

@@ -45,6 +45,7 @@
 
 /**
  * @author Spencer Gibb
+ * @author Seungbin Ko
  */
 public class NettyWriteResponseFilter implements GlobalFilter, Ordered {
 
@@ -92,12 +93,21 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
 					if (connection == null) {
 						return Mono.empty();
 					}
+					ServerHttpResponse response = exchange.getResponse();
+					if (response.isCommitted()) {
+						if (log.isDebugEnabled()) {
+							log.debug("NettyWriteResponseFilter response already committed, discarding inbound: "
+									+ connection.channel().id().asShortText() + ", outbound: "
+									+ exchange.getLogPrefix());
+						}
+						cleanup(exchange);
+						return Mono.empty();
+					}
 					if (log.isTraceEnabled()) {
 						log.trace("NettyWriteResponseFilter start inbound: "
 								+ connection.channel().id().asShortText() + ", outbound: "
 								+ exchange.getLogPrefix());
 					}
-					ServerHttpResponse response = exchange.getResponse();
 
 					// TODO: needed?
 					final Flux<DataBuffer> body = connection

spring-cloud-gateway-server-webflux/src/test/java/org/springframework/cloud/gateway/filter/NettyWriteResponseFilterTests.java

@@ -21,17 +21,26 @@
 
 import io.netty.buffer.ByteBuf;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.netty.Connection;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.core.io.buffer.PooledDataBuffer;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
 import org.springframework.mock.http.server.reactive.MockServerHttpResponse;
+import org.springframework.mock.web.server.MockServerWebExchange;
 
 import static io.netty.buffer.PooledByteBufAllocator.DEFAULT;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.CLIENT_RESPONSE_CONN_ATTR;
 
 /**
  * @author Violeta Georgieva
+ * @author Seungbin Ko
  */
 public class NettyWriteResponseFilterTests {
 
@@ -45,6 +54,20 @@ public void testWrap_DefaultDataBufferFactory() {
 		doTestWrap(new MockServerHttpResponse());
 	}
 
+	@Test
+	public void committedResponseDisposesConnectionWithoutWriting() {
+		NettyWriteResponseFilter filter = new NettyWriteResponseFilter(new ArrayList<>(), null);
+		MockServerWebExchange exchange = MockServerWebExchange.from(MockServerHttpRequest.get("/").build());
+		exchange.getResponse().setComplete().block();
+		Connection connection = mock(Connection.class);
+		exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
+
+		filter.filter(exchange, ex -> Mono.empty()).block();
+
+		verify(connection).dispose();
+		verify(connection, never()).inbound();
+	}
+
 	private void doTestWrap(MockServerHttpResponse response) {
 		NettyWriteResponseFilter filter = new NettyWriteResponseFilter(new ArrayList<>(), null);