@Test
    public void testIncompleteTransactionAbortOnShutdown() {
        // create a sender with retries = 1
        int maxRetries = 1;
        Metrics m = new Metrics();
        SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
        try {
            TransactionManager txnManager = new TransactionManager(logContext, "testIncompleteTransactionAbortOnShutdown", 6000, 100, apiVersions);
            Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
                    maxRetries, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, txnManager, apiVersions);

            ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
            TopicPartition tp = new TopicPartition("testIncompleteTransactionAbortOnShutdown", 1);

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

            txnManager.beginTransaction();
            txnManager.maybeAddPartition(tp);
            client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp, Errors.NONE)));
            sender.runOnce();
            sender.initiateClose();
            AssertEndTxnRequestMatcher endTxnMatcher = new AssertEndTxnRequestMatcher(TransactionResult.ABORT);
            client.prepareResponse(endTxnMatcher, new EndTxnResponse(new EndTxnResponseData()
                                                                         .setErrorCode(Errors.NONE.code())
                                                                         .setThrottleTimeMs(0)));
            sender.run();
            assertTrue(endTxnMatcher.matched, "Response didn't match in test");
        } finally {
            m.close();
        }
    }
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