@Test
    public void testTransactionAbortedExceptionOnAbortWithoutError() throws InterruptedException {
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
        TransactionManager txnManager = new TransactionManager(logContext, "testTransactionAbortedExceptionOnAbortWithoutError", 60000, 100, apiVersions);

        setupWithTransactionState(txnManager, false, null);
        doInitTransactions(txnManager, producerIdAndEpoch);
        // Begin the transaction
        txnManager.beginTransaction();
        txnManager.maybeAddPartition(tp0);
        client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp0, Errors.NONE)));
        // Run it once so that the partition is added to the transaction.
        sender.runOnce();
        // Append a record to the accumulator.
        FutureRecordMetadata metadata = appendToAccumulator(tp0, time.milliseconds(), "key", "value");
        // Now abort the transaction manually.
        txnManager.beginAbort();
        // Try to send.
        // This should abort the existing transaction and
        // drain all the unsent batches with a TransactionAbortedException.
        sender.runOnce();
        // Now attempt to fetch the result for the record.
        TestUtils.assertFutureThrows(metadata, TransactionAbortedException.class);
    }
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