@Test
    public void testExpiryOfAllSentBatchesShouldCauseUnresolvedSequences() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        setupWithTransactionState(transactionManager);
        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        assertTrue(transactionManager.hasProducerId());

        assertEquals(0, transactionManager.sequenceNumber(tp0));

        // Send first ProduceRequest
        Future<RecordMetadata> request1 = appendToAccumulator(tp0, 0L, "key", "value");
        sender.runOnce();  // send request
        sendIdempotentProducerResponse(0, tp0, Errors.NOT_LEADER_OR_FOLLOWER, -1);

        sender.runOnce();  // receive response
        assertEquals(1L, transactionManager.sequenceNumber(tp0));

        Node node = metadata.fetch().nodes().get(0);
        time.sleep(15000L);
        client.disconnect(node.idString());
        client.backoff(node, 10);

        sender.runOnce(); // now expire the batch.

        assertFutureFailure(request1, TimeoutException.class);
        assertTrue(transactionManager.hasUnresolvedSequence(tp0));
        assertFalse(client.hasInFlightRequests());
        Deque<ProducerBatch> batches = accumulator.getDeque(tp0);
        assertEquals(0, batches.size());
        assertEquals(producerId, transactionManager.producerIdAndEpoch().producerId);

        // In the next run loop, we bump the epoch and clear the unresolved sequences
        sender.runOnce();
        assertEquals(1, transactionManager.producerIdAndEpoch().epoch);
        assertFalse(transactionManager.hasUnresolvedSequence(tp0));
    }
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