@Test
    public void testSimple() throws Exception {
        long totalMemory = 64 * 1024;
        int size = 1024;
        BufferPool pool = new BufferPool(totalMemory, size, metrics, time, metricGroup);
        ByteBuffer buffer = pool.allocate(size, maxBlockTimeMs);
        assertEquals(size, buffer.limit(), "Buffer size should equal requested size.");
        assertEquals(totalMemory - size, pool.unallocatedMemory(), "Unallocated memory should have shrunk");
        assertEquals(totalMemory - size, pool.availableMemory(), "Available memory should have shrunk");
        buffer.putInt(1);
        buffer.flip();
        pool.deallocate(buffer);
        assertEquals(totalMemory, pool.availableMemory(), "All memory should be available");
        assertEquals(totalMemory - size, pool.unallocatedMemory(), "But now some is on the free list");
        buffer = pool.allocate(size, maxBlockTimeMs);
        assertEquals(0, buffer.position(), "Recycled buffer should be cleared.");
        assertEquals(buffer.capacity(), buffer.limit(), "Recycled buffer should be cleared.");
        pool.deallocate(buffer);
        assertEquals(totalMemory, pool.availableMemory(), "All memory should be available");
        assertEquals(totalMemory - size, pool.unallocatedMemory(), "Still a single buffer on the free list");
        buffer = pool.allocate(2 * size, maxBlockTimeMs);
        pool.deallocate(buffer);
        assertEquals(totalMemory, pool.availableMemory(), "All memory should be available");
        assertEquals(totalMemory - size, pool.unallocatedMemory(), "Non-standard size didn't go to the free list.");
    }
Test the simple non-blocking allocation paths
clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java