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 testEpochBumpOnOutOfOrderSequenceForNextBatchWhenBatchInFlightFails() throws Exception {
        // When a batch failed after the producer epoch is bumped, the sequence number of
        // that partition must be reset for any subsequent batches sent.
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();

        // Retries once
        setupWithTransactionState(transactionManager, false, null, true, 1, 0);

        // Init producer id/epoch
        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        assertEquals(producerId, transactionManager.producerIdAndEpoch().producerId);
        assertEquals(0, transactionManager.producerIdAndEpoch().epoch);

        // Partition 0 - Send first batch
        appendToAccumulator(tp0);
        sender.runOnce();

        // Partition 0 - State is lazily initialized
        assertPartitionState(transactionManager, tp0, producerId, (short) 0, 1, OptionalInt.empty());

        // Partition 0 - Successful response
        sendIdempotentProducerResponse(0, 0, tp0, Errors.NONE, 0, -1);
        sender.runOnce();

        // Partition 0 - Last ack is updated
        assertPartitionState(transactionManager, tp0, producerId, (short) 0, 1, OptionalInt.of(0));

        // Partition 1 - Send first batch
        appendToAccumulator(tp1);
        sender.runOnce();

        // Partition 1 - State is lazily initialized
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 1, OptionalInt.empty());

        // Partition 1 - Successful response
        sendIdempotentProducerResponse(0, 0, tp1, Errors.NONE, 0, -1);
        sender.runOnce();

        // Partition 1 - Last ack is updated
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 1, OptionalInt.of(0));

        // Partition 0 - Send second batch
        appendToAccumulator(tp0);
        sender.runOnce();

        // Partition 0 - Sequence is incremented
        assertPartitionState(transactionManager, tp0, producerId, (short) 0, 2, OptionalInt.of(0));

        // Partition 1 - Send second batch
        appendToAccumulator(tp1);
        sender.runOnce();

        // Partition 1 - Sequence is incremented
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 2, OptionalInt.of(0));

        // Partition 0 - Failed response with OUT_OF_ORDER_SEQUENCE_NUMBER
        sendIdempotentProducerResponse(0, 1, tp0, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, -1, -1);
        sender.runOnce(); // Receive
        sender.runOnce(); // Bump epoch & Retry

        // Producer epoch is bumped
        assertEquals(1, transactionManager.producerIdAndEpoch().epoch);

        // Partition 0 - State is reset to current producer epoch
        assertPartitionState(transactionManager, tp0, producerId, (short) 1, 1, OptionalInt.empty());

        // Partition 1 - State is not changed. The epoch will be lazily bumped when all in-flight
        // batches are completed
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 2, OptionalInt.of(0));
        assertTrue(transactionManager.hasStaleProducerIdAndEpoch(tp1));

        // Partition 1 - Failed response with NOT_LEADER_OR_FOLLOWER
        sendIdempotentProducerResponse(0, 1, tp1, Errors.NOT_LEADER_OR_FOLLOWER, -1, -1);
        sender.runOnce(); // Receive & Retry

        // Partition 1 - State is not changed.
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 2, OptionalInt.of(0));
        assertTrue(transactionManager.hasStaleProducerIdAndEpoch(tp1));

        // Partition 0 - Successful Response
        sendIdempotentProducerResponse(1, 0, tp0, Errors.NONE, 1, -1);
        sender.runOnce();

        // Partition 0 - Last ack is updated
        assertPartitionState(transactionManager, tp0, producerId, (short) 1, 1, OptionalInt.of(0));

        // Partition 1 - Failed response with NOT_LEADER_OR_FOLLOWER
        sendIdempotentProducerResponse(0, 1, tp1, Errors.NOT_LEADER_OR_FOLLOWER, -1, -1);
        sender.runOnce(); // Receive & Fail the batch (retries exhausted)

        // Partition 1 - State is not changed. It will be lazily updated when the next batch is sent.
        assertPartitionState(transactionManager, tp1, producerId, (short) 0, 2, OptionalInt.of(0));
        assertTrue(transactionManager.hasStaleProducerIdAndEpoch(tp1));

        // Partition 1 - Send third batch
        appendToAccumulator(tp1);
        sender.runOnce();

        // Partition 1 - Epoch is bumped, sequence is reset
        assertPartitionState(transactionManager, tp1, producerId, (short) 1, 1, OptionalInt.empty());
        assertFalse(transactionManager.hasStaleProducerIdAndEpoch(tp1));

        // Partition 1 - Successful Response
        sendIdempotentProducerResponse(1, 0, tp1, Errors.NONE, 0, -1);
        sender.runOnce();

        // Partition 1 - Last ack is updated
        assertPartitionState(transactionManager, tp1, producerId, (short) 1, 1, OptionalInt.of(0));

        // Partition 0 - Send third batch
        appendToAccumulator(tp0);
        sender.runOnce();

        // Partition 0 - Sequence is incremented
        assertPartitionState(transactionManager, tp0, producerId, (short) 1, 2, OptionalInt.of(0));

        // Partition 0 - Successful Response
        sendIdempotentProducerResponse(1, 1, tp0, Errors.NONE, 0, -1);
        sender.runOnce();

        // Partition 0 - Last ack is updated
        assertPartitionState(transactionManager, tp0, producerId, (short) 1, 2, OptionalInt.of(1));
    }