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

Code before:
            return this.isEpochBump ? Priority.EPOCH_BUMP : Priority.INIT_PRODUCER_ID;
        }

        @Override
        FindCoordinatorRequest.CoordinatorType coordinatorType() {
            if (isTransactional()) {
                return FindCoordinatorRequest.CoordinatorType.TRANSACTION;
            } else {
                return null;
            }
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            InitProducerIdResponse initProducerIdResponse = (InitProducerIdResponse) response;
            Errors error = initProducerIdResponse.error();

            if (error == Errors.NONE) {
                ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
                        initProducerIdResponse.data().producerEpoch());
                setProducerIdAndEpoch(producerIdAndEpoch);
                // If this is a 2PC-enabled transaction with keepPreparedTxn=true, transition directly
                // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
                if (enable2PC &&
                    builder.data.keepPreparedTxn() &&
                    initProducerIdResponse.data().ongoingTxnProducerId() != -1
                ) {
                    transitionTo(State.PREPARED_TRANSACTION);
                    // Update the preparedTxnState with the ongoing pid and epoch from the response.
                    // This will be used to complete the transaction later.
                    String serializedState = initProducerIdResponse.data().ongoingTxnProducerId() + 
                            ":" + initProducerIdResponse.data().ongoingTxnProducerEpoch();
                    TransactionManager.this.preparedTxnState = new PreparedTxnState(serializedState);
                } else {
                    transitionTo(State.READY);
                }
                lastError = null;
                if (this.isEpochBump) {
                    resetSequenceNumbers();
                }
                result.done();
            } else if (error == Errors.NOT_COORDINATOR || error == Errors.COORDINATOR_NOT_AVAILABLE) {
                lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                reenqueue();
            } else if (error.exception() instanceof RetriableException) {
                reenqueue();
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
                    error == Errors.CLUSTER_AUTHORIZATION_FAILED) {
                log.info("Abortable authorization error: {}.  Transition the producer state to {}", error.message(), State.ABORTABLE_ERROR);
                lastError = error.exception();
                abortableError(error.exception());

Review comment (performance):
There is a `RecordBatch.NO_PRODUCER_ID` constant, we should use it.

Code after:

        @Override
        FindCoordinatorRequest.CoordinatorType coordinatorType() {
            if (isTransactional()) {
                return FindCoordinatorRequest.CoordinatorType.TRANSACTION;
            } else {
                return null;
            }
        }

        @Override
        public void handleResponse(AbstractResponse response) {
            InitProducerIdResponse initProducerIdResponse = (InitProducerIdResponse) response;
            Errors error = initProducerIdResponse.error();

            if (error == Errors.NONE) {
                ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.data().producerId(),
                        initProducerIdResponse.data().producerEpoch());
                setProducerIdAndEpoch(producerIdAndEpoch);
                // If this is a transaction with keepPreparedTxn=true, transition directly
                // to PREPARED_TRANSACTION state IFF there is an ongoing transaction.
                if (builder.data.keepPreparedTxn() &&
                    initProducerIdResponse.data().ongoingTxnProducerId() != RecordBatch.NO_PRODUCER_ID
                ) {
                    transitionTo(State.PREPARED_TRANSACTION);
                    // Update the preparedTxnState with the ongoing pid and epoch from the response.
                    // This will be used to complete the transaction later.
                    TransactionManager.this.preparedTxnState = new ProducerIdAndEpoch(
                        initProducerIdResponse.data().ongoingTxnProducerId(),
                        initProducerIdResponse.data().ongoingTxnProducerEpoch()
                    );
                } else {
                    transitionTo(State.READY);
                }
                lastError = null;
                if (this.isEpochBump) {
                    resetSequenceNumbers();
                }
                result.done();
            } else if (error == Errors.NOT_COORDINATOR || error == Errors.COORDINATOR_NOT_AVAILABLE) {
                lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
                reenqueue();
            } else if (error.exception() instanceof RetriableException) {
                reenqueue();
            } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED ||
                    error == Errors.CLUSTER_AUTHORIZATION_FAILED) {
                log.info("Abortable authorization error: {}.  Transition the producer state to {}", error.message(), State.ABORTABLE_ERROR);
                lastError = error.exception();
                abortableError(error.exception());
            } else if (error == Errors.INVALID_PRODUCER_EPOCH || error == Errors.PRODUCER_FENCED) {
                // We could still receive INVALID_PRODUCER_EPOCH from old versioned transaction coordinator,