Close response on errors writing to the outputStream (#1372)

Currently all errors during writing to the outputStream are ignored. So the
response is read to the end. This is a problem when the response is
infite or very long running (`hystrix.stream` for example) and results in the
socket never being closed and eating up threads (on zuul and the proxied
backend). With this commit the errors during writing are not ignored resulting
in closing the response.

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

@@ -17,6 +17,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.io.OutputStream;
@@ -184,28 +185,33 @@ else if (context.getResponseGZipped() && isGzipRequested) {
 			}
 		}
 		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.
+			*
+			* @author Johannes Edmeier
+			*/
 			try {
-				if (is != null) {
-					is.close();
+				Object zuulResponse = RequestContext.getCurrentContext()
+						.get("zuulResponse");
+				if (zuulResponse instanceof Closeable) {
+					((Closeable) zuulResponse).close();
 				}
 				outStream.flush();
 				// The container will close the stream for us
 			}
 			catch (IOException ex) {
+			log.warn("Error while sending response to client: " + ex.getMessage());
 			}
 		}
 	}
 
 	private void writeResponse(InputStream zin, OutputStream out) throws Exception {
-		try {
-			byte[] bytes = buffers.get();
-			int bytesRead = -1;
-			while ((bytesRead = zin.read(bytes)) != -1) {
-				out.write(bytes, 0, bytesRead);
-			}
-		}
-		catch(IOException ioe) {
-			log.warn("Error while sending response to client: "+ioe.getMessage());
+		byte[] bytes = buffers.get();
+		int bytesRead = -1;
+		while ((bytesRead = zin.read(bytes)) != -1) {
+			out.write(bytes, 0, bytesRead);
 		}
 	}
 

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/zuul/filters/route/RibbonRoutingFilter.java

@@ -218,6 +218,7 @@ protected String getVerb(HttpServletRequest request) {
 
 	protected void setResponse(ClientHttpResponse resp)
 			throws ClientException, IOException {
+		RequestContext.getCurrentContext().set("zuulResponse", resp);
 		this.helper.setResponse(resp.getStatusCode().value(),
 				resp.getBody() == null ? null : resp.getBody(), resp.getHeaders());
 	}

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/zuul/filters/route/SimpleHostRoutingFilter.java

@@ -42,10 +42,10 @@
 import org.apache.http.HttpRequest;
 import org.apache.http.HttpResponse;
 import org.apache.http.ProtocolException;
-import org.apache.http.client.HttpClient;
 import org.apache.http.client.RedirectStrategy;
 import org.apache.http.client.config.CookieSpecs;
 import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPatch;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
@@ -192,8 +192,8 @@ public Object run() {
 		this.helper.addIgnoredHeaders();
 
 		try {
-			HttpResponse response = forward(this.httpClient, verb, uri, request, headers,
-					params, requestEntity);
+			CloseableHttpResponse response = forward(this.httpClient, verb, uri, request,
+					headers, params, requestEntity);
 			setResponse(response);
 		}
 		catch (Exception ex) {
@@ -278,8 +278,8 @@ public HttpUriRequest getRedirect(HttpRequest request,
 				}).build();
 	}
 
-	private HttpResponse forward(HttpClient httpclient, String verb, String uri,
-			HttpServletRequest request, MultiValueMap<String, String> headers,
+	private CloseableHttpResponse forward(CloseableHttpClient httpclient, String verb,
+			String uri, HttpServletRequest request, MultiValueMap<String, String> headers,
 			MultiValueMap<String, String> params, InputStream requestEntity)
 			throws Exception {
 		Map<String, Object> info = this.helper.debug(verb, uri, headers, params,
@@ -301,7 +301,8 @@ private HttpResponse forward(HttpClient httpclient, String verb, String uri,
 		try {
 			log.debug(httpHost.getHostName() + " " + httpHost.getPort() + " "
 					+ httpHost.getSchemeName());
-			HttpResponse zuulResponse = forwardRequest(httpclient, httpHost, httpRequest);
+			CloseableHttpResponse zuulResponse = forwardRequest(httpclient, httpHost,
+					httpRequest);
 			this.helper.appendDebug(info, zuulResponse.getStatusLine().getStatusCode(),
 					revertHeaders(zuulResponse.getAllHeaders()));
 			return zuulResponse;
@@ -379,8 +380,8 @@ private Header[] convertHeaders(MultiValueMap<String, String> headers) {
 		return list.toArray(new BasicHeader[0]);
 	}
 
-	private HttpResponse forwardRequest(HttpClient httpclient, HttpHost httpHost,
-			HttpRequest httpRequest) throws IOException {
+	private CloseableHttpResponse forwardRequest(CloseableHttpClient httpclient,
+			HttpHost httpHost, HttpRequest httpRequest) throws IOException {
 		return httpclient.execute(httpHost, httpRequest);
 	}
 
@@ -407,6 +408,7 @@ private String getVerb(HttpServletRequest request) {
 	}
 
 	private void setResponse(HttpResponse response) throws IOException {
+		RequestContext.getCurrentContext().set("zuulResponse", response);
 		this.helper.setResponse(response.getStatusLine().getStatusCode(),
 				response.getEntity() == null ? null : response.getEntity().getContent(),
 				revertHeaders(response.getAllHeaders()));

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

@@ -16,9 +16,26 @@
 
 package org.springframework.cloud.netflix.zuul.filters.post;
 
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+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.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.cloud.netflix.zuul.filters.support.FilterConstants.X_ZUUL_DEBUG_HEADER;
+import static org.hamcrest.Matchers.equalTo;
+
 import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.IOException;
+import java.lang.reflect.UndeclaredThrowableException;
 
+import javax.servlet.ServletOutputStream;
 import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
 
 import org.junit.After;
 import org.junit.Before;
@@ -34,11 +51,6 @@
 import com.netflix.zuul.context.Debug;
 import com.netflix.zuul.context.RequestContext;
 
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.springframework.cloud.netflix.zuul.filters.support.FilterConstants.X_ZUUL_DEBUG_HEADER;
-
 /**
  * @author Spencer Gibb
  */
@@ -96,6 +108,34 @@ public void runWithOriginContentLength() throws Exception {
 		assertThat("wrong origin content length", contentLength, equalTo("6"));
 	}
 
+    @Test
+	public void closeResponseOutpusStreamError() throws Exception {
+        HttpServletResponse response = mock(HttpServletResponse.class);
+
+        RequestContext context = new RequestContext();
+        context.setRequest(new MockHttpServletRequest());
+        context.setResponse(response);
+        context.setResponseDataStream(new ByteArrayInputStream("Hello\n".getBytes("UTF-8")));
+        Closeable zuulResponse = mock(Closeable.class);
+        context.set("zuulResponse", zuulResponse);
+        RequestContext.testSetCurrentContext(context);
+
+        SendResponseFilter filter = new SendResponseFilter();
+
+        ServletOutputStream zuuloutputstream = mock(ServletOutputStream.class);
+        doThrow(new IOException("Response to client closed")).when(zuuloutputstream).write(isA(byte[].class), anyInt(), anyInt());
+
+        when(response.getOutputStream()).thenReturn(zuuloutputstream);
+
+        try {
+            filter.run();
+        } catch (UndeclaredThrowableException ex) {
+            assertThat(ex.getUndeclaredThrowable().getMessage(), is("Response to client closed"));
+        }
+
+        verify(zuulResponse).close();
+    }
+
 	private void runFilter(String characterEncoding, String content, boolean streamContent) throws Exception {
 		MockHttpServletResponse response = new MockHttpServletResponse();
 		SendResponseFilter filter = createFilter(content, characterEncoding, response, streamContent);