KAFKA-18884: Move TransactionMetadata to transaction-coordinator module#19699
Conversation
Signed-off-by: PoAn Yang <payang@apache.org>
Signed-off-by: PoAn Yang <payang@apache.org>
| // the authorization check to indicate that they were not added to the transaction. | ||
| val partitionErrors = unauthorizedTopicErrors ++ nonExistingTopicErrors ++ | ||
| authorizedPartitions.map(_ -> Errors.OPERATION_NOT_ATTEMPTED) | ||
| authorizedPartitions.stream().map(_ -> Errors.OPERATION_NOT_ATTEMPTED).toList.asScala.toSeq |
| ); | ||
| } | ||
|
|
||
| public Either<Errors, TxnTransitMetadata> prepareIncrementProducerEpoch( |
There was a problem hiding this comment.
it can throw exception and callers can convert the exception to error. we don't need to introduce Either I think
Signed-off-by: PoAn Yang <payang@apache.org>
|
@chia7712, I addressed all comments. Could you review again when you have time? Thanks. |
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for this patch
| throw new IllegalStateException("Cannot allocate any more producer epochs for producerId " + producerId); | ||
|
|
||
| short bumpedEpoch = (short) (producerEpoch + 1); | ||
| Short produceEpochResult; |
|
|
||
| short bumpedEpoch = (short) (producerEpoch + 1); | ||
| Short produceEpochResult; | ||
| Short lastProducerEpochResult; |
| TransactionState.PREPARE_ABORT | ||
|
|
||
| if (nextState == TransactionState.PREPARE_ABORT && txnMetadata.pendingState.contains(TransactionState.PREPARE_EPOCH_FENCE)) { | ||
| if (nextState == TransactionState.PREPARE_ABORT && txnMetadata.pendingState.isPresent && txnMetadata.pendingState.get.equals(TransactionState.PREPARE_EPOCH_FENCE)) { |
There was a problem hiding this comment.
txnMetadata.pendingState.filter(s => s == TransactionState.PREPARE_EPOCH_FENCE).isPresent)
| } else { | ||
| txnMetadata.prepareIncrementProducerEpoch(transactionTimeoutMs, expectedProducerIdAndEpoch.map(_.epoch), | ||
| time.milliseconds()) | ||
| try { |
There was a problem hiding this comment.
try {
if (txnMetadata.isProducerEpochExhausted && expectedProducerIdAndEpoch.forall(_.epoch == txnMetadata.producerEpoch))
Right(txnMetadata.prepareProducerIdRotation(producerIdManager.generateProducerId(), transactionTimeoutMs, time.milliseconds(),
expectedProducerIdAndEpoch.isDefined))
else
Right(txnMetadata.prepareIncrementProducerEpoch(transactionTimeoutMs, expectedProducerIdAndEpoch.map(e => Short.box(e.epoch)).toJava,
time.milliseconds()))
} catch {
case e: Exception => Left(Errors.forException(e))
}| // PrepareEpochFence has slightly different epoch bumping logic so don't include it here. | ||
| // Note that, it can only happen when the current state is Ongoing. | ||
| isEpochFence = txnMetadata.pendingState.contains(TransactionState.PREPARE_EPOCH_FENCE) | ||
| isEpochFence = txnMetadata.pendingState.isPresent && txnMetadata.pendingState.get.equals(TransactionState.PREPARE_EPOCH_FENCE) |
There was a problem hiding this comment.
txnMetadata.pendingState.filter(s => s == TransactionState.PREPARE_EPOCH_FENCE).isPresent
| txnMetadata.inLock(() => { | ||
| if (txnMetadataCacheEntry.coordinatorEpoch == idCoordinatorEpochAndMetadata.coordinatorEpoch | ||
| && txnMetadata.pendingState.contains(TransactionState.DEAD) | ||
| && txnMetadata.pendingState.isPresent && txnMetadata.pendingState.get.equals(TransactionState.DEAD) |
There was a problem hiding this comment.
txnMetadata.pendingState.filter(s => s == TransactionState.DEAD).isPresent
| // the authorization check to indicate that they were not added to the transaction. | ||
| val partitionErrors = unauthorizedTopicErrors ++ nonExistingTopicErrors ++ | ||
| authorizedPartitions.map(_ -> Errors.OPERATION_NOT_ATTEMPTED) | ||
| authorizedPartitions.stream().map(_ -> Errors.OPERATION_NOT_ATTEMPTED).toList.asScala |
There was a problem hiding this comment.
authorizedPartitions.asScala.map(_ -> Errors.OPERATION_NOT_ATTEMPTED)
| private volatile long txnStartTimestamp; | ||
| private volatile long txnLastUpdateTimestamp; | ||
| private TransactionVersion clientTransactionVersion; | ||
| private static final Logger LOGGER = LoggerFactory.getLogger(TransactionMetadata.class); |
Signed-off-by: PoAn Yang <payang@apache.org>
Signed-off-by: PoAn Yang <payang@apache.org>
|
@chia7712 Thanks for the suggestion. I address all comments. |
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for this patch. a couple of comments are left. PTAL
| short producerEpoch = TransactionMetadata.this.producerEpoch; | ||
| short lastProducerEpoch = TransactionMetadata.this.lastProducerEpoch; | ||
| int txnTimeoutMs = TransactionMetadata.this.txnTimeoutMs; | ||
| Set<TopicPartition> topicPartitions = Set.copyOf(TransactionMetadata.this.topicPartitions); |
There was a problem hiding this comment.
not sure why a deep copy is necessary?
| public TxnTransitMetadata prepareNoTransit() { | ||
| // do not call transitTo as it will set the pending state, a follow-up call to abort the transaction will set its pending state | ||
| return new TxnTransitMetadata(producerId, prevProducerId, nextProducerId, producerEpoch, lastProducerEpoch, txnTimeoutMs, | ||
| state, new HashSet<>(topicPartitions), txnStartTimestamp, txnLastUpdateTimestamp, clientTransactionVersion); |
| // In the case of a new producer, producerEpoch will be -1 and bumpedEpoch will be 0 | ||
| produceEpochResult = bumpedEpoch; | ||
| lastProducerEpochResult = RecordBatch.NO_PRODUCER_EPOCH; | ||
| } else { |
There was a problem hiding this comment.
please flatten it.
var data = new TransitionData(TransactionState.EMPTY);
short bumpedEpoch = (short) (producerEpoch + 1);
if (expectedProducerEpoch.isEmpty()) {
// If no expected epoch was provided by the producer, bump the current epoch and set the last epoch to -1
// In the case of a new producer, producerEpoch will be -1 and bumpedEpoch will be 0
data.producerEpoch = bumpedEpoch;
data.lastProducerEpoch = RecordBatch.NO_PRODUCER_EPOCH;
} else if (producerEpoch == RecordBatch.NO_PRODUCER_EPOCH || expectedProducerEpoch.get() == producerEpoch) {
// If the expected epoch matches the current epoch, or if there is no current epoch, the producer is attempting
// to continue after an error and no other producer has been initialized. Bump the current and last epochs.
// The no current epoch case means this is a new producer; producerEpoch will be -1 and bumpedEpoch will be 0
data.producerEpoch = bumpedEpoch;
data.lastProducerEpoch = producerEpoch;
} else if (expectedProducerEpoch.get() == lastProducerEpoch) {
// If the expected epoch matches the previous epoch, it is a retry of a successful call, so just return the
// current epoch without bumping. There is no danger of this producer being fenced, because a new producer
// calling InitProducerId would have caused the last epoch to be set to -1.
// Note that if the IBP is prior to 2.4.IV1, the lastProducerId and lastProducerEpoch will not be written to
// the transaction log, so a retry that spans a coordinator change will fail. We expect this to be a rare case.
data.producerEpoch = producerEpoch;
data.lastProducerEpoch = lastProducerEpoch;
} else {
// Otherwise, the producer has a fenced epoch and should receive an PRODUCER_FENCED error
LOGGER.info("Expected producer epoch {} does not match current producer epoch {} or previous producer epoch {}",
expectedProducerEpoch.get(), producerEpoch, lastProducerEpoch);
throw Errors.PRODUCER_FENCED.exception();
}
data.txnTimeoutMs = newTxnTimeoutMs;
data.topicPartitions = Set.of();
data.txnStartTimestamp = -1L;
data.txnLastUpdateTimestamp = updateTimestamp;| long nextProducerId, | ||
| long updateTimestamp, | ||
| boolean noPartitionAdded) { | ||
| short updatedProducerEpoch; |
There was a problem hiding this comment.
var data = new TransitionData(newState);
if (clientTransactionVersion.supportsEpochBump()) {
// We already ensured that we do not overflow here. MAX_SHORT is the highest possible value.
data.producerEpoch = (short) (producerEpoch + 1);
data.lastProducerEpoch = producerEpoch;
} else {
data.producerEpoch = producerEpoch;
data.lastProducerEpoch = lastProducerEpoch;
}
// With transaction V2, it is allowed to abort the transaction without adding any partitions. Then, the transaction
// start time is uncertain but it is still required. So we can use the update time as the transaction start time.
data.txnStartTimestamp = noPartitionAdded ? updateTimestamp : txnStartTimestamp;
data.nextProducerId = nextProducerId;
data.txnLastUpdateTimestamp = updateTimestamp;
data.clientTransactionVersion = clientTransactionVersion;
return prepareTransitionTo(data);| } | ||
|
|
||
| public TxnTransitMetadata prepareComplete(long updateTimestamp) { | ||
| TransactionState newState = state == TransactionState.PREPARE_COMMIT ? |
There was a problem hiding this comment.
// Since the state change was successfully written to the log, unset the flag for a failed epoch fence
hasFailedEpochFence = false;
var data = new TransitionData(state == TransactionState.PREPARE_COMMIT ?
TransactionState.COMPLETE_COMMIT : TransactionState.COMPLETE_ABORT);
// In the prepareComplete transition for the overflow case, the lastProducerEpoch is kept at MAX-1,
// which is the last epoch visible to the client.
// Internally, however, during the transition between prepareAbort/prepareCommit and prepareComplete, the producer epoch
// reaches MAX but the client only sees the transition as MAX-1 followed by 0.
// When an epoch overflow occurs, we set the producerId to nextProducerId and reset the epoch to 0,
// but lastProducerEpoch remains MAX-1 to maintain consistency with what the client last saw.
if (clientTransactionVersion.supportsEpochBump() && nextProducerId != RecordBatch.NO_PRODUCER_ID) {
data.producerId = nextProducerId;
data.producerEpoch = 0;
} else {
data.producerId = producerId;
data.producerEpoch = producerEpoch;
}
data.nextProducerId = RecordBatch.NO_PRODUCER_ID;
data.topicPartitions = Set.of();
data.txnLastUpdateTimestamp = updateTimestamp;
return prepareTransitionTo(data);| long transitProducerId = transitMetadata.producerId(); | ||
| short transitLastProducerEpoch = transitMetadata.lastProducerEpoch(); | ||
|
|
||
| if (isAtLeastTransactionsV2 && |
There was a problem hiding this comment.
if (isAtLeastTransactionsV2 &&
(txnState == TransactionState.COMPLETE_COMMIT || txnState == TransactionState.COMPLETE_ABORT) &&
transitProducerEpoch == 0) {
return transitLastProducerEpoch == lastProducerEpoch && transitMetadata.prevProducerId() == producerId;
}
if (isAtLeastTransactionsV2 &&
(txnState == TransactionState.PREPARE_COMMIT || txnState == TransactionState.PREPARE_ABORT)) {
return transitLastProducerEpoch == producerEpoch && transitProducerId == producerId;
}
return transitProducerEpoch == producerEpoch && transitProducerId == producerId;| return txnLastUpdateTimestamp; | ||
| } | ||
|
|
||
| public void setClientTransactionVersion(TransactionVersion clientTransactionVersion) { |
There was a problem hiding this comment.
Could we remove the prefix set?
| assertEquals(0, channelManager.numTxnsWithPendingMarkers) | ||
| assertEquals(0, channelManager.queueForBroker(broker1.id).get.totalNumMarkers) | ||
| assertEquals(None, txnMetadata2.pendingState) | ||
| assertTrue(txnMetadata2.pendingState.isEmpty) |
There was a problem hiding this comment.
assertEquals(Optional.empty(), txnMetadata2.pendingState)| assertEquals(0, channelManager.numTxnsWithPendingMarkers) | ||
| assertEquals(0, channelManager.queueForBroker(broker1.id).get.totalNumMarkers) | ||
| assertEquals(None, txnMetadata2.pendingState) | ||
| assertTrue(txnMetadata2.pendingState.isEmpty) |
There was a problem hiding this comment.
assertEquals(Optional.empty(), txnMetadata2.pendingState)| assertEquals(0, channelManager.numTxnsWithPendingMarkers) | ||
| assertEquals(0, channelManager.queueForBroker(broker1.id).get.totalNumMarkers) | ||
| assertEquals(None, txnMetadata2.pendingState) | ||
| assertTrue(txnMetadata2.pendingState.isEmpty) |
Signed-off-by: PoAn Yang <payang@apache.org>
|
@FrankYang0529 @chia7712 How are things going here? I can help review if we need to get it over the line. |
|
I just merged trunk, but CI cannot pass. I need some time to fix these. Thanks. |
Signed-off-by: PoAn Yang <payang@apache.org>
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for this patch. I have a couple of questions. Please take a look
| return pendingState.isPresent(); | ||
| } | ||
|
|
||
| public ReentrantLock lock() { |
There was a problem hiding this comment.
Could we avoid exposing the lock, even during testing?
| producerEpoch = transitMetadata.producerEpoch(); | ||
| lastProducerEpoch = transitMetadata.lastProducerEpoch(); | ||
| txnTimeoutMs = transitMetadata.txnTimeoutMs(); | ||
| topicPartitions = transitMetadata.topicPartitions(); |
There was a problem hiding this comment.
Line#135 uses Set.copyOf to create an immutable copy, which makes topicPartitions immutable. This could potentially lead a bug.
Perhaps we should use a HashSet instead to ensure mutability
| " while it already a pending state " + pendingState.get()); | ||
|
|
||
| if (data.producerId < 0) | ||
| throw new IllegalArgumentException("Illegal new producer id " + producerId); |
| if (data.state.validPreviousStates().contains(this.state)) { | ||
| TxnTransitMetadata transitMetadata = new TxnTransitMetadata( | ||
| data.producerId, this.producerId, data.nextProducerId, data.producerEpoch, data.lastProducerEpoch, | ||
| data.txnTimeoutMs, data.state, new HashSet<>(data.topicPartitions), |
There was a problem hiding this comment.
I have another comment related to this immutability. If you intend to keep TxnTransitMetadata#topicPartitions mutable, then line#135 should use a HashSet instead of Set.copyOf, since the latter creates a immutable set
| " completing transaction state transition while it does not have a pending state"); | ||
| }); | ||
|
|
||
| if (!toState.equals(transitMetadata.txnState())) { |
There was a problem hiding this comment.
if (!toState.equals(transitMetadata.txnState())) throwStateTransitionFailure(transitMetadata);
switch (toState) {| // is created for the first time and it could stay like this until transitioning | ||
| // to Dead. | ||
| if (data.state != TransactionState.DEAD && data.producerEpoch < 0) | ||
| throw new IllegalArgumentException("Illegal new producer epoch " + producerEpoch); |
There was a problem hiding this comment.
this one is also data.producerEpoch?
| ); | ||
| } | ||
|
|
||
| private class TransitionData { |
There was a problem hiding this comment.
Can we have some javadoc here explaining how this class is used?
|
Thanks for the PR! |
Signed-off-by: PoAn Yang <payang@apache.org>
Signed-off-by: PoAn Yang <payang@apache.org>
chia7712
left a comment
There was a problem hiding this comment.
@FrankYang0529 thanks for updates.
| TransactionState txnState, | ||
| Set<TopicPartition> topicPartitions, | ||
| // The TransactionMetadata#topicPartitions field is mutable. | ||
| // To avoid deepcopy when assigning value from TxnTransitMetadata to TransactionMetadata, use HashSet here. |
| if (data.state.validPreviousStates().contains(this.state)) { | ||
| TxnTransitMetadata transitMetadata = new TxnTransitMetadata( | ||
| data.producerId, this.producerId, data.nextProducerId, data.producerEpoch, data.lastProducerEpoch, | ||
| data.txnTimeoutMs, data.state, new HashSet<>(data.topicPartitions), |
There was a problem hiding this comment.
are we not sure if this set should be mutable or not?
We discussed whether TxnTransitMetadata should hold a reference to topicPartitions or perform a deep copy. I prefer to keep the original behavior of holding the reference
val transitMetadata = new TxnTransitMetadata(producerId, this.producerId, nextProducerId, producerEpoch, lastProducerEpoch, txnTimeoutMs, state,
// no deep copy
topicPartitions.asJava, txnStartTimestamp, txnLastUpdateTimestamp, clientTransactionVersion)@FrankYang0529 WDYT?
Signed-off-by: PoAn Yang <payang@apache.org>
|
I open https://issues.apache.org/jira/browse/KAFKA-19609 to move the related tests to |
| this.lastProducerEpoch = lastProducerEpoch; | ||
| this.txnTimeoutMs = txnTimeoutMs; | ||
| this.state = state; | ||
| this.topicPartitions = new HashSet<>(topicPartitions); |
There was a problem hiding this comment.
Since topicPartitions is always empty in production when creating TransactionMetadata, we could remove it from the constructor to simplify the code
| import java.util.HashSet; | ||
|
|
||
| /** | ||
| * Immutable object representing the target transition of the transaction metadata |
There was a problem hiding this comment.
Immutable is no longer correct. Could you please update it in the follow-up?
…tionMetadata with non-empty topic partitions (#20370) This is followup PR for #19699. * Update TransactionLog#readTxnRecordValue to initialize TransactionMetadata with non-empty topic partitions * Update `TxnTransitMetadata` comment, because it's not immutable. Reviewers: TengYao Chi <kitingiao@gmail.com>, Justine Olshan <jolshan@confluent.io>, Kuan-Po Tseng <brandboat@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
…tionMetadata with non-empty topic partitions (apache#20370) This is followup PR for apache#19699. * Update TransactionLog#readTxnRecordValue to initialize TransactionMetadata with non-empty topic partitions * Update `TxnTransitMetadata` comment, because it's not immutable. Reviewers: TengYao Chi <kitingiao@gmail.com>, Justine Olshan <jolshan@confluent.io>, Kuan-Po Tseng <brandboat@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
topicPartitionsfield usesHashSetinstead ofSet, becauseit's mutable field.
prepare*methods, they can use current valueas default input in
prepareTransitionTo. However, in Java, it doesn'tsupport function default input value. To avoid a lot of duplicated code
or assign value to wrong field, we add a private class
TransitionData.It can get current
TransactionMetadatavalue as default value andprepare*methods just need to assign updated value.Reviewers: Justine Olshan jolshan@confluent.io, Artem Livshits
alivshits@confluent.io, Chia-Ping Tsai chia7712@gmail.com