Repository: netty/netty
Issue: ByteToMessageDecoder doesn't release cumulation buffer

ByteToMessageDecoder doesn't correctly release internal "cumulation" ByteBuf in case decoder is removed from pipeline during decode operation and all bytes of the buffer were already read (i.e. buf.isReadable() == false).

Buffer is typically released in method handlerRemoved, but only in case it still has content to be read. In case buffer was fully read, it causes leaks detectable by ResourceLeakDetector (LEAK: ByteBuf.release() was not called before it's garbage-collected). Detection can be speeded up with setting -Dio.netty.leakDetectionLevel=PARANOID.

The pull request which follows contains two test cases which demonstrate the problem. Test ByteToMessageDecoderTest#testInternalBufferClearReadPart correctly clears the buffer, while ByteToMessageDecoderTest#testInternalBufferClearReadAll fails. The issue is fixed by a change in handlerRemoved method of ByteToMessageDecoder.

Issue was reproduced in the latest 4.0.16-snapshot


Issue comments:
@vschafer good catch! Thanks for the tests and fix!


Pull request review comments:

Patch that resolved the issue:
diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
index 50dc125886a..637f4cd9d02 100644
--- a/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
@@ -109,6 +109,8 @@ public final void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
             ByteBuf bytes = buf.readBytes(readable);
             buf.release();
             ctx.fireChannelRead(bytes);
+        } else {
+            buf.release();
         }
         cumulation = null;
         ctx.fireChannelReadComplete();


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
index 09d5020073a..082ac8a2fca 100644
--- a/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/ByteToMessageDecoderTest.java
@@ -72,4 +72,50 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) t
         buf.release();
         b.release();
     }
+
+    /**
+     * Verifies that internal buffer of the ByteToMessageDecoder is released once decoder is removed from pipeline. In
+     * this case input is read fully.
+     */
+    @Test
+    public void testInternalBufferClearReadAll() {
+
+        final ByteBuf buf = Unpooled.buffer().writeBytes(new byte[]{'a'});
+        EmbeddedChannel channel = new EmbeddedChannel(new ByteToMessageDecoder() {
+            @Override
+            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+                ByteBuf byteBuf = internalBuffer();
+                Assert.assertEquals(1, byteBuf.refCnt());
+                in.readByte();
+                // Removal from pipeline should clear internal buffer
+                ctx.pipeline().remove(this);
+                Assert.assertEquals(0, byteBuf.refCnt());
+            }
+        });
+        channel.writeInbound(buf.copy());
+
+    }
+
+    /**
+     * Verifies that internal buffer of the ByteToMessageDecoder is released once decoder is removed from pipeline. In
+     * this case input was not fully read.
+     */
+    @Test
+    public void testInternalBufferClearReadPartly() {
+
+        final ByteBuf buf = Unpooled.buffer().writeBytes(new byte[]{'a', 'b'});
+        EmbeddedChannel channel = new EmbeddedChannel(new ByteToMessageDecoder() {
+            @Override
+            protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+                ByteBuf byteBuf = internalBuffer();
+                Assert.assertEquals(1, byteBuf.refCnt());
+                in.readByte();
+                // Removal from pipeline should clear internal buffer
+                ctx.pipeline().remove(this);
+                Assert.assertEquals(0, byteBuf.refCnt());
+            }
+        });
+        channel.writeInbound(buf.copy());
+
+    }
 }