Connection leaks in Zuul "post" filters

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/zuul/ZuulConfiguration.java

@@ -33,6 +33,8 @@
 import org.springframework.cloud.netflix.zuul.filters.RouteLocator;
 import org.springframework.cloud.netflix.zuul.filters.SimpleRouteLocator;
 import org.springframework.cloud.netflix.zuul.filters.ZuulProperties;
+import org.springframework.cloud.netflix.zuul.filters.error.ReleaseResourcesErrorFilter;
+import org.springframework.cloud.netflix.zuul.filters.post.PrePostFilter;
 import org.springframework.cloud.netflix.zuul.filters.post.SendErrorFilter;
 import org.springframework.cloud.netflix.zuul.filters.post.SendResponseFilter;
 import org.springframework.cloud.netflix.zuul.filters.pre.DebugFilter;
@@ -146,6 +148,16 @@ public SendErrorFilter sendErrorFilter() {
 		return new SendErrorFilter();
 	}
 
+	@Bean
+	public PrePostFilter prePostFilter() {
+		return new PrePostFilter();
+	}
+
+	@Bean
+	public ReleaseResourcesErrorFilter resourcesCleanupilter() {
+		return new ReleaseResourcesErrorFilter();
+	}
+
 	@Bean
 	public SendForwardFilter sendForwardFilter() {
 		return new SendForwardFilter();

spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/zuul/filters/error/ReleaseResourcesErrorFilter.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2013-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.netflix.zuul.filters.error;
+
+import com.netflix.zuul.ZuulFilter;
+import com.netflix.zuul.context.RequestContext;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Releases underneath connections created by "route" filters, but not released by "post" filters.
+ *
+ * In most exceptional cases "error" filters are getting executed before "post" filters, therefore closing input stream blindly might lead to
+ * unpredictable behaviour of "post" filters.
+ *
+ * This filter coordinates work with {@link org.springframework.cloud.netflix.zuul.filters.post.PrePostFilter} to give "post" filters chance to be executed first
+ *
+ * TODO: Better fix would to close input stream in {@link com.netflix.zuul.http.ZuulServlet} or {@link org.springframework.cloud.netflix.zuul.web.ZuulController}
+ *
+ * @author Denys Kurylenko
+ */
+public class ReleaseResourcesErrorFilter extends ZuulFilter {
+
+  @Override
+  public String filterType() {
+    return "error";
+  }
+
+  @Override
+  public int filterOrder() {
+    return 0;
+  }
+
+  @Override
+  public boolean shouldFilter() {
+    // don't close connection unless "post" filters attempted to read the data
+    final RequestContext ctx = RequestContext.getCurrentContext();
+    return ctx.getResponseDataStream() != null && (ctx.getResponse().isCommitted() || ctx.getBoolean("post.filters.started"));
+  }
+
+  @Override
+  public Object run() {
+    final InputStream is = RequestContext.getCurrentContext().getResponseDataStream();
+    try {
+      if (is != null) {
+        is.close();
+      }
+    }
+    catch (IOException ex) {
+    }
+    return null;
+  }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2013-2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.netflix.zuul.filters.post;
+
+import com.netflix.zuul.ZuulFilter;
+import com.netflix.zuul.context.RequestContext;
+
+/**
+ * Indicate that "post" filter phase started.
+ *
+ * @author Denys Kurylenko
+ */
+public class PrePostFilter extends ZuulFilter {
+  @Override
+  public String filterType() {
+    return "post";
+  }
+
+  @Override
+  public int filterOrder() {
+    return -1000;
+  }
+
+  @Override
+  public boolean shouldFilter() {
+    return true;
+  }
+
+  @Override
+  public Object run() {
+    final RequestContext ctx = RequestContext.getCurrentContext();
+    ctx.set("post.filters.started", Boolean.TRUE);
+    return null;
+  }
+}

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

@@ -0,0 +1,122 @@
+package org.springframework.cloud.netflix.zuul.filters.post;
+
+import com.netflix.zuul.ZuulFilter;
+import com.netflix.zuul.context.RequestContext;
+import org.apache.commons.io.input.ProxyInputStream;
+import org.apache.commons.lang3.RandomStringUtils;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.IntegrationTest;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.boot.test.TestRestTemplate;
+import org.springframework.cloud.netflix.ribbon.RibbonClient;
+import org.springframework.cloud.netflix.ribbon.RibbonClients;
+import org.springframework.cloud.netflix.zuul.EnableZuulProxy;
+import org.springframework.cloud.netflix.zuul.filters.RouteLocator;
+import org.springframework.cloud.netflix.zuul.filters.route.support.ZuulProxyTestBase;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author Denys Kurylenko
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(classes = SendErrorFilterIntegrationTests.TestConfig.class)
+@WebAppConfiguration
+@IntegrationTest({"server.port: 0",
+                  "zuul.routes.simple: /simple/**"})
+@DirtiesContext
+public class SendErrorFilterIntegrationTests {
+
+  @Value("${local.server.port}")
+  protected int port;
+
+  @Autowired
+  RouteLocator routeLocator;
+
+  @Test
+  public void responseDataStreamGettingClosed() {
+    ResponseEntity<String> result = new TestRestTemplate().exchange(
+        "http://localhost:" + this.port + "/simple",
+        HttpMethod.GET, new HttpEntity<>((Void) null), String.class);
+    assertEquals("true", result.getHeaders().getFirst("x-response-stream-closed"));
+  }
+
+  // Don't use @SpringBootApplication because we don't want to component scan
+  @Configuration
+  @EnableAutoConfiguration
+  @RestController
+  @EnableZuulProxy
+  @RibbonClients({
+      @RibbonClient(name = "simple", configuration = ZuulProxyTestBase.SimpleRibbonClientConfiguration.class) })
+  static class TestConfig extends ZuulProxyTestBase.AbstractZuulProxyApplication
+  {
+    @Bean
+    public ZuulFilter bogusPostFilter(final SendErrorFilter sendErrorFilter)
+    {
+      return new ZuulFilter()
+      {
+        @Override
+        public boolean shouldFilter()
+        {
+          return true;
+        }
+
+        @Override
+        public Object run()
+        {
+          final RequestContext ctx = RequestContext.getCurrentContext();
+          final InputStream responseDataStream = ctx.getResponseDataStream();
+          if (responseDataStream != null)
+          {
+            ctx.setResponseDataStream(new CloseAwareInputStream(responseDataStream));
+          }
+          throw new IllegalStateException("Internal exception");
+        }
+
+        @Override
+        public String filterType()
+        {
+          return "post";
+        }
+
+        @Override
+        public int filterOrder()
+        {
+          return sendErrorFilter.filterOrder() - 10;
+        }
+      };
+    }
+  }
+
+  static class CloseAwareInputStream extends ProxyInputStream
+  {
+    public CloseAwareInputStream(InputStream proxy)
+    {
+      super(proxy);
+    }
+
+    @Override
+    public void close() throws IOException
+    {
+      RequestContext.getCurrentContext().getResponse().setHeader("x-response-stream-closed", "true");
+      super.close();
+    }
+  }
+}