This is an automated email from the ASF dual-hosted git repository. mmerli pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push: new c6f49f8 Proxy should close connections immediately after exception caught (#1602) c6f49f8 is described below commit c6f49f8659a1caa0eb5aeb178196242bd85585e2 Author: Matteo Merli <mme...@apache.org> AuthorDate: Wed Apr 18 14:38:50 2018 -0700 Proxy should close connections immediately after exception caught (#1602) --- .../org/apache/pulsar/proxy/server/ProxyConnection.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java index d2c7f97..eaa006d 100644 --- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java +++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java @@ -28,6 +28,7 @@ import javax.net.ssl.SSLSession; import org.apache.pulsar.broker.authentication.AuthenticationDataCommand; import org.apache.pulsar.broker.authentication.AuthenticationDataSource; +import org.apache.pulsar.client.impl.ClientCnx; import org.apache.pulsar.common.api.Commands; import org.apache.pulsar.common.api.PulsarHandler; import org.apache.pulsar.common.api.proto.PulsarApi; @@ -88,7 +89,7 @@ public class ProxyConnection extends PulsarHandler implements FutureListener<Voi static final Counter rejectedConnections = Counter .build("pulsar_proxy_rejected_connections", "Counter for connections rejected due to throttling").create() .register(); - + public ProxyConnection(ProxyService proxyService) { super(30, TimeUnit.SECONDS); this.service = proxyService; @@ -111,7 +112,7 @@ public class ProxyConnection extends PulsarHandler implements FutureListener<Voi super.channelUnregistered(ctx); activeConnections.dec(); } - + @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); @@ -126,6 +127,16 @@ public class ProxyConnection extends PulsarHandler implements FutureListener<Voi if (directProxyHandler != null && directProxyHandler.outboundChannel != null) { directProxyHandler.outboundChannel.close(); } + + LOG.info("[{}] Connection closed", remoteAddress); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + super.exceptionCaught(ctx, cause); + LOG.warn("[{}] Got exception {} : {}", remoteAddress, cause.getClass().getSimpleName(), cause.getMessage(), + ClientCnx.isKnownException(cause) ? null : cause); + ctx.close(); } @Override -- To stop receiving notification emails like this one, please contact mme...@apache.org.