Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java (Java)

Code before:

            InitProducerIdHandler handler = new InitProducerIdHandler(new InitProducerIdRequest.Builder(requestData),
                    isEpochBump);
            enqueueRequest(handler);
            return handler.result;
        }, State.INITIALIZING, "initTransactions");
    }

    public synchronized void beginTransaction() {
        ensureTransactional();
        throwIfPendingState("beginTransaction");
        maybeFailWithError();
        transitionTo(State.IN_TRANSACTION);
    }

    /**
     * Prepare a transaction for a two-phase commit.
     * This transitions the transaction to the PREPARED_TRANSACTION state.
     * The preparedTxnState is set with the current producer ID and epoch.
     */
    public synchronized void prepareTransaction() {
        ensureTransactional();
        throwIfPendingState("prepareTransaction");
        maybeFailWithError();
        transitionTo(State.PREPARED_TRANSACTION);
        this.preparedTxnState = new PreparedTxnState(
            this.producerIdAndEpoch.producerId,
            this.producerIdAndEpoch.epoch
        );
    }

    public synchronized TransactionalRequestResult beginCommit() {
        return handleCachedTransactionRequestResult(() -> {
            maybeFailWithError();
            transitionTo(State.COMMITTING_TRANSACTION);
            return beginCompletingTransaction(TransactionResult.COMMIT);
        }, State.COMMITTING_TRANSACTION, "commitTransaction");
    }

    public synchronized TransactionalRequestResult beginAbort() {
        return handleCachedTransactionRequestResult(() -> {
            if (currentState != State.ABORTABLE_ERROR)
                maybeFailWithError();
            transitionTo(State.ABORTING_TRANSACTION);

            // We're aborting the transaction, so there should be no need to add new partitions
            newPartitionsInTransaction.clear();
            return beginCompletingTransaction(TransactionResult.ABORT);
        }, State.ABORTING_TRANSACTION, "abortTransaction");
    }


Review comment (suggestion):
Let's just change the type of `preparedTxnState` member to be `private volatile ProducerIdAndEpoch preparedTxnState` then we can just assign numbers to it (it even has a convenient `NONE` constant).  Then the `KafkaProducer` would create the corresponding `PreparedTxnState` object.

Code after:
            InitProducerIdHandler handler = new InitProducerIdHandler(new InitProducerIdRequest.Builder(requestData),
                    isEpochBump);
            enqueueRequest(handler);
            return handler.result;
        }, State.INITIALIZING, "initTransactions");
    }

    public synchronized void beginTransaction() {
        ensureTransactional();
        throwIfPendingState("beginTransaction");
        maybeFailWithError();
        transitionTo(State.IN_TRANSACTION);
    }

    /**
     * Prepare a transaction for a two-phase commit.
     * This transitions the transaction to the PREPARED_TRANSACTION state.
     * The preparedTxnState is set with the current producer ID and epoch.
     */
    public synchronized void prepareTransaction() {
        ensureTransactional();
        throwIfPendingState("prepareTransaction");
        maybeFailWithError();
        transitionTo(State.PREPARED_TRANSACTION);
        this.preparedTxnState = new ProducerIdAndEpoch(
            this.producerIdAndEpoch.producerId,
            this.producerIdAndEpoch.epoch
        );
    }

    public synchronized TransactionalRequestResult beginCommit() {
        return handleCachedTransactionRequestResult(() -> {
            maybeFailWithError();
            transitionTo(State.COMMITTING_TRANSACTION);
            return beginCompletingTransaction(TransactionResult.COMMIT);
        }, State.COMMITTING_TRANSACTION, "commitTransaction");
    }

    public synchronized TransactionalRequestResult beginAbort() {
        return handleCachedTransactionRequestResult(() -> {
            if (currentState != State.ABORTABLE_ERROR)
                maybeFailWithError();
            transitionTo(State.ABORTING_TRANSACTION);

            // We're aborting the transaction, so there should be no need to add new partitions
            newPartitionsInTransaction.clear();
            return beginCompletingTransaction(TransactionResult.ABORT);
        }, State.ABORTING_TRANSACTION, "abortTransaction");
    }

    private TransactionalRequestResult beginCompletingTransaction(TransactionResult transactionResult) {