@Test
    public void testReceiveFailedBatchTwiceWithTransactions() throws Exception {
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
        apiVersions.update("0", NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3));
        TransactionManager txnManager = new TransactionManager(logContext, "testFailTwice", 60000, 100, apiVersions);

        setupWithTransactionState(txnManager);
        doInitTransactions(txnManager, producerIdAndEpoch);

        txnManager.beginTransaction();
        txnManager.maybeAddPartition(tp0);
        client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp0, Errors.NONE)));
        sender.runOnce();

        // Send first ProduceRequest
        Future<RecordMetadata> request1 = appendToAccumulator(tp0);
        sender.runOnce();  // send request

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

        sender.runOnce(); // now expire the batch.
        assertFutureFailure(request1, TimeoutException.class);

        time.sleep(20);

        sendIdempotentProducerResponse(0, tp0, Errors.INVALID_TXN_STATE, -1);
        sender.runOnce(); // receive late response

        // Loop once and confirm that the transaction manager does not enter a fatal error state
        sender.runOnce();
        assertTrue(txnManager.hasAbortableError());
        TransactionalRequestResult result = txnManager.beginAbort();
        sender.runOnce();

        respondToEndTxn(Errors.NONE);
        sender.runOnce();
        assertTrue(txnManager::isInitializing);
        prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
        sender.runOnce();
        assertTrue(txnManager::isReady);

        assertTrue(result.isSuccessful());
        result.await();

        txnManager.beginTransaction();
    }
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