Repository: netty/netty
Issue: ChunkedWriteHandler does not close successful ChunkedInputs

### Expected behavior
`ChunkedWriteHandler` invokes `ChunkedInput#close()` after the input is fully consumed and `ChunkedInput#isEndOfInput()` returns `true.`

### Actual behavior
`ChunkedInput#close()` is not invoked when the input completes successfully. This can result in resource leaks when the input is based on an external resource, like a file handle.

### Steps to reproduce

Run the attached test.

### Minimal yet complete reproducer code (or URL to code)

```java
@Test
public void testCloseSuccessfulChunkedInput() throws Exception {
    final int totalChunks = 10;
    final AtomicBoolean closeInvoked = new AtomicBoolean();

    ChunkedInput<ByteBuf> input = new ChunkedInput<ByteBuf>() {
        int chunksProduced = 0;

        @Override
        public boolean isEndOfInput() throws Exception {
            return chunksProduced >= totalChunks;
        }

        @Override
        public void close() throws Exception {
            closeInvoked.set(true);
        }

        @Override
        public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
            return readChunk(ctx.alloc());
        }

        @Override
        public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
            ByteBuf buf = allocator.buffer(4);
            buf.writeInt(chunksProduced);
            chunksProduced++;
            return buf;
        }

        @Override
        public long length() {
            return totalChunks;
        }

        @Override
        public long progress() {
            return chunksProduced;
        }
    };

    EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());

    ch.writeAndFlush(input).sync();

    for (int i = 0; i < totalChunks; i++) {
        ByteBuf buf = ch.readOutbound();
        assertEquals(i, buf.readInt());
    }

    assertTrue(closeInvoked.get()); // this assertion fails
}
```

### Netty version

Problem exists only in `4.1.33.Final`.

### JVM version (e.g. `java -version`)

```
java version "1.8.0_201"
Java(TM) SE Runtime Environment (build 1.8.0_201-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.201-b09, mixed mode)
```

### OS version (e.g. `uname -a`)

```
Darwin MacBook-Pro.local 18.2.0 Darwin Kernel Version 18.2.0: Thu Dec 20 20:46:53 PST 2018; root:xnu-4903.241.1~1/RELEASE_X86_64 x86_64
```

This issue can be fixed by closing the input in a write future listener [here](https://github.com/netty/netty/blob/netty-4.1.33.Final/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L286-L287) after the input is consumed. I have a branch with a tentative fix https://github.com/lutovich/netty/commit/1534f7778ad8c14d488fe17916c5baed5aaedc60 and can create a PR if it looks okay.

Issue comments:
@lutovich fix looks good. Please submit a PR :)
@normanmaurer done, thank you.

Pull request review comments:
please also call `channel.finish()` and assert return value.
replace by `writeOutbound(...)` and assert return value
replace by `writeOutbound(...)` 
please also call `channel.finish()` and assert return value.
nit: final
private
private 
private 
final 
@lutovich Can we move this call out of the `if...else` block?
@lutovich seems like this was missed ?
I changed it back on purpose in the second commit to make sure listener only invokes `ChunkedInput#close()` on failure. On success it also needs to read progress and length from the input. Does this make sense?
ah ok yeah.
@lutovich We assume here that only `in.isEndOfInput` or `in.length` were able to fail, right? I would probably move this try/catch upper to be more "targeted" and specific. Otherwise, it's not obvious from the first glance why do we try to close the input once again here. 
@lutovich Not related to this specific change, but... I wonder if there's a chance that the write was already set to success/failure 🤔 Here's [a comment](https://github.com/netty/netty/blob/33c18469349d45b139506446f356e2b4f3af7086/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java#L217-L225) from `doFlush`, not sure if the same is applicable here.
@kachayev yeah, I think this try-catch is to handle failures in `#isEndOfInput()` and `#length()`. Added a commit with a tighter try-catch as you suggest. It makes code a bit longer and adds a `continue`. Not sure it's an improvement but please take a look :)
@kachayev I'm not sure, to be honest. `currentWrite` is a `PendingWrite` and uses `trySuccess/tryFailure` to complete the underlying promise. It shouldn't be a problem to invoke `success/failure` more than one, right?
We need Python's try/catch/**else** :) We can mark that first block failed in a separate variable, and test its value instead of continue.
Right, should not be a problem. I'm just trying to understand if closing the input (potentially?) twice will cause any troubles.
I think it should be ok... if not we can do another PR.

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 f39328dff7a..1a1822b5973 100644
--- a/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java
+++ b/handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java
@@ -166,22 +166,28 @@ private void discard(Throwable cause) {
             Object message = currentWrite.msg;
             if (message instanceof ChunkedInput) {
                 ChunkedInput<?> in = (ChunkedInput<?>) message;
+                boolean endOfInput;
+                long inputLength;
                 try {
-                    if (!in.isEndOfInput()) {
-                        if (cause == null) {
-                            cause = new ClosedChannelException();
-                        }
-                        currentWrite.fail(cause);
-                    } else {
-                        currentWrite.success(in.length());
-                    }
+                    endOfInput = in.isEndOfInput();
+                    inputLength = in.length();
                     closeInput(in);
                 } catch (Exception e) {
+                    closeInput(in);
                     currentWrite.fail(e);
                     if (logger.isWarnEnabled()) {
-                        logger.warn(ChunkedInput.class.getSimpleName() + ".isEndOfInput() failed", e);
+                        logger.warn(ChunkedInput.class.getSimpleName() + " failed", e);
                     }
-                    closeInput(in);
+                    continue;
+                }
+
+                if (!endOfInput) {
+                    if (cause == null) {
+                        cause = new ClosedChannelException();
+                    }
+                    currentWrite.fail(cause);
+                } else {
+                    currentWrite.success(inputLength);
                 }
             } else {
                 if (cause == null) {
@@ -249,8 +255,8 @@ private void doFlush(final ChannelHandlerContext ctx) {
                         ReferenceCountUtil.release(message);
                     }
 
-                    currentWrite.fail(t);
                     closeInput(chunks);
+                    currentWrite.fail(t);
                     break;
                 }
 
@@ -283,8 +289,12 @@ public void operationComplete(ChannelFuture future) throws Exception {
                                 closeInput(chunks);
                                 currentWrite.fail(future.cause());
                             } else {
-                                currentWrite.progress(chunks.progress(), chunks.length());
-                                currentWrite.success(chunks.length());
+                                // read state of the input in local variables before closing it
+                                long inputProgress = chunks.progress();
+                                long inputLength = chunks.length();
+                                closeInput(chunks);
+                                currentWrite.progress(inputProgress, inputLength);
+                                currentWrite.success(inputLength);
                             }
                         }
                     });
@@ -293,7 +303,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                         @Override
                         public void operationComplete(ChannelFuture future) throws Exception {
                             if (!future.isSuccess()) {
-                                closeInput((ChunkedInput<?>) pendingMessage);
+                                closeInput(chunks);
                                 currentWrite.fail(future.cause());
                             } else {
                                 currentWrite.progress(chunks.progress(), chunks.length());
@@ -305,7 +315,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                         @Override
                         public void operationComplete(ChannelFuture future) throws Exception {
                             if (!future.isSuccess()) {
-                                closeInput((ChunkedInput<?>) pendingMessage);
+                                closeInput(chunks);
                                 currentWrite.fail(future.cause());
                             } else {
                                 currentWrite.progress(chunks.progress(), chunks.length());


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 5b03048ba69..be6951d88bd 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,8 @@
 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.ChannelPromise;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
@@ -33,9 +33,11 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.nio.channels.Channels;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static java.util.concurrent.TimeUnit.*;
 import static org.junit.Assert.*;
 
 public class ChunkedWriteHandlerTest {
@@ -433,6 +435,142 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
         assertEquals(1, chunks.get());
     }
 
+    @Test
+    public void testCloseSuccessfulChunkedInput() {
+        int chunks = 10;
+        TestChunkedInput input = new TestChunkedInput(chunks);
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        assertTrue(ch.writeOutbound(input));
+
+        for (int i = 0; i < chunks; i++) {
+            ByteBuf buf = ch.readOutbound();
+            assertEquals(i, buf.readInt());
+            buf.release();
+        }
+
+        assertTrue(input.isClosed());
+        assertFalse(ch.finish());
+    }
+
+    @Test
+    public void testCloseFailedChunkedInput() {
+        Exception error = new Exception("Unable to produce a chunk");
+        ThrowingChunkedInput input = new ThrowingChunkedInput(error);
+
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        try {
+            ch.writeOutbound(input);
+            fail("Exception expected");
+        } catch (Exception e) {
+            assertEquals(error, e);
+        }
+
+        assertTrue(input.isClosed());
+        assertFalse(ch.finish());
+    }
+
+    @Test
+    public void testWriteListenerInvokedAfterSuccessfulChunkedInputClosed() throws Exception {
+        final TestChunkedInput input = new TestChunkedInput(2);
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        final AtomicBoolean inputClosedWhenListenerInvoked = new AtomicBoolean();
+        final CountDownLatch listenerInvoked = new CountDownLatch(1);
+
+        ChannelFuture writeFuture = ch.write(input);
+        writeFuture.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) {
+                inputClosedWhenListenerInvoked.set(input.isClosed());
+                listenerInvoked.countDown();
+            }
+        });
+        ch.flush();
+
+        assertTrue(listenerInvoked.await(10, SECONDS));
+        assertTrue(writeFuture.isSuccess());
+        assertTrue(inputClosedWhenListenerInvoked.get());
+        assertTrue(ch.finishAndReleaseAll());
+    }
+
+    @Test
+    public void testWriteListenerInvokedAfterFailedChunkedInputClosed() throws Exception {
+        final ThrowingChunkedInput input = new ThrowingChunkedInput(new RuntimeException());
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        final AtomicBoolean inputClosedWhenListenerInvoked = new AtomicBoolean();
+        final CountDownLatch listenerInvoked = new CountDownLatch(1);
+
+        ChannelFuture writeFuture = ch.write(input);
+        writeFuture.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) {
+                inputClosedWhenListenerInvoked.set(input.isClosed());
+                listenerInvoked.countDown();
+            }
+        });
+        ch.flush();
+
+        assertTrue(listenerInvoked.await(10, SECONDS));
+        assertFalse(writeFuture.isSuccess());
+        assertTrue(inputClosedWhenListenerInvoked.get());
+        assertFalse(ch.finish());
+    }
+
+    @Test
+    public void testWriteListenerInvokedAfterChannelClosedAndInputFullyConsumed() throws Exception {
+        // use empty input which has endOfInput = true
+        final TestChunkedInput input = new TestChunkedInput(0);
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        final AtomicBoolean inputClosedWhenListenerInvoked = new AtomicBoolean();
+        final CountDownLatch listenerInvoked = new CountDownLatch(1);
+
+        ChannelFuture writeFuture = ch.write(input);
+        writeFuture.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) {
+                inputClosedWhenListenerInvoked.set(input.isClosed());
+                listenerInvoked.countDown();
+            }
+        });
+        ch.close(); // close channel to make handler discard the input on subsequent flush
+        ch.flush();
+
+        assertTrue(listenerInvoked.await(10, SECONDS));
+        assertTrue(writeFuture.isSuccess());
+        assertTrue(inputClosedWhenListenerInvoked.get());
+        assertFalse(ch.finish());
+    }
+
+    @Test
+    public void testWriteListenerInvokedAfterChannelClosedAndInputNotFullyConsumed() throws Exception {
+        // use non-empty input which has endOfInput = false
+        final TestChunkedInput input = new TestChunkedInput(42);
+        EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
+
+        final AtomicBoolean inputClosedWhenListenerInvoked = new AtomicBoolean();
+        final CountDownLatch listenerInvoked = new CountDownLatch(1);
+
+        ChannelFuture writeFuture = ch.write(input);
+        writeFuture.addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) {
+                inputClosedWhenListenerInvoked.set(input.isClosed());
+                listenerInvoked.countDown();
+            }
+        });
+        ch.close(); // close channel to make handler discard the input on subsequent flush
+        ch.flush();
+
+        assertTrue(listenerInvoked.await(10, SECONDS));
+        assertFalse(writeFuture.isSuccess());
+        assertTrue(inputClosedWhenListenerInvoked.get());
+        assertFalse(ch.finish());
+    }
+
     private static void check(Object... inputs) {
         EmbeddedChannel ch = new EmbeddedChannel(new ChunkedWriteHandler());
 
@@ -524,4 +662,96 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
 
         assertEquals(BYTES.length, read);
     }
+
+    private static final class TestChunkedInput implements ChunkedInput<ByteBuf> {
+        private final int chunksToProduce;
+
+        private int chunksProduced;
+        private volatile boolean closed;
+
+        TestChunkedInput(int chunksToProduce) {
+            this.chunksToProduce = chunksToProduce;
+        }
+
+        @Override
+        public boolean isEndOfInput() {
+            return chunksProduced >= chunksToProduce;
+        }
+
+        @Override
+        public void close() {
+            closed = true;
+        }
+
+        @Override
+        public ByteBuf readChunk(ChannelHandlerContext ctx) {
+            return readChunk(ctx.alloc());
+        }
+
+        @Override
+        public ByteBuf readChunk(ByteBufAllocator allocator) {
+            ByteBuf buf = allocator.buffer();
+            buf.writeInt(chunksProduced);
+            chunksProduced++;
+            return buf;
+        }
+
+        @Override
+        public long length() {
+            return chunksToProduce;
+        }
+
+        @Override
+        public long progress() {
+            return chunksProduced;
+        }
+
+        boolean isClosed() {
+            return closed;
+        }
+    }
+
+    private static final class ThrowingChunkedInput implements ChunkedInput<ByteBuf> {
+        private final Exception error;
+
+        private volatile boolean closed;
+
+        ThrowingChunkedInput(Exception error) {
+            this.error = error;
+        }
+
+        @Override
+        public boolean isEndOfInput() {
+            return false;
+        }
+
+        @Override
+        public void close() {
+            closed = true;
+        }
+
+        @Override
+        public ByteBuf readChunk(ChannelHandlerContext ctx) throws Exception {
+            return readChunk(ctx.alloc());
+        }
+
+        @Override
+        public ByteBuf readChunk(ByteBufAllocator allocator) throws Exception {
+            throw error;
+        }
+
+        @Override
+        public long length() {
+            return -1;
+        }
+
+        @Override
+        public long progress() {
+            return -1;
+        }
+
+        boolean isClosed() {
+            return closed;
+        }
+    }
 }