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

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

        Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
        sender.runOnce();  // connect.
        sender.runOnce();  // send.
        String id = client.requests().peek().destination();
        Node node = new Node(Integer.parseInt(id), "localhost", 0);
        assertEquals(1, client.inFlightRequestCount());
        assertTrue(client.isReady(node, time.milliseconds()), "Client ready status should be true");
        client.disconnect(id);
        assertEquals(0, client.inFlightRequestCount());
        assertFalse(client.isReady(node, time.milliseconds()), "Client ready status should be false");
        sender.runOnce(); // receive error
        sender.runOnce(); // reset producer ID because epoch is maxed out

        prepareAndReceiveInitProducerId(producerId + 1, Errors.NONE);
        sender.runOnce(); // nothing to do, since the pid has changed. We should check the metrics for errors.
        assertEquals(1, client.inFlightRequestCount(), "Expected requests to be retried after pid change");

        assertFalse(responseFuture.isDone());
        assertEquals(1, (long) transactionManager.sequenceNumber(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