Catch-all exception handling for outbound ChannelHandler

Question

In Netty you have the concept of inbound and outbound handlers. A catch-all inbound exception handler is implemented simply by adding a channel handler at the end (the tail) of the pipeline and implementing an exceptionCaught override. The exception happening along the inbound pipeline will travel along the handlers until meeting the last one, if not handled along the way.

There isn't an exact opposite for outgoing handlers. Instead (according to Netty in Action, page 94) you need to either add a listener to the channel's Future or a listener to the Promise passed into the write method of your Handler.

As I am not sure where to insert the former, I thought I'd go for the latter, so I made the following ChannelOutboundHandler:

    /**
     * Catch and log errors happening in the outgoing direction
     *
     * @see <p>p94 in "Netty In Action"</p>
     */
    private ChannelOutboundHandlerAdapter createOutgoingErrorHandler() {
        return new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
                logger.info("howdy! (never gets this far)");

                final ChannelFutureListener channelFutureListener = future -> {
                    if (!future.isSuccess()) {
                        future.cause().printStackTrace();
                        // ctx.writeAndFlush(serverErrorJSON("an error!"));
                        future.channel().writeAndFlush(serverErrorJSON("an error!"));
                        future.channel().close();
                    }
                };
                promise.addListener(channelFutureListener);
                ctx.write(msg, promise);
            }
        };

This is added to the head of the pipeline:

    @Override
    public void addHandlersToPipeline(final ChannelPipeline pipeline) {
        pipeline.addLast(
                createOutgoingErrorHandler(),
                new HttpLoggerHandler(), // an error in this `write` should go "up"
                authHandlerFactory.get(),
                // etc

The problem is that the write method of my error handler is never called if I throw a runtime exception in the HttpLoggerHandler.write().

How would I make this work? An error in any of the outgoing handlers should "bubble up" to the one attached to the head.

An important thing to note is that I don't merely want to close the channel, I want to write an error message back to the client (as seen from serverErrorJSON('...'). During my trials of shuffling around the order of the handlers (also trying out stuff from this answer), I have gotten the listener activated, but I was unable to write anything. If I used ctx.write() in the listener, it seems as if I got into a loop, while using future.channel().write... didn't do anything.

Answer 1

I found a very simple solution that allows both inbound and outbound exceptions to reach the same exception handler positioned as the last ChannelHandler in the pipeline.

My pipeline is setup as follows:

        //Inbound propagation
        socketChannel.pipeline()
          .addLast(new Decoder())
          .addLast(new ExceptionHandler());

        //Outbound propagation
        socketChannel.pipeline()
          .addFirst(new OutboundExceptionRouter())
          .addFirst(new Encoder());

This is the content of my ExceptionHandler, it logs caught exceptions:

    public class ExceptionHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            log.error("Exception caught on channel", cause);
        }
    }

Now the magic that allows even outbound exceptions to be handled by ExceptionHandler happens in the OutBoundExceptionRouter:

    public class OutboundExceptionRouter extends ChannelOutboundHandlerAdapter {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
            super.write(ctx, msg, promise);
        }
    }

This is the first outbound handler invoked in my pipeline, what it does is add a listener to the outbound write promise which will execute future.channel().pipeline().fireExceptionCaught(future.cause()); when the promise fails. The fireExceptionCaught method propagates the exception through the pipeline in the inbound direction, eventually reaching the ExceptionHandler.

---

In case anyone is interested, as of Netty 4.1, the reason why we need to add a listener to get the exception is because after performing a writeAndFlush to the channel, the invokeWrite0 method is called in AbstractChannelHandlerContext.java which wraps the write operation in a try catch block. The catch block notifies the Promise instead of calling fireExceptionCaught like the invokeChannelRead method does for inbound messages.

Answer 2

Basically what you did is correct... The only thing that is not correct is the order of the handlers. Your ChannelOutboundHandlerAdapter mast be placed "as last outbound handler" in the pipeline. Which means it should be like this:

pipeline.addLast(
        new HttpLoggerHandler(),
        createOutgoingErrorHandler(),
        authHandlerFactory.get());

The reason for this is that outbound events from from the tail to the head of the pipeline while inbound events flow from the head to the tail.

Answer 3

There does not seem to be a generalized concept of a catch-all exception handler for outgoing handlers that will catch errors regardless of where. This means, unless you registered a listener to catch a certain error a runtime error will probably result in the error being "swallowed", leaving you scratching your head for why nothing is being returned.

That said, maybe it doesn't make sense to have a handler/listener that always will execute given an error (as it needs to be very general), but it does make logging errors a bit tricker than need be.

After writing a bunch of learning tests (which I suggest checking out!) I ended up with these insights, which are basically the names of my JUnit tests (after some regex manipulation):

• a listener can write to a channel after the parent write has completed
• a write listener can remove listeners from the pipeline and write on an erronous write
• all listeners are invoked on success if the same promise is passed on
• an error handler near the tail cannot catch an error from a handler nearer the head
• netty does not invoke the next handlers write on runtime exception
• netty invokes a write listener once on a normal write
• netty invokes a write listener once on an erronous write
• netty invokes the next handlers write with its written message
• promises can be used to listen for next handlers success or failure
• promises can be used to listen for non immediate handlers outcome if the promise is passed on
• promises cannot be used to listen for non immediate handlers outcome if a new promise is passed on
• promises cannot be used to listen for non immediate handlers outcome if the promise is not passed on
• only the listener added to the final write is invoked on error if the promise is not passed on
• only the listener added to the final write is invoked on success if the promise is not passed on
• write listeners are invoked from the tail

This insight means, given the example in the question, that if an error should arise near the tail and authHandler does not pass the promise on, then the error handler near the head will never be invoked, as it is being supplied with a new promise, as ctx.write(msg) is essentially ctx.channel.write(msg, newPromise()).

In our situation we ended up solving the situation by injecting the same shareable error handling inbetween all the business logic handlers.

The handler looked like this

@ChannelHandler.Sharable
class OutboundErrorHandler extends ChannelOutboundHandlerAdapter {

    private final static Logger logger = LoggerFactory.getLogger(OutboundErrorHandler.class);
    private Throwable handledCause = null;

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        ctx.write(msg, promise).addListener(writeResult -> handleWriteResult(ctx, writeResult));
    }

    private void handleWriteResult(ChannelHandlerContext ctx, Future<?> writeResult) {
        if (!writeResult.isSuccess()) {
            final Throwable cause = writeResult.cause();

            if (cause instanceof ClosedChannelException) {
                // no reason to close an already closed channel - just ignore
                return;
            }

            // Since this handler is shared and added multiple times
            // we need to avoid spamming the logs N number of times for the same error
            if (handledCause == cause) return;
            handledCause = cause;

            logger.error("Uncaught exception on write!", cause);

            // By checking on channel writability and closing the channel after writing the error message,
            // only the first listener will signal the error to the client
            final Channel channel = ctx.channel();
            if (channel.isWritable()) {
                ctx.writeAndFlush(serverErrorJSON(cause.getMessage()), channel.newPromise());
                ctx.close();
            }
        }
    }
}

Then in our pipeline setup we have this

// Prepend the error handler to every entry in the pipeline. 
// The intention behind this is to have a catch-all
// outbound error handler and thereby avoiding the need to attach a
// listener to every ctx.write(...).
final OutboundErrorHandler outboundErrorHandler = new OutboundErrorHandler();
for (Map.Entry<String, ChannelHandler> entry : pipeline) {
    pipeline.addBefore(entry.getKey(), entry.getKey() + "#OutboundErrorHandler", outboundErrorHandler);
}