@Override
        public void run() {
            while (alive) {
                try {
                    BufferReference ref = (BufferReference) garbageCollectedBuffers.remove(); //blocks
                    ref.clear();
                    //this cannot race with a release() call because an object is either reachable or not,
                    //release() can only happen before its GC'ed, and enqueue can only happen after.
                    //if the ref was enqueued it must then not have been released
                    BufferMetadata metadata = buffersInFlight.remove(ref);

                    if (metadata == null) {
                        //it can happen rarely that the buffer was release()ed properly (so no metadata) and yet
                        //the reference object to it remains reachable for a short period of time after release()
                        //and hence gets enqueued. this is because we keep refs in a ConcurrentHashMap which cleans
                        //up keys lazily.
                        continue;
                    }

                    availableMemory.addAndGet(metadata.sizeBytes);
                    log.error("Reclaimed buffer of size {} and identity {} that was not properly release()ed. This is a bug.", metadata.sizeBytes, ref.hashCode);
                } catch (InterruptedException e) {
                    log.debug("interrupted", e);
                    //ignore, we're a daemon thread
                }
            }
            log.info("GC listener shutting down");
        }
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