more reads and relax detector

Author: mhl-b

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/MissingReadDetector.java

@@ -13,44 +13,49 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.concurrent.ScheduledFuture;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.time.TimeProvider;
 import org.elasticsearch.common.util.concurrent.FutureUtils;
 
 import java.util.concurrent.TimeUnit;
 
 /**
  * When channel auto-read is disabled handlers are responsible to read from channel.
- * But it's hard to detect when read is missing. This helper class throws assertion errors
+ * But it's hard to detect when read is missing. This helper class print warnings
  * when no reads where detected in given time interval. Normally, in tests, 30 seconds is enough
  * to avoid test hang for too long, but can be increased if needed.
  */
 class MissingReadDetector extends ChannelDuplexHandler {
-    final long interval;
-    final TimeProvider timer;
-    long reqTimeMs;
-    long respTimeMs;
-    ScheduledFuture<?> checker;
+
+    private static final Logger logger = LogManager.getLogger(MissingReadDetector.class);
+
+    private final long interval;
+    private final TimeProvider timer;
+    private long reqTimeMs;
+    private long respTimeMs;
+    private ScheduledFuture<?> checker;
 
     MissingReadDetector(TimeProvider timer, long missingReadInterval) {
         this.interval = missingReadInterval;
         this.timer = timer;
     }
 
     @Override
-    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
+    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         checker = ctx.channel().eventLoop().scheduleAtFixedRate(() -> {
             if (respTimeMs >= reqTimeMs) { // stale read
                 long now = timer.absoluteTimeInMillis();
                 if (now >= respTimeMs + interval) {
-                    ctx.fireExceptionCaught(new AssertionError("stale channel, no reads for " + (now - respTimeMs) + " ms"));
+                    logger.warn("haven't read from channel for {}", (now - respTimeMs));
                 }
             }
         }, interval, interval, TimeUnit.MILLISECONDS);
         super.channelRegistered(ctx);
     }
 
     @Override
-    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
         if (checker != null) {
             FutureUtils.cancel(checker);
         }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpContentSizeHandler.java

@@ -123,6 +123,7 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request) {
                 isContinueExpected = true;
             } else {
                 ctx.writeAndFlush(EXPECTATION_FAILED_CLOSE.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE);
+                ctx.read();
                 return;
             }
         }
@@ -136,6 +137,7 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request) {
                 decoder.reset();
             }
             ctx.writeAndFlush(TOO_LARGE.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
+            ctx.read();
         } else {
             ignoreContent = false;
             currentContentLength = 0;
@@ -155,6 +157,7 @@ private void handleContent(ChannelHandlerContext ctx, HttpContent msg) {
             if (currentContentLength > maxContentLength) {
                 msg.release();
                 ctx.writeAndFlush(TOO_LARGE_CLOSE.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE);
+                ctx.read();
             } else {
                 ctx.fireChannelRead(msg);
             }

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandler.java

@@ -509,6 +509,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         } else {
             serverTransport.onException(channel, (Exception) cause);
         }
+        ctx.read();
     }
 
     @Override

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -371,7 +371,7 @@ protected HttpMessage createMessage(String[] initialLine) throws Exception {
             ch.pipeline().addLast(new FlowControlHandler());
             if (Assertions.ENABLED) {
                 // missing reads are hard to catch, but we can detect absence of reads within interval
-                long missingReadIntervalMs = 30_000;
+                long missingReadIntervalMs = 10_000;
                 ch.pipeline().addLast(new MissingReadDetector(transport.threadPool, missingReadIntervalMs));
             }
             // disable auto-read and issue first read, following reads must come from handlers

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4HttpServerTransportCloseNotifyTests.java

@@ -251,7 +251,7 @@ public void close() {
             server.dispatcher.reqQueue.forEach(r -> r.request.getHttpRequest().release());
             server.netty.stop();
             server.threadPool.shutdownNow();
-            safeAwait(client.netty.config().group().shutdownGracefully());
+            safeAwait(client.netty.config().group().shutdownGracefully(0,0, TimeUnit.SECONDS));
         }
     }