@Override
    protected void bufferToBeReleased(ByteBuffer justReleased) {
        BufferReference ref = new BufferReference(justReleased); //used ro lookup only
        BufferMetadata metadata = buffersInFlight.remove(ref);
        if (metadata == null)
            //its impossible for the buffer to have already been GC'ed (because we have a hard ref to it
            //in the function arg) so this means either a double free or not our buffer.
            throw new IllegalArgumentException("returned buffer " + ref.hashCode + " was never allocated by this pool");
        if (metadata.sizeBytes != justReleased.capacity()) {
            //this is a bug
            throw new IllegalStateException("buffer " + ref.hashCode + " has capacity " + justReleased.capacity() + " but recorded as " + metadata.sizeBytes);
        }
        log.trace("released buffer of size {} and identity {}", metadata.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