Repository: apache/kafka
File: storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java (Java)

Code before:
                    if (duplicateBatch.isPresent()) {
                        return new AnalyzeAndValidateProducerStateResult(updatedProducers, completedTxns, duplicateBatch);
                    }
                }

                if (origin == AppendOrigin.CLIENT || origin == AppendOrigin.COORDINATOR) {
                    // Verify that if the record is transactional & the append origin is client/coordinator, that we either have an ongoing transaction or verified transaction state.
                    // This guarantees that transactional records are never written to the log outside of the transaction coordinator's knowledge of an open transaction on
                    // the partition. If we do not have an ongoing transaction or correct guard, return an error and do not append.
                    // There are two phases -- the first append to the log and subsequent appends.
                    //
                    // 1. First append: Verification starts with creating a VerificationGuard, sending a verification request to the transaction coordinator, and
                    // given a "verified" response, continuing the append path. (A non-verified response throws an error.) We create the unique VerificationGuard for the transaction
                    // to ensure there is no race between the transaction coordinator response and an abort marker getting written to the log. We need a unique guard because we could
                    // have a sequence of events where we start a transaction verification, have the transaction coordinator send a verified response, write an abort marker,
                    // start a new transaction not aware of the partition, and receive the stale verification (ABA problem). With a unique VerificationGuard, this sequence would not
                    // result in appending to the log and would return an error. The guard is removed after the first append to the transaction and from then, we can rely on phase 2.
                    //
                    // 2. Subsequent appends: Once we write to the transaction, the in-memory state currentTxnFirstOffset is populated. This field remains until the
                    // transaction is completed or aborted. We can guarantee the transaction coordinator knows about the transaction given step 1 and that the transaction is still
                    // ongoing. If the transaction is expected to be ongoing, we will not set a VerificationGuard. If the transaction is aborted, hasOngoingTransaction is false and
                    // requestVerificationGuard is the sentinel, so we will throw an error. A subsequent produce request (retry) should create verification state and return to phase 1.
                    if (batch.isTransactional()
                            && !hasOngoingTransaction(batch.producerId(), batch.producerEpoch())
                            && batchMissingRequiredVerification(batch, requestVerificationGuard)) {
                        throw new InvalidTxnStateException("Record was not part of an ongoing transaction");
                    }
                }

                // We cache offset metadata for the start of each transaction. This allows us to
                // compute the last stable offset without relying on additional index lookups.
                Optional<LogOffsetMetadata> firstOffsetMetadata = batch.isTransactional()
                    ? Optional.of(new LogOffsetMetadata(batch.baseOffset(), appendOffsetMetadata.segmentBaseOffset, relativePositionInSegment))
                    : Optional.empty();

                Optional<CompletedTxn> maybeCompletedTxn = UnifiedLog.updateProducers(producerStateManager, batch, updatedProducers, firstOffsetMetadata, origin);
                maybeCompletedTxn.ifPresent(completedTxns::add);
            }

            relativePositionInSegment += batch.sizeInBytes();
        }
        return new AnalyzeAndValidateProducerStateResult(updatedProducers, completedTxns, Optional.empty());
    }

    private boolean batchMissingRequiredVerification(MutableRecordBatch batch, VerificationGuard requestVerificationGuard) {
        return producerStateManager.producerStateManagerConfig().transactionVerificationEnabled()
                && !batch.isControlBatch()
                && !verificationGuard(batch.producerId()).verify(requestVerificationGuard);
    }

    /**

Review comment (bug):
One more question: I don't have the full bug context for Kafka streams, but should we cover additional cases for invalid batch epoch? for example: `batch.producerEpoch` > `entry.producerEpoch`

Code after:
                    if (duplicateBatch.isPresent()) {
                        return new AnalyzeAndValidateProducerStateResult(updatedProducers, completedTxns, duplicateBatch);
                    }
                }

                if (origin == AppendOrigin.CLIENT || origin == AppendOrigin.COORDINATOR) {
                    // Verify that if the record is transactional & the append origin is client/coordinator, that we either have an ongoing transaction or verified transaction state.
                    // This guarantees that transactional records are never written to the log outside of the transaction coordinator's knowledge of an open transaction on
                    // the partition. If we do not have an ongoing transaction or correct guard, return an error and do not append.
                    // There are two phases -- the first append to the log and subsequent appends.
                    //
                    // 1. First append: Verification starts with creating a VerificationGuard, sending a verification request to the transaction coordinator, and
                    // given a "verified" response, continuing the append path. (A non-verified response throws an error.) We create the unique VerificationGuard for the transaction
                    // to ensure there is no race between the transaction coordinator response and an abort marker getting written to the log. We need a unique guard because we could
                    // have a sequence of events where we start a transaction verification, have the transaction coordinator send a verified response, write an abort marker,
                    // start a new transaction not aware of the partition, and receive the stale verification (ABA problem). With a unique VerificationGuard, this sequence would not
                    // result in appending to the log and would return an error. The guard is removed after the first append to the transaction and from then, we can rely on phase 2.
                    //
                    // 2. Subsequent appends: Once we write to the transaction, the in-memory state currentTxnFirstOffset is populated. This field remains until the
                    // transaction is completed or aborted. We can guarantee the transaction coordinator knows about the transaction given step 1 and that the transaction is still
                    // ongoing. If the transaction is expected to be ongoing, we will not set a VerificationGuard. If the transaction is aborted, hasOngoingTransaction is false and
                    // requestVerificationGuard is the sentinel, so we will throw an error. A subsequent produce request (retry) should create verification state and return to phase 1.
                    if (batch.isTransactional() && !hasOngoingTransaction(batch.producerId(), batch.producerEpoch())) {
                        // Check epoch first: if producer epoch is stale, throw recoverable InvalidProducerEpochException.
                        ProducerStateEntry entry = producerStateManager.activeProducers().get(batch.producerId());
                        if (entry != null && batch.producerEpoch() < entry.producerEpoch()) {
                            String message = "Epoch of producer " + batch.producerId() + " is " + batch.producerEpoch() + 
                                ", which is smaller than the last seen epoch " + entry.producerEpoch();
                            throw new InvalidProducerEpochException(message);
                        }
                        
                        // Only check verification if epoch is current
                        if (batchMissingRequiredVerification(batch, requestVerificationGuard)) {
                            throw new InvalidTxnStateException("Record was not part of an ongoing transaction");
                        }
                    }
                }

                // We cache offset metadata for the start of each transaction. This allows us to
                // compute the last stable offset without relying on additional index lookups.
                Optional<LogOffsetMetadata> firstOffsetMetadata = batch.isTransactional()
                    ? Optional.of(new LogOffsetMetadata(batch.baseOffset(), appendOffsetMetadata.segmentBaseOffset, relativePositionInSegment))
                    : Optional.empty();

                Optional<CompletedTxn> maybeCompletedTxn = UnifiedLog.updateProducers(producerStateManager, batch, updatedProducers, firstOffsetMetadata, origin);
                maybeCompletedTxn.ifPresent(completedTxns::add);
            }

            relativePositionInSegment += batch.sizeInBytes();
        }
        return new AnalyzeAndValidateProducerStateResult(updatedProducers, completedTxns, Optional.empty());