void runOnce() {
        if (transactionManager != null) {
            try {
                transactionManager.maybeResolveSequences();

                RuntimeException lastError = transactionManager.lastError();

                // do not continue sending if the transaction manager is in a failed state
                if (transactionManager.hasFatalError()) {
                    if (lastError != null)
                        maybeAbortBatches(lastError);
                    client.poll(retryBackoffMs, time.milliseconds());
                    return;
                }

                if (transactionManager.hasAbortableError() && shouldHandleAuthorizationError(lastError)) {
                    return;
                }

                // Check whether we need a new producerId. If so, we will enqueue an InitProducerId
                // request which will be sent below
                transactionManager.bumpIdempotentEpochAndResetIdIfNeeded();

                if (maybeSendAndPollTransactionalRequest()) {
                    return;
                }
            } catch (AuthenticationException e) {
                // This is already logged as error, but propagated here to perform any clean ups.
                log.trace("Authentication exception while processing transactional request", e);
                transactionManager.authenticationFailed(e);
            }
        }

        long currentTimeMs = time.milliseconds();
        long pollTimeout = sendProducerData(currentTimeMs);
        client.poll(pollTimeout, currentTimeMs);
    }
@Test
    public void testRecordsFlushedImmediatelyOnTransactionCompletion() throws Exception {
        try (Metrics m = new Metrics()) {
            int lingerMs = 50;
            SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

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

            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 a couple records and assert that they are not sent immediately (due to linger).
            appendToAccumulator(tp0);
            appendToAccumulator(tp0);
            sender.runOnce();
            assertFalse(client.hasInFlightRequests());

            // Now begin the commit and assert that the Produce request is sent immediately
            // without waiting for the linger.
            TransactionalRequestResult commitResult = txnManager.beginCommit();
            runUntil(sender, client::hasInFlightRequests);

            // Respond to the produce request and wait for the EndTxn request to be sent.
            respondToProduce(tp0, Errors.NONE, 1L);
            runUntil(sender, txnManager::hasInFlightRequest);

            // Respond to the expected EndTxn request.
            respondToEndTxn(Errors.NONE);
            runUntil(sender, txnManager::isReady);

            assertTrue(commitResult.isSuccessful());
            commitResult.await();

            // Finally, we want to assert that the linger time is still effective
            // when the new transaction begins.
            txnManager.beginTransaction();
            addPartitionToTxn(sender, txnManager, tp0);

            appendToAccumulator(tp0);
            appendToAccumulator(tp0);
            time.sleep(lingerMs - 1);
            sender.runOnce();
            assertFalse(client.hasInFlightRequests());
            assertTrue(accumulator.hasUndrained());

            time.sleep(1);
            runUntil(sender, client::hasInFlightRequests);
            assertFalse(accumulator.hasUndrained());
        }
    }