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

Code before:
        updateLastAckedOffset(response, batch);
        removeInFlightBatch(batch);
    }

    public synchronized void transitionToUninitialized(RuntimeException exception) {
        transitionTo(State.UNINITIALIZED);
        if (pendingTransition != null) {
            pendingTransition.result.fail(exception);
        }
        lastError = null;
    }

    public synchronized void maybeTransitionToErrorState(RuntimeException exception) {
        if (exception instanceof ClusterAuthorizationException
                || exception instanceof TransactionalIdAuthorizationException
                || exception instanceof ProducerFencedException
                || exception instanceof UnsupportedVersionException
                || exception instanceof InvalidPidMappingException) {
            transitionToFatalError(exception);
        } else if (isTransactional()) {
            if (needToTriggerEpochBumpFromClient() && !isCompleting()) {
                clientSideEpochBumpRequired = true;
            }
            transitionToAbortableError(exception);
        }
    }

    synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
        maybeTransitionToErrorState(exception);
        removeInFlightBatch(batch);

        if (hasFatalError()) {
            log.debug("Ignoring batch {} with producer id {}, epoch {}, and sequence number {} " +
                            "since the producer is already in fatal error state", batch, batch.producerId(),
                    batch.producerEpoch(), batch.baseSequence(), exception);
            return;
        }

        if (exception instanceof OutOfOrderSequenceException && !isTransactional()) {
            log.error("The broker returned {} for topic-partition {} with producerId {}, epoch {}, and sequence number {}",
                    exception, batch.topicPartition, batch.producerId(), batch.producerEpoch(), batch.baseSequence());

            // If we fail with an OutOfOrderSequenceException, we have a gap in the log. Bump the epoch for this
            // partition, which will reset the sequence number to 0 and allow us to continue
            requestIdempotentEpochBumpForPartition(batch.topicPartition);
        } else if (exception instanceof UnknownProducerIdException) {
            // If we get an UnknownProducerId for a partition, then the broker has no state for that producer. It will
            // therefore accept a write with sequence number 0. We reset the sequence number for the partition here so
            // that the producer can continue after aborting the transaction. All inflight-requests to this partition
            // will also fail with an UnknownProducerId error, so the sequence will remain at 0. Note that if the
            // broker supports bumping the epoch, we will later reset all sequence numbers after calling InitProducerId

Review comment (refactor):
maybe not a change we need now, but it isn't totally clear from the method name that this should only be called from the sender thread. Maybe we should refactor this in the future.

Code after:
                lastAckedSequence);

        updateLastAckedOffset(response, batch);
        removeInFlightBatch(batch);
    }

    public synchronized void transitionToUninitialized(RuntimeException exception) {
        transitionTo(State.UNINITIALIZED);
        if (pendingTransition != null) {
            pendingTransition.result.fail(exception);
        }
        lastError = null;
    }

    public synchronized void maybeTransitionToErrorState(RuntimeException exception) {
        if (exception instanceof ClusterAuthorizationException
                || exception instanceof TransactionalIdAuthorizationException
                || exception instanceof ProducerFencedException
                || exception instanceof UnsupportedVersionException
                || exception instanceof InvalidPidMappingException) {
            transitionToFatalError(exception);
        } else if (isTransactional()) {
            // RetriableExceptions from the Sender thread are converted to Abortable errors
            // because they indicate that the transaction cannot be completed after all retry attempts.
            // This conversion ensures the application layer treats these errors as abortable,
            // preventing duplicate message delivery.
            if (exception instanceof RetriableException ||
                    exception instanceof InvalidTxnStateException) {
                exception = new TransactionAbortableException("Transaction Request was aborted after exhausting retries.", exception);
            }

            if (needToTriggerEpochBumpFromClient() && !isCompleting()) {
                clientSideEpochBumpRequired = true;
            }
            transitionToAbortableError(exception);
        }
    }

    synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
        maybeTransitionToErrorState(exception);
        removeInFlightBatch(batch);

        if (hasFatalError()) {
            log.debug("Ignoring batch {} with producer id {}, epoch {}, and sequence number {} " +
                            "since the producer is already in fatal error state", batch, batch.producerId(),
                    batch.producerEpoch(), batch.baseSequence(), exception);
            return;
        }

        if (exception instanceof OutOfOrderSequenceException && !isTransactional()) {
            log.error("The broker returned {} for topic-partition {} with producerId {}, epoch {}, and sequence number {}",