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

        Metrics m = new Metrics();
        SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

        Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, 10,
            senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

        appendToAccumulator(tp0); // failed response
        appendToAccumulator(tp1); // success response
        sender.runOnce();  // connect and send.

        assertEquals(1, client.inFlightRequestCount());

        Map<TopicPartition, OffsetAndError> responses = new LinkedHashMap<>();
        responses.put(tp1, new OffsetAndError(-1, Errors.NOT_LEADER_OR_FOLLOWER));
        responses.put(tp0, new OffsetAndError(-1, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER));
        client.respond(produceResponse(responses));
        sender.initiateClose(); // initiate close
        sender.runOnce(); // out of order sequence error triggers producer ID reset because epoch is maxed out

        TestUtils.waitForCondition(() -> {
            prepareInitProducerResponse(Errors.NONE, producerId + 1, (short) 1);
            sender.runOnce();
            return !accumulator.hasUndrained();
        }, 5000, "Failed to drain batches");
    }
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