@Override
    protected void bufferToBeReturned(ByteBuffer justAllocated) {
        BufferReference ref = new BufferReference(justAllocated, garbageCollectedBuffers);
        BufferMetadata metadata = new BufferMetadata(justAllocated.capacity());
        if (buffersInFlight.put(ref, metadata) != null)
            //this is a bug. it means either 2 different co-existing buffers got
            //the same identity or we failed to register a released/GC'ed buffer
            throw new IllegalStateException("allocated buffer identity " + ref.hashCode + " already registered as in use?!");

        log.trace("allocated buffer of size {} and identity {}", sizeBytes, ref.hashCode);
    }
An extension of SimpleMemoryPool that tracks allocated buffers and logs an error when they "leak"
(when they are garbage-collected without having been release()ed).
THIS IMPLEMENTATION IS A DEVELOPMENT/DEBUGGING AID AND IS NOT MEANT PRO PRODUCTION USE.
clients/src/main/java/org/apache/kafka/common/memory/GarbageCollectedMemoryPool.java