@Test
    public void testNodeNotReady() {
        final long producerId = 123456L;
        time = new MockTime(10);
        client = new MockClient(time, metadata);

        TransactionManager transactionManager = new TransactionManager(new LogContext(), "testNodeNotReady",
                60000, 100L, new ApiVersions());
        setupWithTransactionState(transactionManager, false, null, true);
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);
        transactionManager.initializeTransactions();
        sender.runOnce();

        Node node = metadata.fetch().nodes().get(0);
        client.delayReady(node, REQUEST_TIMEOUT + 20);
        prepareFindCoordinatorResponse(Errors.NONE, "testNodeNotReady");
        sender.runOnce();
        sender.runOnce();
        assertNotNull(transactionManager.coordinator(CoordinatorType.TRANSACTION), "Coordinator not found");

        client.throttle(node, REQUEST_TIMEOUT + 20);
        prepareFindCoordinatorResponse(Errors.NONE, "Coordinator not found");
        prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
        waitForProducerId(transactionManager, producerIdAndEpoch);
    }
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