@Test
    public void testAwaitPendingRecordsBeforeCommittingTransaction() throws Exception {
        try (Metrics m = new Metrics()) {
            SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

            TransactionManager txnManager = new TransactionManager(logContext, "txnId", 6000, 100, apiVersions);
            setupWithTransactionState(txnManager);

            Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
                1, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, txnManager, apiVersions);

            // Begin a transaction and successfully add one partition to it.
            ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
            doInitTransactions(txnManager, producerIdAndEpoch);
            txnManager.beginTransaction();
            addPartitionToTxn(sender, txnManager, tp0);

            // Send one Produce request.
            appendToAccumulator(tp0);
            runUntil(sender, () -> client.requests().size() == 1);
            assertFalse(accumulator.hasUndrained());
            assertTrue(client.hasInFlightRequests());
            assertTrue(txnManager.hasInflightBatches(tp0));

            // Enqueue another record and then commit the transaction. We expect the unsent record to
            // get sent before the transaction can be completed.
            appendToAccumulator(tp0);
            txnManager.beginCommit();
            runUntil(sender, () -> client.requests().size() == 2);

            assertTrue(txnManager.isCompleting());
            assertFalse(txnManager.hasInFlightRequest());
            assertTrue(txnManager.hasInflightBatches(tp0));

            // Now respond to the pending Produce requests.
            respondToProduce(tp0, Errors.NONE, 0L);
            respondToProduce(tp0, Errors.NONE, 1L);
            runUntil(sender, txnManager::hasInFlightRequest);

            // Finally, respond to the expected EndTxn request.
            respondToEndTxn(Errors.NONE);
            runUntil(sender, txnManager::isReady);
        }
    }
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