Retained[Duplicate|Slice] operations should not increase the reference count for UnreleasableByteBuf

Motivation:
UnreleasableByteBuf operations are designed to not modify the reference count of the underlying buffer. The Retained[Duplicate|Slice] operations violate this assumption and can cause the underlying buffer's reference count to be increased, but never allow for it to be decreased. This may lead to memory leaks.

Modifications:
- UnreleasableByteBuf's Retained[Duplicate|Slice] should leave the reference count of the parent buffer unchanged after the operation completes.

Result:
No more memory leaks due to usage of the Retained[Duplicate|Slice] on an UnreleasableByteBuf object.

buffer/src/main/java/io/netty/buffer/UnreleasableByteBuf.java

@@ -57,7 +57,10 @@ final class UnreleasableByteBuf extends WrappedByteBuf {
 
     @Override
     public ByteBuf readRetainedSlice(int length) {
-        return new UnreleasableByteBuf(buf.readRetainedSlice(length));
+        // We could call buf.readSlice(..), and then call buf.release(). However this creates a leak in unit tests
+        // because the release method on UnreleasableByteBuf will never allow the leak record to be cleaned up.
+        // So we just use readSlice(..) because the end result should be logically equivalent.
+        return readSlice(length);
     }
 
     @Override
@@ -67,7 +70,10 @@ final class UnreleasableByteBuf extends WrappedByteBuf {
 
     @Override
     public ByteBuf retainedSlice() {
-        return new UnreleasableByteBuf(buf.retainedSlice());
+        // We could call buf.retainedSlice(), and then call buf.release(). However this creates a leak in unit tests
+        // because the release method on UnreleasableByteBuf will never allow the leak record to be cleaned up.
+        // So we just use slice() because the end result should be logically equivalent.
+        return slice();
     }
 
     @Override
@@ -77,7 +83,10 @@ final class UnreleasableByteBuf extends WrappedByteBuf {
 
     @Override
     public ByteBuf retainedSlice(int index, int length) {
-        return new UnreleasableByteBuf(buf.retainedSlice(index, length));
+        // We could call buf.retainedSlice(..), and then call buf.release(). However this creates a leak in unit tests
+        // because the release method on UnreleasableByteBuf will never allow the leak record to be cleaned up.
+        // So we just use slice(..) because the end result should be logically equivalent.
+        return slice(index, length);
     }
 
     @Override
@@ -87,7 +96,10 @@ final class UnreleasableByteBuf extends WrappedByteBuf {
 
     @Override
     public ByteBuf retainedDuplicate() {
-        return new UnreleasableByteBuf(buf.retainedDuplicate());
+        // We could call buf.retainedDuplicate(), and then call buf.release(). However this creates a leak in unit tests
+        // because the release method on UnreleasableByteBuf will never allow the leak record to be cleaned up.
+        // So we just use duplicate() because the end result should be logically equivalent.
+        return duplicate();
     }
 
     @Override

buffer/src/test/java/io/netty/buffer/AbstractByteBufTest.java

@@ -50,6 +50,7 @@ import static io.netty.buffer.Unpooled.LITTLE_ENDIAN;
 import static io.netty.buffer.Unpooled.buffer;
 import static io.netty.buffer.Unpooled.copiedBuffer;
 import static io.netty.buffer.Unpooled.directBuffer;
+import static io.netty.buffer.Unpooled.unreleasableBuffer;
 import static io.netty.buffer.Unpooled.wrappedBuffer;
 import static io.netty.util.internal.EmptyArrays.EMPTY_BYTES;
 import static org.hamcrest.CoreMatchers.is;
@@ -3460,6 +3461,106 @@ public abstract class AbstractByteBufTest {
         testSliceCapacityChange(true);
     }
 
+    @Test
+    public void testRetainedSliceUnreleasble1() {
+        testRetainedSliceUnreleasble(true, true);
+    }
+
+    @Test
+    public void testRetainedSliceUnreleasble2() {
+        testRetainedSliceUnreleasble(true, false);
+    }
+
+    @Test
+    public void testRetainedSliceUnreleasble3() {
+        testRetainedSliceUnreleasble(false, true);
+    }
+
+    @Test
+    public void testRetainedSliceUnreleasble4() {
+        testRetainedSliceUnreleasble(false, false);
+    }
+
+    @Test
+    public void testReadRetainedSliceUnreleasble1() {
+        testReadRetainedSliceUnreleasble(true, true);
+    }
+
+    @Test
+    public void testReadRetainedSliceUnreleasble2() {
+        testReadRetainedSliceUnreleasble(true, false);
+    }
+
+    @Test
+    public void testReadRetainedSliceUnreleasble3() {
+        testReadRetainedSliceUnreleasble(false, true);
+    }
+
+    @Test
+    public void testReadRetainedSliceUnreleasble4() {
+        testReadRetainedSliceUnreleasble(false, false);
+    }
+
+    @Test
+    public void testRetainedDuplicateUnreleasble1() {
+        testRetainedDuplicateUnreleasble(true, true);
+    }
+
+    @Test
+    public void testRetainedDuplicateUnreleasble2() {
+        testRetainedDuplicateUnreleasble(true, false);
+    }
+
+    @Test
+    public void testRetainedDuplicateUnreleasble3() {
+        testRetainedDuplicateUnreleasble(false, true);
+    }
+
+    @Test
+    public void testRetainedDuplicateUnreleasble4() {
+        testRetainedDuplicateUnreleasble(false, false);
+    }
+
+    private void testRetainedSliceUnreleasble(boolean initRetainedSlice, boolean finalRetainedSlice) {
+        ByteBuf buf = newBuffer(8);
+        ByteBuf buf1 = initRetainedSlice ? buf.retainedSlice() : buf.slice().retain();
+        ByteBuf buf2 = unreleasableBuffer(buf1);
+        ByteBuf buf3 = finalRetainedSlice ? buf2.retainedSlice() : buf2.slice().retain();
+        assertFalse(buf3.release());
+        assertFalse(buf2.release());
+        buf1.release();
+        assertTrue(buf.release());
+        assertEquals(0, buf1.refCnt());
+        assertEquals(0, buf.refCnt());
+    }
+
+    private void testReadRetainedSliceUnreleasble(boolean initRetainedSlice, boolean finalRetainedSlice) {
+        ByteBuf buf = newBuffer(8);
+        ByteBuf buf1 = initRetainedSlice ? buf.retainedSlice() : buf.slice().retain();
+        ByteBuf buf2 = unreleasableBuffer(buf1);
+        ByteBuf buf3 = finalRetainedSlice ? buf2.readRetainedSlice(buf2.readableBytes())
+                                          : buf2.readSlice(buf2.readableBytes()).retain();
+        assertFalse(buf3.release());
+        assertFalse(buf2.release());
+        buf1.release();
+        assertTrue(buf.release());
+        assertEquals(0, buf1.refCnt());
+        assertEquals(0, buf.refCnt());
+    }
+
+    private void testRetainedDuplicateUnreleasble(boolean initRetainedDuplicate, boolean finalRetainedDuplicate) {
+        ByteBuf buf = newBuffer(8);
+        ByteBuf buf1 = initRetainedDuplicate ? buf.retainedDuplicate() : buf.duplicate().retain();
+        ByteBuf buf2 = unreleasableBuffer(buf1);
+        ByteBuf buf3 = finalRetainedDuplicate ? buf2.retainedDuplicate() : buf2.duplicate().retain();
+        assertFalse(buf3.release());
+        assertFalse(buf2.release());
+        buf1.release();
+        assertTrue(buf.release());
+        assertEquals(0, buf1.refCnt());
+        assertEquals(0, buf.refCnt());
+    }
+
     private void testDuplicateCapacityChange(boolean retainedDuplicate) {
         ByteBuf buf = newBuffer(8);
         ByteBuf dup = retainedDuplicate ? buf.retainedDuplicate() : buf.duplicate();