InputStream should always be closed after reading contents from upstreams (#2501)

* InputStream should always be closed after reading contents from upstreams

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/zuul/filters/post/SendResponseFilter.java

@@ -191,12 +191,24 @@ public class SendResponseFilter extends ZuulFilter {
 		}
 		finally {
 			/**
-			* Closing the wrapping InputStream itself has no effect on closing the underlying tcp connection since it's a wrapped stream. I guess for http
-			* keep-alive. When closing the wrapping stream it tries to reach the end of the current request, which is impossible for infinite http streams. So
-			* instead of closing the InputStream we close the HTTP response.
+			* We must ensure that the InputStream provided by our upstream pooling mechanism is ALWAYS closed
+		 	* even in the case of wrapped streams, which are supplied by pooled sources such as Apache's
+		 	* PoolingHttpClientConnectionManager. In that particular case, the underlying HTTP connection will
+		 	* be returned back to the connection pool iif either close() is explicitly called, a read
+			* error occurs, or the end of the underlying stream is reached. If, however a write error occurs, we will
+			* end up leaking a connection from the pool without an explicit close()
 			*
 			* @author Johannes Edmeier
 			*/
+			if (is != null) {
+				try {
+					is.close();
+				}
+				catch (Exception ex) {
+					log.warn("Error while closing upstream input stream", ex);
+				}
+			}
+
 			try {
 				Object zuulResponse = RequestContext.getCurrentContext()
 						.get("zuulResponse");
@@ -207,7 +219,7 @@ public class SendResponseFilter extends ZuulFilter {
 				// The container will close the stream for us
 			}
 			catch (IOException ex) {
-			log.warn("Error while sending response to client: " + ex.getMessage());
+				log.warn("Error while sending response to client: " + ex.getMessage());
 			}
 		}
 	}

spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/zuul/filters/post/SendResponseFilterTests.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.netflix.zuul.filters.post;
 import java.io.ByteArrayInputStream;
 import java.io.Closeable;
 import java.io.IOException;
+import java.io.InputStream;
 import java.lang.reflect.UndeclaredThrowableException;
 
 import javax.servlet.ServletOutputStream;
@@ -48,6 +49,7 @@ import static org.mockito.Matchers.anyInt;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.springframework.cloud.netflix.zuul.filters.support.FilterConstants.X_ZUUL_DEBUG_HEADER;
@@ -115,13 +117,14 @@ public class SendResponseFilterTests {
 	}
 
 	@Test
-	public void closeResponseOutpusStreamError() throws Exception {
+	public void closeResponseOutputStreamError() throws Exception {
 		HttpServletResponse response = mock(HttpServletResponse.class);
+		InputStream mockStream = spy(new ByteArrayInputStream("Hello\n".getBytes("UTF-8")));
 
 		RequestContext context = new RequestContext();
 		context.setRequest(new MockHttpServletRequest());
 		context.setResponse(response);
-		context.setResponseDataStream(new ByteArrayInputStream("Hello\n".getBytes("UTF-8")));
+		context.setResponseDataStream(mockStream);
 		Closeable zuulResponse = mock(Closeable.class);
 		context.set("zuulResponse", zuulResponse);
 		RequestContext.testSetCurrentContext(context);
@@ -140,6 +143,29 @@ public class SendResponseFilterTests {
 		}
 
 		verify(zuulResponse).close();
+		verify(mockStream).close();
+	}
+
+	@Test
+	public void testCloseResponseDataStream() throws Exception {
+		HttpServletResponse response = mock(HttpServletResponse.class);
+		InputStream mockStream = spy(new ByteArrayInputStream("Hello\n".getBytes("UTF-8")));
+
+		RequestContext context = new RequestContext();
+		context.setRequest(new MockHttpServletRequest());
+		context.setResponse(response);
+		context.setResponseDataStream(mockStream);
+		Closeable zuulResponse = mock(Closeable.class);
+		context.set("zuulResponse", zuulResponse);
+		RequestContext.testSetCurrentContext(context);
+
+		when(response.getOutputStream()).thenReturn(mock(ServletOutputStream.class));
+
+		SendResponseFilter filter = new SendResponseFilter();
+
+		filter.run();
+
+		verify(mockStream).close();
 	}
 
 	private void runFilter(String characterEncoding, String content, boolean streamContent) throws Exception {