bob-barrett commented on a change in pull request #8239:
URL: https://github.com/apache/kafka/pull/8239#discussion_r454920777
##########
File path:
core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
##########
@@ -394,8 +395,8 @@ class TransactionCoordinator(brokerId: Int,
if (nextState == PrepareAbort &&
txnMetadata.pendingState.contains(PrepareEpochFence)) {
// We should clear the pending state to make way for the
transition to PrepareAbort and also bump
// the epoch in the transaction metadata we are about to
append.
+ isEpochFence = true
txnMetadata.pendingState = None
- txnMetadata.lastProducerEpoch = txnMetadata.producerEpoch
Review comment:
So I think this was a small issue with the KIP-360 implementation. When
we fence an epoch, we intend to transition to a -1 value for the last epoch
(since we only need to save the last epoch when re-initializing, in order to
detect retries):
```
def prepareFenceProducerEpoch(): TxnTransitMetadata = {
if (producerEpoch == Short.MaxValue)
throw new IllegalStateException(s"Cannot fence producer with epoch
equal to Short.MaxValue since this would overflow")
// If we've already failed to fence an epoch (because the write to the
log failed), we don't increase it again.
// This is safe because we never return the epoch to client if we fail
to fence the epoch
val bumpedEpoch = if (hasFailedEpochFence) producerEpoch else
(producerEpoch + 1).toShort
prepareTransitionTo(PrepareEpochFence, producerId, bumpedEpoch,
RecordBatch.NO_PRODUCER_EPOCH, txnTimeoutMs,
topicPartitions.toSet, txnStartTimestamp, txnLastUpdateTimestamp)
}
```
But if we fence a producer as part of an InitProducerId call, we immediately
call `endTransactions`, and we don't pass the new last epoch value. So this
line would wind up overwriting the -1 with the epoch that was getting fenced.
When I went back to figure out why I had made this change, I realized that
just removing the line isn't quite right either. We should actually set the
value explicitly to -1. There's no case when we'd want any value other than -1
as part of ending transactions, since we only need to save the last epoch when
we successfully re-initialize a PID, and we'd want to clear the value if it
previously existed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
