@Test
    public void testCleanupMemoryAvailabilityOnMetricsException() throws Exception {
        BufferPool bufferPool = spy(new BufferPool(2, 1, new Metrics(), time, metricGroup));
        doThrow(new OutOfMemoryError()).when(bufferPool).recordWaitTime(anyLong());

        bufferPool.allocate(1, 0);
        try {
            bufferPool.allocate(2, 1000);
            fail("Expected oom.");
        } catch (OutOfMemoryError expected) {
        }
        assertEquals(1, bufferPool.availableMemory());
        assertEquals(0, bufferPool.queued());
        assertEquals(1, bufferPool.unallocatedMemory());
        //This shouldn't timeout
        bufferPool.allocate(1, 0);

        verify(bufferPool).recordWaitTime(anyLong());
    }
Test if the  waiter that is waiting on availability of more memory is cleaned up when an interruption occurs
clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java