Ensure PooledByteBuf.initUnpooled(...) correctly set the allocator

Motivation:

Commit 8dda984afe introduced a regression which lead to the situation that the allocator is not set when PooledByteBuf.initUnpooled(...) is called. Thus it was possible that PooledByteBuf.alloc() returns null or the wrong allocator if multiple PooledByteBufAllocator are used in an application.

Modifications:

- Correctly set the allocator
- Add test-case

Result:

Fixes [#6436].

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

@@ -43,30 +43,26 @@ abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
     }
 
     void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
+        init0(chunk, handle, offset, length, maxLength, cache);
+    }
+
+    void initUnpooled(PoolChunk<T> chunk, int length) {
+        init0(chunk, 0, chunk.offset, length, length, null);
+    }
+
+    private void init0(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
         assert handle >= 0;
         assert chunk != null;
 
         this.chunk = chunk;
-        this.handle = handle;
         memory = chunk.memory;
         allocator = chunk.arena.parent;
+        this.cache = cache;
+        this.handle = handle;
         this.offset = offset;
         this.length = length;
         this.maxLength = maxLength;
         tmpNioBuf = null;
-        this.cache = cache;
-    }
-
-    void initUnpooled(PoolChunk<T> chunk, int length) {
-        assert chunk != null;
-
-        this.chunk = chunk;
-        handle = 0;
-        memory = chunk.memory;
-        offset = chunk.offset;
-        this.length = maxLength = length;
-        tmpNioBuf = null;
-        cache = null;
     }
 
     /**

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

@@ -35,6 +35,7 @@ import java.util.concurrent.locks.LockSupport;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 public class PooledByteBufAllocatorTest extends AbstractByteBufAllocatorTest {
@@ -159,6 +160,26 @@ public class PooledByteBufAllocatorTest extends AbstractByteBufAllocatorTest {
         }
     }
 
+    @Test
+    public void testAllocNotNull() {
+        PooledByteBufAllocator allocator = new PooledByteBufAllocator(true, 1, 1, 8192, 11, 0, 0, 0);
+        // Huge allocation
+        testAllocNotNull(allocator, allocator.chunkSize() + 1);
+        // Normal allocation
+        testAllocNotNull(allocator, 1024);
+        // Small allocation
+        testAllocNotNull(allocator, 512);
+        // Tiny allocation
+        testAllocNotNull(allocator, 1);
+    }
+
+    private static void testAllocNotNull(PooledByteBufAllocator allocator, int capacity) {
+        ByteBuf buffer = allocator.heapBuffer(capacity);
+        assertNotNull(buffer.alloc());
+        assertTrue(buffer.release());
+        assertNotNull(buffer.alloc());
+    }
+
     @Test
     public void testFreePoolChunk() {
         int chunkSize = 16 * 1024 * 1024;