Repository: netty/netty
Issue: ChunkedWriteHandler may report a failed write as a successful one
Labels: defect

### Expected behavior

If the writing of the last chunk of the chunked input message failed, the corresponding promise should be failed with an appropriate exception. 

### Actual behavior

`operationComplete` callback [reports success](https://github.com/netty/netty/blob/fa84e2b3af45ec7fd47909eff0aa7d2be5a54972/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L268) with no check of the `future.isSuccesful()`. If I'm not missing anything and this was not done on purpose, I'll submit a PR with a fix.

### Steps to reproduce

n/a

### Minimal yet complete reproducer code (or URL to code)

n/a

### Netty version

4.1.32

### JVM version (e.g. `java -version`)

n/a

### OS version (e.g. `uname -a`)

n/a

Issue comments:
Another somewhat related question.

When using `HttpChunkedInput` with `ChunkedWriteHandler`, `lastHttpContent` would be written/flushed even in case write promise is already reported as a failed one. We rely on `endOfInput` [being](https://github.com/netty/netty/blob/fa84e2b3af45ec7fd47909eff0aa7d2be5a54972/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L222) `true` after `closeInput` was called [here](https://github.com/netty/netty/blob/fa84e2b3af45ec7fd47909eff0aa7d2be5a54972/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L277) or [here](https://github.com/netty/netty/blob/fa84e2b3af45ec7fd47909eff0aa7d2be5a54972/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L289), which [doesn't work](https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkedInput.java#L72-L74) for  `HttpChunkedInput`.
@kachayev yes this is a bug... would you mind open a PR with a fix ?
@normanmaurer Sure! Will do shortly.

Pull request review comments:
do we also need to release the message ?
Not here, I think... I assume someone who resolved the promise (either with success or with failure) have already done this by calling `closeInput`. That's how the code is organized now. As we're not allocating any new chunks (calling `readChunk`), everything else should be the responsibility of a specific input implementation.
good point. Maybe add a comment ?
Will do!
You also need to call `ReferenceCountUtil.release(msg)` before failing the promise. 
just use `ch.finish()` and assert the return value.
just use `ch.finish()` and assert the return value.
You also need to call `ReferenceCountUtil.release(msg)` before failing the promise.
just use `ch.finish()` and assert the return value.
Call `ReferenceCountUtil.release(msg)` before failing the promise.
just use `ch.finish()` and assert the return value.
call `ReferenceCountUtil.release(msg)` before failing the promise.
just use `ch.finish()` and assert the return value.
@normanmaurer `ch.finish` would return `false` because we effectively don't do any writes here. All messages are still in `ChunkedWriteHandler` queue, until `flush`. Meaning both `inboundMessages` and `outboundMessages` queues are empty.
@kachayev sure which is fine just use `assertFalse(...)` as this is what you expect, or what I am missing ?
@normanmaurer Oh, yeah. You're right! 🤦‍♂️It's just mentally hard to read `assertFalse(ch.finish())` as "all good, just no messages" :) I'll leave a comment why this is expected.
@normanmaurer Updated once again. I also "fixed" a couple of older tests to be consistent in terms of `ch.finish()` usage.

Patch that resolved the issue:
diff --git a/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java b/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java
index b4805d98ef0..f39328dff7a 100644
--- a/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java
+++ b/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java
@@ -209,6 +209,21 @@ private void doFlush(final ChannelHandlerContext ctx) {
             if (currentWrite == null) {
                 break;
             }
+
+            if (currentWrite.promise.isDone()) {
+                // This might happen e.g. in the case when a write operation
+                // failed, but there're still unconsumed chunks left.
+                // Most chunked input sources would stop generating chunks
+                // and report end of input, but this doesn't work with any
+                // source wrapped in HttpChunkedInput.
+                // Note, that we're not trying to release the message/chunks
+                // as this had to be done already by someone who resolved the
+                // promise (using ChunkedInput.close method).
+                // See https://github.com/netty/netty/issues/8700.
+                this.currentWrite = null;
+                continue;
+            }
+
             final PendingWrite currentWrite = this.currentWrite;
             final Object pendingMessage = currentWrite.msg;
 
@@ -264,9 +279,13 @@ private void doFlush(final ChannelHandlerContext ctx) {
                     f.addListener(new ChannelFutureListener() {
                         @Override
                         public void operationComplete(ChannelFuture future) throws Exception {
-                            currentWrite.progress(chunks.progress(), chunks.length());
-                            currentWrite.success(chunks.length());
-                            closeInput(chunks);
+                            if (!future.isSuccess()) {
+                                closeInput(chunks);
+                                currentWrite.fail(future.cause());
+                            } else {
+                                currentWrite.progress(chunks.progress(), chunks.length());
+                                currentWrite.success(chunks.length());
+                            }
                         }
                     });
                 } else if (channel.isWritable()) {


Test patch:
diff --git a/handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java b/handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java
index 66b69516fc0..5b03048ba69 100644
--- a/handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java
+++ b/handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java
@@ -21,8 +21,11 @@
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.util.CharsetUtil;
+import io.netty.util.ReferenceCountUtil;
 import org.junit.Test;
 
 import java.io.ByteArrayInputStream;
@@ -31,10 +34,9 @@
 import java.io.IOException;
 import java.nio.channels.Channels;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 public class ChunkedWriteHandlerTest {
     private static final byte[] BYTES = new byte[1024 * 64];
@@ -162,8 +164,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
 
         EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
         ch.writeAndFlush(input).addListener(listener).syncUninterruptibly();
-        ch.checkException();
-        ch.finish();
+        assertTrue(ch.finish());
 
         // the listener should have been notified
         assertTrue(listenerNotified.get());
@@ -220,13 +221,218 @@ public long progress() {
 
         EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
         ch.writeAndFlush(input).syncUninterruptibly();
-        ch.checkException();
         assertTrue(ch.finish());
 
         assertEquals(0, ch.readOutbound());
         assertNull(ch.readOutbound());
     }
 
+    @Test
+    public void testWriteFailureChunkedStream() throws IOException {
+        checkFirstFailed(new ChunkedStream(new ByteArrayInputStream(BYTES)));
+    }
+
+    @Test
+    public void testWriteFailureChunkedNioStream() throws IOException {
+        checkFirstFailed(new ChunkedNioStream(Channels.newChannel(new ByteArrayInputStream(BYTES))));
+    }
+
+    @Test
+    public void testWriteFailureChunkedFile() throws IOException {
+        checkFirstFailed(new ChunkedFile(TMP));
+    }
+
+    @Test
+    public void testWriteFailureChunkedNioFile() throws IOException {
+        checkFirstFailed(new ChunkedNioFile(TMP));
+    }
+
+    @Test
+    public void testWriteFailureUnchunkedData() throws IOException {
+        checkFirstFailed(Unpooled.wrappedBuffer(BYTES));
+    }
+
+    @Test
+    public void testSkipAfterFailedChunkedStream() throws IOException {
+        checkSkipFailed(new ChunkedStream(new ByteArrayInputStream(BYTES)),
+                        new ChunkedStream(new ByteArrayInputStream(BYTES)));
+    }
+
+    @Test
+    public void testSkipAfterFailedChunkedNioStream() throws IOException {
+        checkSkipFailed(new ChunkedNioStream(Channels.newChannel(new ByteArrayInputStream(BYTES))),
+                        new ChunkedNioStream(Channels.newChannel(new ByteArrayInputStream(BYTES))));
+    }
+
+    @Test
+    public void testSkipAfterFailedChunkedFile() throws IOException {
+        checkSkipFailed(new ChunkedFile(TMP), new ChunkedFile(TMP));
+    }
+
+    @Test
+    public void testSkipAfterFailedChunkedNioFile() throws IOException {
+        checkSkipFailed(new ChunkedNioFile(TMP), new ChunkedFile(TMP));
+    }
+
+    // See https://github.com/netty/netty/issues/8700.
+    @Test
+    public void testFailureWhenLastChunkFailed() throws IOException {
+        ChannelOutboundHandlerAdapter failLast = new ChannelOutboundHandlerAdapter() {
+            private int passedWrites;
+
+            @Override
+            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                if (++this.passedWrites < 4) {
+                    ctx.write(msg, promise);
+                } else {
+                    ReferenceCountUtil.release(msg);
+                    promise.tryFailure(new RuntimeException());
+                }
+            }
+        };
+
+        EmbeddedChannel ch = new EmbeddedChannel(failLast, new ChunkedWriteHandler());
+        ChannelFuture r = ch.writeAndFlush(new ChunkedFile(TMP, 1024 * 16)); // 4 chunks
+        assertTrue(ch.finish());
+
+        assertFalse(r.isSuccess());
+        assertTrue(r.cause() instanceof RuntimeException);
+
+        // 3 out of 4 chunks were already written
+        int read = 0;
+        for (;;) {
+            ByteBuf buffer = ch.readOutbound();
+            if (buffer == null) {
+                break;
+            }
+            read += buffer.readableBytes();
+            buffer.release();
+        }
+
+        assertEquals(1024 * 16 * 3, read);
+    }
+
+    @Test
+    public void testDiscardPendingWritesOnInactive() throws IOException {
+
+        final AtomicBoolean closeWasCalled = new AtomicBoolean(false);
+
+        ChunkedInput<ByteBuf> notifiableInput = new ChunkedInput<ByteBuf>() {
+            private boolean done;
+            private final ByteBuf buffer = Unpooled.copiedBuffer("Test", CharsetUtil.ISO_8859_1);
+
+            @Override
+            public boolean isEndOfInput() throws Exception {
+                return done;
+            }
+
+            @Override
+            public void close() throws Exception {
+                buffer.release();
+                closeWasCalled.set(true);
+            }
+
+            @Deprecated
+            @Override
+            public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
+                return readChunk(ctx.alloc());
+            }
+
+            @Override
+            public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
+                if (done) {
+                    return null;
+                }
+                done = true;
+                return buffer.retainedDuplicate();
+            }
+
+            @Override
+            public long length() {
+                return -1;
+            }
+
+            @Override
+            public long progress() {
+                return 1;
+            }
+        };
+
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        // Write 3 messages and close channel before flushing
+        ChannelFuture r1 = ch.write(new ChunkedFile(TMP));
+        ChannelFuture r2 = ch.write(new ChunkedNioFile(TMP));
+        ch.write(notifiableInput);
+
+        // Should be `false` as we do not expect any messages to be written
+        assertFalse(ch.finish());
+
+        assertFalse(r1.isSuccess());
+        assertFalse(r2.isSuccess());
+        assertTrue(closeWasCalled.get());
+    }
+
+    // See https://github.com/netty/netty/issues/8700.
+    @Test
+    public void testStopConsumingChunksWhenFailed() {
+        final ByteBuf buffer = Unpooled.copiedBuffer("Test", CharsetUtil.ISO_8859_1);
+        final AtomicInteger chunks = new AtomicInteger(0);
+
+        ChunkedInput<ByteBuf> nonClosableInput = new ChunkedInput<ByteBuf>() {
+            @Override
+            public boolean isEndOfInput() throws Exception {
+                return chunks.get() >= 5;
+            }
+
+            @Override
+            public void close() throws Exception {
+                // no-op
+            }
+
+            @Deprecated
+            @Override
+            public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
+                return readChunk(ctx.alloc());
+            }
+
+            @Override
+            public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
+                chunks.incrementAndGet();
+                return buffer.retainedDuplicate();
+            }
+
+            @Override
+            public long length() {
+                return -1;
+            }
+
+            @Override
+            public long progress() {
+                return 1;
+            }
+        };
+
+        ChannelOutboundHandlerAdapter noOpWrites = new ChannelOutboundHandlerAdapter() {
+            @Override
+            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                ReferenceCountUtil.release(msg);
+                promise.tryFailure(new RuntimeException());
+            }
+        };
+
+        EmbeddedChannel ch = new EmbeddedChannel(noOpWrites, new ChunkedWriteHandler());
+        ch.writeAndFlush(nonClosableInput).awaitUninterruptibly();
+        // Should be `false` as we do not expect any messages to be written
+        assertFalse(ch.finish());
+        buffer.release();
+
+        // We should expect only single chunked being read from the input.
+        // It's possible to get a race condition here between resolving a promise and
+        // allocating a new chunk, but should be fine when working with embedded channels.
+        assertEquals(1, chunks.get());
+    }
+
     private static void check(Object... inputs) {
         EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
 
@@ -255,4 +461,67 @@ private static void check(Object... inputs) {
 
         assertEquals(BYTES.length * inputs.length, read);
     }
+
+    private static void checkFirstFailed(Object input) {
+        ChannelOutboundHandlerAdapter noOpWrites = new ChannelOutboundHandlerAdapter() {
+            @Override
+            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                ReferenceCountUtil.release(msg);
+                promise.tryFailure(new RuntimeException());
+            }
+        };
+
+        EmbeddedChannel ch = new EmbeddedChannel(noOpWrites, new ChunkedWriteHandler());
+        ChannelFuture r = ch.writeAndFlush(input);
+
+        // Should be `false` as we do not expect any messages to be written
+        assertFalse(ch.finish());
+        assertTrue(r.cause() instanceof RuntimeException);
+    }
+
+    private static void checkSkipFailed(Object input1, Object input2) {
+        ChannelOutboundHandlerAdapter failFirst = new ChannelOutboundHandlerAdapter() {
+            private boolean alreadyFailed;
+
+            @Override
+            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                if (alreadyFailed) {
+                    ctx.write(msg, promise);
+                } else {
+                    this.alreadyFailed = true;
+                    ReferenceCountUtil.release(msg);
+                    promise.tryFailure(new RuntimeException());
+                }
+            }
+        };
+
+        EmbeddedChannel ch = new EmbeddedChannel(failFirst, new ChunkedWriteHandler());
+        ChannelFuture r1 = ch.write(input1);
+        ChannelFuture r2 = ch.writeAndFlush(input2).awaitUninterruptibly();
+        assertTrue(ch.finish());
+
+        assertTrue(r1.cause() instanceof RuntimeException);
+        assertTrue(r2.isSuccess());
+
+        // note, that after we've "skipped" the first write,
+        // we expect to see the second message, chunk by chunk
+        int i = 0;
+        int read = 0;
+        for (;;) {
+            ByteBuf buffer = ch.readOutbound();
+            if (buffer == null) {
+                break;
+            }
+            while (buffer.isReadable()) {
+                assertEquals(BYTES[i++], buffer.readByte());
+                read++;
+                if (i == BYTES.length) {
+                    i = 0;
+                }
+            }
+            buffer.release();
+        }
+
+        assertEquals(BYTES.length, read);
+    }
 }