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 testInvalidTxnStateIsAnAbortableError() throws Exception {
        ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
        apiVersions.update("0", NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3));
        TransactionManager txnManager = new TransactionManager(logContext, "testInvalidTxnState", 60000, 100, apiVersions);

        setupWithTransactionState(txnManager);
        doInitTransactions(txnManager, producerIdAndEpoch);

        txnManager.beginTransaction();
        txnManager.maybeAddPartition(tp0);
        client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp0, Errors.NONE)));
        sender.runOnce();

        Future<RecordMetadata> request = appendToAccumulator(tp0);
        sender.runOnce();  // send request
        sendIdempotentProducerResponse(0, tp0, Errors.INVALID_TXN_STATE, -1);

        // Return InvalidTxnState error. It should be abortable.
        sender.runOnce();
        assertFutureFailure(request, InvalidTxnStateException.class);
        assertTrue(txnManager.hasAbortableError());
        TransactionalRequestResult result = txnManager.beginAbort();
        sender.runOnce();

        // Once the transaction is aborted, we should be able to begin a new one.
        respondToEndTxn(Errors.NONE);
        sender.runOnce();
        assertTrue(txnManager::isInitializing);
        prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
        sender.runOnce();
        assertTrue(txnManager::isReady);

        assertTrue(result.isSuccessful());
        result.await();

        txnManager.beginTransaction();
    }