@Test
    public void testCloseNotifyWaiters() throws Exception {
        final int numWorkers = 2;

        BufferPool pool = new BufferPool(1, 1, metrics, Time.SYSTEM, metricGroup);
        ByteBuffer buffer = pool.allocate(1, Long.MAX_VALUE);

        ExecutorService executor = Executors.newFixedThreadPool(numWorkers);
        Callable<Void> work = () -> {
            assertThrows(KafkaException.class, () -> pool.allocate(1, Long.MAX_VALUE));
            return null;
        };
        for (int i = 0; i < numWorkers; ++i) {
            executor.submit(work);
        }

        TestUtils.waitForCondition(() -> pool.queued() == numWorkers, "Awaiting " + numWorkers + " workers to be blocked on allocation");

        // Close the buffer pool. This should notify all waiters.
        pool.close();

        TestUtils.waitForCondition(() -> pool.queued() == 0, "Awaiting " + numWorkers + " workers to be interrupted from allocation");

        pool.deallocate(buffer);
    }
This test creates lots of threads that hammer on the pool
clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java