@Test
    public void testNoDoubleDeallocation() throws Exception {
        long totalSize = 1024 * 1024;
        String metricGrpName = "producer-custom-metrics";
        MatchingBufferPool pool = new MatchingBufferPool(totalSize, batchSize, metrics, time, metricGrpName);
        setupWithTransactionState(null, false, pool);

        // Send first ProduceRequest
        Future<RecordMetadata> request1 = appendToAccumulator(tp0);
        sender.runOnce();  // send request
        assertEquals(1, client.inFlightRequestCount());
        assertEquals(1, sender.inFlightBatches(tp0).size());

        time.sleep(REQUEST_TIMEOUT);
        assertFalse(pool.allMatch());

        sender.runOnce();  // expire the batch
        assertTrue(request1.isDone());
        assertTrue(pool.allMatch(), "The batch should have been de-allocated");
        assertTrue(pool.allMatch());

        sender.runOnce();
        assertTrue(pool.allMatch(), "The batch should have been de-allocated");
        assertEquals(0, client.inFlightRequestCount());
        assertEquals(0, sender.inFlightBatches(tp0).size());
    }
Tests the code path where the target node to send FindCoordinator or InitProducerId
is not ready.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java