Change the contract of ResourceLeakDetector.open() so that unsampled resources are recycled

- This also fixes the problem introduced while trying to implement #1612 (Allow to disable resource leak detection).

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

@@ -1568,7 +1568,9 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf {
             c.freeIfNecessary();
         }
 
-        leak.close();
+        if (leak != null) {
+            leak.close();
+        }
     }
 
     @Override

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

@@ -18,7 +18,6 @@ package io.netty.buffer;
 
 import io.netty.util.Recycler;
 import io.netty.util.ResourceLeak;
-import io.netty.util.ResourceLeakDetector;
 
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
@@ -145,7 +144,7 @@ abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
             this.handle = -1;
             memory = null;
             chunk.arena.free(chunk, handle);
-            if (ResourceLeakDetector.isEnabled()) {
+            if (leak != null) {
                 leak.close();
             } else {
                 recycle();
@@ -157,7 +156,6 @@ abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
     private void recycle() {
         Recycler.Handle recyclerHandle = this.recyclerHandle;
         if (recyclerHandle != null) {
-            setRefCnt(1);
             ((Recycler<Object>) recycler()).recycle(this, recyclerHandle);
         }
     }

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

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

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

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

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

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

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

@@ -51,7 +51,9 @@ class ReadOnlyByteBufferBuf extends AbstractReferenceCountedByteBuf {
 
     @Override
     protected void deallocate() {
-        leak.close();
+        if (leak != null) {
+            leak.close();
+        }
     }
 
     @Override

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

@@ -518,7 +518,10 @@ public class UnpooledDirectByteBuf extends AbstractReferenceCountedByteBuf {
         if (!doNotFree) {
             PlatformDependent.freeDirectBuffer(buffer);
         }
-        leak.close();
+
+        if (leak != null) {
+            leak.close();
+        }
     }
 
     @Override

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

@@ -450,7 +450,10 @@ public class UnpooledUnsafeDirectByteBuf extends AbstractReferenceCountedByteBuf
         if (!doNotFree) {
             PlatformDependent.freeDirectBuffer(buffer);
         }
-        leak.close();
+
+        if (leak != null) {
+            leak.close();
+        }
     }
 
     @Override

common/src/main/java/io/netty/util/HashedWheelTimer.java

@@ -290,7 +290,9 @@ public class HashedWheelTimer implements Timer {
             Thread.currentThread().interrupt();
         }
 
-        leak.close();
+        if (leak != null) {
+            leak.close();
+        }
 
         Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
         for (Set<HashedWheelTimeout> bucket: wheel) {

common/src/main/java/io/netty/util/ResourceLeakDetector.java

@@ -40,13 +40,6 @@ public final class ResourceLeakDetector<T> {
 
     private static final int DEFAULT_SAMPLING_INTERVAL = 113;
 
-    private static final ResourceLeak NOOP = new ResourceLeak() {
-        @Override
-        public boolean close() {
-            return false;
-        }
-    };
-
     /**
      * Enables or disabled the resource leak detection.
      */
@@ -107,9 +100,15 @@ public final class ResourceLeakDetector<T> {
         tail.prev = head;
     }
 
+    /**
+     * Creates a new {@link ResourceLeak} which is expected to be closed via {@link ResourceLeak#close()} when the
+     * related resource is deallocated.
+     *
+     * @return the {@link ResourceLeak} or {@code null}
+     */
     public ResourceLeak open(T obj) {
         if (disabled || leakCheckCnt ++ % samplingInterval != 0) {
-            return NOOP;
+            return null;
         }
 
         reportLeak();