Uncaught exceptions in Netty pipeline

patch by norman maurer; reviewed by jasobrown for CASSANDRA-13649

CHANGES.txt

@@ -1,4 +1,5 @@
 2.2.11
+ * Uncaught exceptions in Netty pipeline (CASSANDRA-13649)
  * Prevent integer overflow on exabyte filesystems (CASSANDRA-13067) 
  * Fix queries with LIMIT and filtering on clustering columns (CASSANDRA-11223)
  * Fix potential NPE when resume bootstrap fails (CASSANDRA-13272)

src/java/org/apache/cassandra/transport/Message.java

@@ -540,10 +540,14 @@ public abstract class Message
             flusher.queued.add(item);
             flusher.start();
         }
+    }
+
+    @ChannelHandler.Sharable
+    public static final class ExceptionHandler extends ChannelInboundHandlerAdapter
+    {
 
         @Override
         public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause)
-        throws Exception
         {
             // Provide error message to client in case channel is still open
             UnexpectedChannelExceptionHandler handler = new UnexpectedChannelExceptionHandler(ctx.channel(), false);

src/java/org/apache/cassandra/transport/Server.java

@@ -270,6 +270,7 @@ public class Server implements CassandraDaemon.Server
         private static final Frame.Decompressor frameDecompressor = new Frame.Decompressor();
         private static final Frame.Compressor frameCompressor = new Frame.Compressor();
         private static final Frame.Encoder frameEncoder = new Frame.Encoder();
+        private static final Message.ExceptionHandler exceptionHandler = new Message.ExceptionHandler();
         private static final Message.Dispatcher dispatcher = new Message.Dispatcher();
         private static final ConnectionLimitHandler connectionLimitHandler = new ConnectionLimitHandler();
 
@@ -303,6 +304,14 @@ public class Server implements CassandraDaemon.Server
             pipeline.addLast("messageDecoder", messageDecoder);
             pipeline.addLast("messageEncoder", messageEncoder);
 
+            // The exceptionHandler will take care of handling exceptionCaught(...) events while still running
+            // on the same EventLoop as all previous added handlers in the pipeline. This is important as the used
+            // eventExecutorGroup may not enforce strict ordering for channel events.
+            // As the exceptionHandler runs in the EventLoop as the previous handlers we are sure all exceptions are
+            // correctly handled before the handler itself is removed.
+            // See https://issues.apache.org/jira/browse/CASSANDRA-13649
+            pipeline.addLast("exceptionHandler", exceptionHandler);
+
             pipeline.addLast(server.eventExecutorGroup, "executor", dispatcher);
         }
     }