@SuppressWarnings("deprecation")
    @Test
    public void testExpiredBatchesInMultiplePartitions() throws Exception {
        long deliveryTimeoutMs = 1500L;
        setupWithTransactionState(null, true, null);

        // Send multiple ProduceRequest across multiple partitions.
        Future<RecordMetadata> request1 = appendToAccumulator(tp0, time.milliseconds(), "k1", "v1");
        Future<RecordMetadata> request2 = appendToAccumulator(tp1, time.milliseconds(), "k2", "v2");

        // Send request.
        sender.runOnce();
        assertEquals(1, client.inFlightRequestCount());
        assertEquals(1, sender.inFlightBatches(tp0).size(), "Expect one in-flight batch in accumulator");

        Map<TopicPartition, ProduceResponse.PartitionResponse> responseMap = new HashMap<>();
        responseMap.put(tp0, new ProduceResponse.PartitionResponse(Errors.NONE, 0L, 0L, 0L));
        client.respond(new ProduceResponse(responseMap));

        // Successfully expire both batches.
        time.sleep(deliveryTimeoutMs);
        sender.runOnce();
        assertEquals(0, sender.inFlightBatches(tp0).size(), "Expect zero in-flight batch in accumulator");

        ExecutionException e = assertThrows(ExecutionException.class, request1::get);
        assertInstanceOf(TimeoutException.class, e.getCause());

        e = assertThrows(ExecutionException.class, request2::get);
        assertInstanceOf(TimeoutException.class, e.getCause());
    }
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