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 testNodeLatencyStats() throws Exception {
        try (Metrics m = new Metrics()) {
            // Create a new record accumulator with non-0 partitionAvailabilityTimeoutMs
            // otherwise it wouldn't update the stats.
            RecordAccumulator.PartitionerConfig config = new RecordAccumulator.PartitionerConfig(false, 42);
            long totalSize = 1024 * 1024;
            accumulator = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
                DELIVERY_TIMEOUT_MS, config, m, "producer-metrics", time, apiVersions, null,
                new BufferPool(totalSize, batchSize, m, time, "producer-internal-metrics"));

            SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
            Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL, 1,
                senderMetrics, time, REQUEST_TIMEOUT, 1000L, null, new ApiVersions());

            // Produce and send batch.
            long time1 = time.milliseconds();
            appendToAccumulator(tp0, 0L, "key", "value");
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // We were able to send the batch out, so both the ready and drain values should be the same.
            RecordAccumulator.NodeLatencyStats stats = accumulator.getNodeLatencyStats(0);
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time1, stats.readyTimeMs);

            // Make the node 1 not ready.
            client.throttle(metadata.fetch().nodeById(0), 100);

            // Time passes, but we don't have anything to send.
            time.sleep(10);
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // Stats shouldn't change as we didn't have anything ready.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time1, stats.readyTimeMs);

            // Produce a new batch, but we won't be able to send it because node is not ready.
            long time2 = time.milliseconds();
            appendToAccumulator(tp0, 0L, "key", "value");
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // The ready time should move forward, but drain time shouldn't change.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);

            // Time passes, we keep trying to send, but the node is not ready.
            time.sleep(10);
            time2 = time.milliseconds();
            sender.runOnce();
            assertEquals(1, client.inFlightRequestCount(), "We should have a single produce request in flight.");

            // The ready time should move forward, but drain time shouldn't change.
            assertEquals(time1, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);

            // Finally, time passes beyond the throttle and the node is ready.
            time.sleep(100);
            time2 = time.milliseconds();
            sender.runOnce();
            assertEquals(2, client.inFlightRequestCount(), "We should have 2 produce requests in flight.");

            // Both times should move forward
            assertEquals(time2, stats.drainTimeMs);
            assertEquals(time2, stats.readyTimeMs);
        }
    }