[#3789] Correctly reset markers for all allocations when using PooledByteBufAllocator

Motivation:

We need to ensure all markers are reset when doing an allocation via the PooledByteBufAllocator. This was not the always the case.

Modifications:

Move all logic that needs to get executed when reuse a PooledByteBuf into one place and call it.

Result:

Correct behavior

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

@@ -101,8 +101,7 @@ public abstract class AbstractByteBuf extends ByteBuf {
                     "readerIndex: %d, writerIndex: %d (expected: 0 <= readerIndex <= writerIndex <= capacity(%d))",
                     readerIndex, writerIndex, capacity()));
         }
-        this.readerIndex = readerIndex;
-        this.writerIndex = writerIndex;
+        setIndex0(readerIndex, writerIndex);
         return this;
     }
 
@@ -1152,7 +1151,12 @@ public abstract class AbstractByteBuf extends ByteBuf {
         }
     }
 
-    void discardMarks() {
+    final void setIndex0(int readerIndex, int writerIndex) {
+        this.readerIndex = readerIndex;
+        this.writerIndex = writerIndex;
+    }
+
+    final void discardMarks() {
         markedReaderIndex = markedWriterIndex = 0;
     }
 }

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

@@ -51,8 +51,6 @@ abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
         this.offset = offset;
         this.length = length;
         this.maxLength = maxLength;
-        setIndex(0, 0);
-        discardMarks();
         tmpNioBuf = null;
         this.cache = cache;
     }
@@ -65,11 +63,20 @@ abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
         memory = chunk.memory;
         offset = 0;
         this.length = maxLength = length;
-        setIndex(0, 0);
         tmpNioBuf = null;
         cache = null;
     }
 
+    /**
+     * Method must be called before reuse this {@link PooledByteBufAllocator}
+     */
+    final void reuse(int maxCapacity) {
+        maxCapacity(maxCapacity);
+        setRefCnt(1);
+        setIndex0(0, 0);
+        discardMarks();
+    }
+
     @Override
     public final int capacity() {
         return length;

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

@@ -37,8 +37,7 @@ final class PooledDirectByteBuf extends PooledByteBuf<ByteBuffer> {
 
     static PooledDirectByteBuf newInstance(int maxCapacity) {
         PooledDirectByteBuf buf = RECYCLER.get();
-        buf.setRefCnt(1);
-        buf.maxCapacity(maxCapacity);
+        buf.reuse(maxCapacity);
         return buf;
     }
 

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

@@ -36,8 +36,7 @@ final class PooledHeapByteBuf extends PooledByteBuf<byte[]> {
 
     static PooledHeapByteBuf newInstance(int maxCapacity) {
         PooledHeapByteBuf buf = RECYCLER.get();
-        buf.setRefCnt(1);
-        buf.maxCapacity(maxCapacity);
+        buf.reuse(maxCapacity);
         return buf;
     }
 

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

@@ -41,8 +41,7 @@ final class PooledUnsafeDirectByteBuf extends PooledByteBuf<ByteBuffer> {
 
     static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) {
         PooledUnsafeDirectByteBuf buf = RECYCLER.get();
-        buf.setRefCnt(1);
-        buf.maxCapacity(maxCapacity);
+        buf.reuse(maxCapacity);
         return buf;
     }
 

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

@@ -40,7 +40,16 @@ public abstract class AbstractPooledByteBufTest extends AbstractByteBufTest {
 
     @Test
     public void testDiscardMarks() {
-        ByteBuf buf = newBuffer(4);
+        testDiscardMarks(4);
+    }
+
+    @Test
+    public void testDiscardMarksUnpooled() {
+        testDiscardMarks(32 * 1024 * 1024);
+    }
+
+    private void testDiscardMarks(int capacity) {
+        ByteBuf buf = newBuffer(capacity);
         buf.writeShort(1);
 
         buf.skipBytes(1);
@@ -49,7 +58,7 @@ public abstract class AbstractPooledByteBufTest extends AbstractByteBufTest {
         buf.markWriterIndex();
         assertTrue(buf.release());
 
-        ByteBuf buf2 = newBuffer(4);
+        ByteBuf buf2 = newBuffer(capacity);
 
         assertSame(unwrapIfNeeded(buf), unwrapIfNeeded(buf2));