Repository: netty/netty
Issue: ChannelOption for FIRE_EXCEPTION_ON_FAILURE?

I want to be able to react to *any* outbound channel failure.

I can add a handler to the tail of my pipeline that will call `promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE)` for every outbound event, but this is not sufficient.

Specifically, in the case of a `MessageToMessageEncoder` that outputs more than one object, new promises are created for all but the last object:
https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java#L112-L127

If one of those objects then triggers a failure on a subsequent outbound handler, the exception is suppressed.  There might be other scenarios where this can happen as well.

How can I react to these failures when I don't have the opportunity to inject a `ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE`?

Would it be reasonable to have a `ChannelOption` for `FIRE_EXCEPTION_ON_FAILURE`?

### Netty version

4.1.30.Final

Issue comments:
@Bennett-Lynch that's a good question.. I like the idea of the `ChannelOption` but I would need to investigate if this is easily doable without breaking any API (I suspect it should be doable).
@Bennett-Lynch https://github.com/netty/netty/pull/8448 PTAL
@normanmaurer I would like to re-open this issue, as I believe that the current implementation of making users manually inject `ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE` to writes is still cumbersome, and problematic in more complex pipelines that might have writes triggered by multiple sources.

The `MessageToMessageEncoder` fix was helpful, but I am still encountering use cases where I would like to be able to default this behavior to ON for an entire pipeline.

For example, if I have a handler that detects some special case and writes a response object to a pipeline (think `100-Continue`, where this is not the default write entry point), I have to ensure that it writes to the `Channel`, and not the `ChannelHandlerContext`, in order to pass through my `FIRE_EXCEPTION_ON_FAILURE` injecting-handler.  Otherwise, I would have to insert a `FIRE_EXCEPTION_ON_FAILURE` injecting-handler before and after every single handler, as writes may enter the pipeline from any point, and they mail fail at any point. This introduces some unnecessary coupling and is just generally awkward.

I want to be able to know about all outbound failures (even if they're my own programming fault), in a simple and transparent manner.  I'm okay paying a small performance cost for it, if it's opt-in.

@trustin @bryce-anderson @ejona86 Can I convince you of the same?  Is there a better solution that I am not thinking of?  The motivation that I'm after is fairly simple, in that I want guaranteed visibility on any outbound failure.
> I have to ensure that it writes to the Channel, and not the ChannelHandlerContext, in order to pass through my FIRE_EXCEPTION_ON_FAILURE injecting-handler

This is generally necessary. While I agree it isn't obvious to new users, any time you get it wrong is a recipe for weird things to happen.

Although I'm also not quite sure why writing to the Channel would break things in this particular case; it would eventually go through the current Handler's ctx either way.

> Otherwise, I would have to insert a FIRE_EXCEPTION_ON_FAILURE injecting-handler before and after every single handler, as writes may enter the pipeline from any point, and they mail fail at any point.

If you need it between ever handler, then I'm not sure how FIRE_EXCEPTION_ON_FAILURE would be implemented. It seems in both cases there's a problem with adding a listener for every handler in the pipeline, which would not be great.


@ejona86 My concern is that I'm essentially *required* to write to the tail of the pipeline every time.  And there are valid use cases where you don't *always* want to write to the tail end of the pipeline.  A simple encoder-like handler would be one.

If I have a pipeline like so:

`{Socket} - [ A ] - [ B ] - [ C ] - [ D ]`

Where `A` is the first inbound handler, and `D` is the tail.

Assume that writes can insert the pipeline at any of the handlers.  And assume that any of the handlers can have their write fail for some unpredictable reason.

To safely cover my "requirements" (guaranteed visibility on any outbound failure), I would need to insert a handler that injects `ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE` in between all of the handlers picture above.

This is actually doable today, but it entails incredibly clunky behavior in place of what seems like should be a simple channel option.

There's not strictly a problem with adding a listener for every handler.  Every handler can simply remove-then-add to ensure they are idempotent:

```
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    promise.removeListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
    promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
    ctx.write(msg, promise);
}
```

Besides being clunky, this probably carries unnecessary performance costs as well.
Okay. I see what you were meaning now. In any case a "ChannelOption for FIRE_EXCEPTION_ON_FAILURE" wouldn't fix this behavior; the Channel has just as hard of a time managing the future failure as you are having now.

The only idea I have would be for {ctx, channel, pipeline}.newPromise() to return a promise with the listener already added. The main way to "break" that behavior would be to construct a DefaultChannelPromise manually, which would be pretty rare. But having a promise with an auto-installed listener seems a bit too magical.
@ejona86 The ask isn't so much to magically auto-install a listener, but to simply offer a channel option that will revert the behavior back to how it used to behave in Netty 3: https://github.com/netty/netty/pull/8448

Pull request review comments:

Patch that resolved the issue:
diff --git a/transport/src/main/java/io/netty/channel/AbstractChannelHandlerContext.java b/transport/src/main/java/io/netty/channel/AbstractChannelHandlerContext.java
index 0a81bd6b0f5..1b3058e4e4d 100644
--- a/transport/src/main/java/io/netty/channel/AbstractChannelHandlerContext.java
+++ b/transport/src/main/java/io/netty/channel/AbstractChannelHandlerContext.java
@@ -878,12 +878,13 @@ private static boolean inExceptionCaught(Throwable cause) {
 
     @Override
     public ChannelPromise newPromise() {
-        return new DefaultChannelPromise(channel(), executor());
+        return DefaultChannelPipeline.fireExceptionOndFailure(new DefaultChannelPromise(channel(), executor()));
     }
 
     @Override
     public ChannelProgressivePromise newProgressivePromise() {
-        return new DefaultChannelProgressivePromise(channel(), executor());
+        return DefaultChannelPipeline.fireExceptionOndFailure(
+                new DefaultChannelProgressivePromise(channel(), executor()));
     }
 
     @Override
@@ -897,7 +898,7 @@ public ChannelFuture newSucceededFuture() {
 
     @Override
     public ChannelFuture newFailedFuture(Throwable cause) {
-        return new FailedChannelFuture(channel(), executor(), cause);
+        return DefaultChannelPipeline.fireExceptionOndFailure(new FailedChannelFuture(channel(), executor(), cause));
     }
 
     private boolean isNotValidPromise(ChannelPromise promise, boolean allowVoidPromise) {
diff --git a/transport/src/main/java/io/netty/channel/ChannelOption.java b/transport/src/main/java/io/netty/channel/ChannelOption.java
index 97bf31545c4..626cf700423 100644
--- a/transport/src/main/java/io/netty/channel/ChannelOption.java
+++ b/transport/src/main/java/io/netty/channel/ChannelOption.java
@@ -129,6 +129,12 @@ public static <T> ChannelOption<T> newInstance(String name) {
     public static final ChannelOption<Boolean> SINGLE_EVENTEXECUTOR_PER_GROUP =
             valueOf("SINGLE_EVENTEXECUTOR_PER_GROUP");
 
+    /**
+     * If {@code true} {@link ChannelPipeline#fireExceptionCaught(Throwable)} is called whenever a {@link ChannelFuture}
+     * is failed.
+     */
+    public static final ChannelOption<Boolean> FIRE_EXCEPTION_ON_FAILURE = valueOf("FIRE_EXCEPTION_ON_FAILURE");
+
     /**
      * Creates a new {@link ChannelOption} with the specified unique {@code name}.
      */
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
index 4118708637c..198e5580596 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelConfig.java
@@ -23,18 +23,7 @@
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
-import static io.netty.channel.ChannelOption.ALLOCATOR;
-import static io.netty.channel.ChannelOption.AUTO_CLOSE;
-import static io.netty.channel.ChannelOption.AUTO_READ;
-import static io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS;
-import static io.netty.channel.ChannelOption.MAX_MESSAGES_PER_READ;
-import static io.netty.channel.ChannelOption.MESSAGE_SIZE_ESTIMATOR;
-import static io.netty.channel.ChannelOption.RCVBUF_ALLOCATOR;
-import static io.netty.channel.ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP;
-import static io.netty.channel.ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK;
-import static io.netty.channel.ChannelOption.WRITE_BUFFER_LOW_WATER_MARK;
-import static io.netty.channel.ChannelOption.WRITE_BUFFER_WATER_MARK;
-import static io.netty.channel.ChannelOption.WRITE_SPIN_COUNT;
+import static io.netty.channel.ChannelOption.*;
 import static io.netty.util.internal.ObjectUtil.checkNotNull;
 
 /**
@@ -64,6 +53,7 @@ public class DefaultChannelConfig implements ChannelConfig {
     private volatile boolean autoClose = true;
     private volatile WriteBufferWaterMark writeBufferWaterMark = WriteBufferWaterMark.DEFAULT;
     private volatile boolean pinEventExecutor = true;
+    private volatile boolean fireExceptionOnFailure;
 
     public DefaultChannelConfig(Channel channel) {
         this(channel, new AdaptiveRecvByteBufAllocator());
@@ -82,7 +72,7 @@ public Map<ChannelOption<?>, Object> getOptions() {
                 CONNECT_TIMEOUT_MILLIS, MAX_MESSAGES_PER_READ, WRITE_SPIN_COUNT,
                 ALLOCATOR, AUTO_READ, AUTO_CLOSE, RCVBUF_ALLOCATOR, WRITE_BUFFER_HIGH_WATER_MARK,
                 WRITE_BUFFER_LOW_WATER_MARK, WRITE_BUFFER_WATER_MARK, MESSAGE_SIZE_ESTIMATOR,
-                SINGLE_EVENTEXECUTOR_PER_GROUP);
+                SINGLE_EVENTEXECUTOR_PER_GROUP, FIRE_EXCEPTION_ON_FAILURE);
     }
 
     protected Map<ChannelOption<?>, Object> getOptions(
@@ -156,6 +146,9 @@ public <T> T getOption(ChannelOption<T> option) {
         if (option == SINGLE_EVENTEXECUTOR_PER_GROUP) {
             return (T) Boolean.valueOf(getPinEventExecutorPerGroup());
         }
+        if (option == FIRE_EXCEPTION_ON_FAILURE) {
+            return (T) Boolean.valueOf(getFireExceptionOnFailure());
+        }
         return null;
     }
 
@@ -188,6 +181,8 @@ public <T> boolean setOption(ChannelOption<T> option, T value) {
             setMessageSizeEstimator((MessageSizeEstimator) value);
         } else if (option == SINGLE_EVENTEXECUTOR_PER_GROUP) {
             setPinEventExecutorPerGroup((Boolean) value);
+        } else if (option == FIRE_EXCEPTION_ON_FAILURE) {
+            setFireExceptionOnFailure((Boolean) value);
         } else {
             return false;
         }
@@ -436,4 +431,13 @@ private boolean getPinEventExecutorPerGroup() {
         return pinEventExecutor;
     }
 
+    private ChannelConfig setFireExceptionOnFailure(boolean fireExceptionOnFailure) {
+        this.fireExceptionOnFailure = fireExceptionOnFailure;
+        return this;
+    }
+
+    // Package-private to allow fast access without using ChannelOption in DefaultChannelPipeline.
+    boolean getFireExceptionOnFailure() {
+        return fireExceptionOnFailure;
+    }
 }
diff --git a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
index 0d3307a5351..207d3cbdd0d 100644
--- a/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
+++ b/transport/src/main/java/io/netty/channel/DefaultChannelPipeline.java
@@ -67,6 +67,7 @@ protected Map<Class<?>, String> initialValue() throws Exception {
     private final Channel channel;
     private final ChannelFuture succeededFuture;
     private final VoidChannelPromise voidPromise;
+
     private final boolean touch = ResourceLeakDetector.isEnabled();
 
     private Map<EventExecutorGroup, EventExecutor> childExecutors;
@@ -1073,12 +1074,20 @@ public final ChannelFuture writeAndFlush(Object msg) {
 
     @Override
     public final ChannelPromise newPromise() {
-        return new DefaultChannelPromise(channel);
+        return fireExceptionOndFailure(new DefaultChannelPromise(channel));
+    }
+
+    static <F extends ChannelFuture> F fireExceptionOndFailure(F future) {
+        ChannelConfig config = future.channel().config();
+        if (config instanceof DefaultChannelConfig && ((DefaultChannelConfig) config).getFireExceptionOnFailure()) {
+            future.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
+        }
+        return future;
     }
 
     @Override
     public final ChannelProgressivePromise newProgressivePromise() {
-        return new DefaultChannelProgressivePromise(channel);
+        return fireExceptionOndFailure(new DefaultChannelProgressivePromise(channel));
     }
 
     @Override
@@ -1088,11 +1097,13 @@ public final ChannelFuture newSucceededFuture() {
 
     @Override
     public final ChannelFuture newFailedFuture(Throwable cause) {
-        return new FailedChannelFuture(channel, null, cause);
+        return fireExceptionOndFailure(new FailedChannelFuture(channel, null, cause));
     }
 
     @Override
     public final ChannelPromise voidPromise() {
+        // The voidPromise always calls fireExceptionCaught(...) so we do not need to take
+        // ChannelOption.FIRE_EXCEPTION_ON_FAILURE into account.
         return voidPromise;
     }
 


Test patch: