You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jolshan (via GitHub)" <gi...@apache.org> on 2024/12/13 20:10:31 UTC
[PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
jolshan opened a new pull request, #18176:
URL: https://github.com/apache/kafka/pull/18176
We want to bump the epoch if we are upgrading to TV2. Given that we already have code in place for this, I thought we could piggyback on the completing transaction epoch bump logic. For just initializing producers, I moved the check to the end of InitTransaction. Note, we have do do this check after we initialize the producer ID to ensure we have updated ApiVersions correctly.
Will add tests to recreate some of the scenarios that caused us to find this bug.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1888849517
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -212,8 +214,10 @@ private boolean isTransitionValid(State source, State target) {
return source == READY || source == ABORTABLE_ERROR;
case INITIALIZING:
return source == UNINITIALIZED || source == ABORTING_TRANSACTION;
+ case UPGRADING:
Review Comment:
We could do that but I didn't know if it could be unsafe if this transition somehow happened in a non-V2 upgrade case. I did consider this approach.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1887281564
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
What is the reason for this condition? I may miss some cases but maybeUpdateTransactionV2Enabled is only called in initialization and commit transaction, none of which set the state to be Ready? When will we have `currentState == State.READY`
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2549853223
Tests are still green with the newest changes (minus the very last commit -- which didn't change functionality) : https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2024-12-17--001.b1b2f4a4-2c8e-4595-8e4a-a3b0429d23d9--1734476806--jolshan--kafka-17705-test--7cd411a93c/report.html
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1888853548
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
+ maybeUpdateTransactionV2Enabled();
Review Comment:
> Also transitioning the version here also means that we enable V2 before we fully finish transaction, could you add a comment to the EndTxnHandler.handleResponse that we can receive a V1 response even though we've transitioned into a V2 mode.
My understanding is that we do this after we add the EndTxn to the handler. My understanding is that the EndTxn request should always complete before the epoch bump is done (otherwise we would fence ourselves in the epoch bump case) so I thought this was safe, but I can think about this again.
> Also should we remove it from the beginTransaction? If we somehow call it from there without going through this path, the upgrade flow won't trigger.
I have removed it from beginTransaction in this PR already.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1887298239
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
On the first initialization we actually "upgrade to v2" because the default is false 😅. We don't want the logic there though. Initialization does set the state to ready right before this is called.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890873113
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
Review Comment:
ah, thanks for the correction, seems like i was thinking about the opposite ordering of events. makes sense
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2542530490
> But just for my understanding, how would the next transaction use TV1? We wait until the result of beginCompletingTransaction before we return.
If the TV2 is ready to use during the current txn, before the endTxn request is sent, then the next txn uses TV2. If the TV2 is ready after the current txn's endTxn and before the next Txn, then the next txn is TV1.
Yeah, I agree current approach is simpler than modifying the beginTransaction. I think it is ok as the compatibility is corrected with the current approach.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1887732290
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
Review Comment:
May be name it `wasTransactionV2Enabled`?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
+ maybeUpdateTransactionV2Enabled();
Review Comment:
Also should we remove it from the beginTransaction? If we somehow call it from there without going through this path, the upgrade flow won't trigger.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
+ maybeUpdateTransactionV2Enabled();
Review Comment:
While unlikely, the transaction can actually complete (it's running concurrently with this code) and set state to `READY`, then the `isUpgradingToV2` logic won't get triggered.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
+ maybeUpdateTransactionV2Enabled();
Review Comment:
Also transitioning the version here also means that we enable V2 before we fully finish transaction, could you add a comment to the EndTxnHandler.handleResponse that we can receive a V1 response even though we've transitioned into a V2 mode.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -212,8 +214,10 @@ private boolean isTransitionValid(State source, State target) {
return source == READY || source == ABORTABLE_ERROR;
case INITIALIZING:
return source == UNINITIALIZED || source == ABORTING_TRANSACTION;
+ case UPGRADING:
Review Comment:
Do we need a separate state? It looks like if we just updated the previous line to allow to transition from `COMMITTING_TRANSACTION` to `INITIALIZING` we could just use the `clientSideEpochBumpRequired` to drive the bump logic.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
I would just pass an argument to distinguish init vs. upgrade (see my other comment about race condition).
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2550009084
There is some issue with `testBumpTransactionalEpochWithTV2Disabled` I will take a look
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1889086622
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -212,8 +214,10 @@ private boolean isTransitionValid(State source, State target) {
return source == READY || source == ABORTABLE_ERROR;
case INITIALIZING:
return source == UNINITIALIZED || source == ABORTING_TRANSACTION;
+ case UPGRADING:
Review Comment:
It should be safe to initialize after commit -- we'd just do an extra epoch bump (and it's in a case if we have a bug that's unknown to us). The thing is that the v1 code path will become mostly dead code in 6 months or so (new client + old broker is a rare combination, much less frequent than old client + new broker) and the value of making it more precise will rapidly diminish, but the complexity will stay.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
Review Comment:
The order of txn requests is enforced by the queue -- endTxn will happen before InitPid because we queue them here. The race condition happens is because at this point the endTxn is executing concurrently, it may or may not finish by the time this line runs by this thread, and therefore it may or may not observe the effects of this line. You can add a sleep before this line and observe the race condition where endTxn finishes before maybeUpdateTransactionV2Enabled is called.
A proper way to do it, is to call this function before enqueuing the request (i.e. move to line 353). Plus add a comment why it's there an not here.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "artemlivshits (via GitHub)" <gi...@apache.org>.
artemlivshits commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1889196691
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +441,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
- public synchronized void maybeUpdateTransactionV2Enabled() {
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
Review Comment:
Stale comment.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -348,10 +347,15 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
isTransactionV2Enabled
);
+ // Maybe update the transaction version here before we enqueue the EndTxn request so there are no races with
Review Comment:
We should probably clarify that this function may set the `clientSideEpochBumpRequired` on upgrade and that's the reason we do it here, before we call logic that checks `clientSideEpochBumpRequired`. But not before we build the EndTxnRequest, because we need to complete the transaction with the version it started.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +441,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
- public synchronized void maybeUpdateTransactionV2Enabled() {
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
+ public synchronized void maybeUpdateTransactionV2Enabled(boolean onInitiatialization) {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean wasTransactionV2Enabled = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ clientSideEpochBumpRequired = !onInitiatialization && !wasTransactionV2Enabled && isTransactionV2Enabled;
Review Comment:
Probably here or maybe at the function definition we should add a comment that explains why we need epoch bump on upgrade.
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -211,7 +211,7 @@ private boolean isTransitionValid(State source, State target) {
case UNINITIALIZED:
return source == READY || source == ABORTABLE_ERROR;
case INITIALIZING:
- return source == UNINITIALIZED || source == ABORTING_TRANSACTION;
+ return source == UNINITIALIZED || source == ABORTING_TRANSACTION || source == COMMITTING_TRANSACTION;
Review Comment:
Super minor nit: the case below has `COMMITTING_TRANSACTION` check before the `ABORTING_TRANSACTION` check.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2546167842
New system test run is also green: http://ducktape-open-source-results.confluent.io.s3-website-us-west-2.amazonaws.com/confluent-open-source-kafka-branch-builder-system-test-results/?prefix=trunk/2024-12-14--001.aa458635-1231-4d5d-81ec-6d2951845439--1734203544--jolshan--kafka-17705-test--933d8d8072
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890866056
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
Review Comment:
The race condition involved the previous "state != READY" check that is now removed from the PR. But the idea was that in the race, the EndTxn request would queue and complete before we called the method to maybe upgrade to TV2. In the case that the transaction completes first, we would transition to READY (rather than committing/aborting transaction) and we wouldn't trigger the "upgrade to TV2" (now just `clientSideEpochBumpRequired`) condition.
Changing the method to not check the state and check on the method parameter eliminated that part of issue, but we also still have the general transition to READY we want to be careful about since we can't easily bump epoch if we are in READY state.
Right now, since we piggyback on `clientSideEpochBumpRequired` we will always transition to INITIALIZING after completing the transaction since we set that value BEFORE enqueueing the EndTxn request. See https://github.com/apache/kafka/blob/0055ef0a49ca09a350b8b99b0b758d88a9b6555c/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L1244.
TLDR; the issue was not starting a new transaction but ensuring we correctly identify the TV2 upgrade case and bumping the epoch (which could be missed in a race due to idiosyncrasies in the state machine).
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2552508794
Failing tests are known flakes and unrelated to this PR. I will go ahead and merge + pick to 4.0 branch.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1887298239
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
On the first initialization we actually "upgrade to v2" because the default is false 😅. We don't want the logic there though.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890873828
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +443,25 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
- public synchronized void maybeUpdateTransactionV2Enabled() {
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets clientSideEpochBumpRequired if upgrading to V2 since we need to bump the epoch.
+ * This is because V2 no longer adds partitions explicitly and there are some edge cases on upgrade
+ * that can be avoided by fencing the old V1 transaction epoch. For example, we won't consider
+ * partitions from the previous transaction as already added to the new V2 transaction if the epoch is fenced.
+ */
+
+ public synchronized void maybeUpdateTransactionV2Enabled(boolean onInitiatialization) {
Review Comment:
This is mostly right, but want to clarify a few subtle things.
We want to handle cases where v1 is completing a transaction while upgrading to v2.
In this scenario, since the V1 transaction has the same epoch as the incoming v2 transaction, there are edge cases where we could miss adding partitions to the new V2 transaction while the v1 transaction was still completing. In the case where the partitions were exactly the same (or a subset) we fence the producer on EndTxn.
In another scenario, where we don't send an EndTxn request (say the producer dies) we would have transactional records on partitions that the coordinator doesn't know about (hanging transaction 😱 )
Both scenarios can be avoided by bumping the epoch after v1 commits. Specifically, the producer will get its new bumped epoch before it can even begin the v2 transaction. With the new epoch, it can differentiate between the partitions it added in the v1 transaction and the ones it will need to add for the v2 transaction.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890875348
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -947,18 +947,20 @@ public void testTransactionManagerEnablesV2() {
TransactionalRequestResult retryResult = transactionManager.beginCommit();
assertTrue(transactionManager.hasOngoingTransaction());
- assertFalse(transactionManager.isTransactionV2Enabled());
+ assertTrue(transactionManager.isTransactionV2Enabled());
prepareEndTxnResponse(Errors.NONE, TransactionResult.COMMIT, producerId, epoch);
Review Comment:
Yup exactly. The v1 transaction will have its original epoch (not bumped) and the next InitProducerId response will have the new epoch as expected.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2542514704
I will take a look at the TransactionManagerTest failures. 👍
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2542512696
I was trying to avoid sending an api in beginTransaction since it doesn't return a result.
But just for my understanding, how would the next transaction use TV1? We wait until the result of beginCompletingTransaction before we return. We also use this approach for the epoch bump and that needs to ensure the producer gets the new epoch before starting the next transaction right?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1887281564
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
What is the reason for this condition? I may miss some cases but maybeUpdateTransactionV2Enabled is only called in initialization and commit transaction, none of which set the state to be Ready? When we will have `currentState == State.READY`
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890711212
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +443,25 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
- public synchronized void maybeUpdateTransactionV2Enabled() {
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
Review Comment:
nit: "to verify whether the.."
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -947,18 +947,20 @@ public void testTransactionManagerEnablesV2() {
TransactionalRequestResult retryResult = transactionManager.beginCommit();
assertTrue(transactionManager.hasOngoingTransaction());
- assertFalse(transactionManager.isTransactionV2Enabled());
+ assertTrue(transactionManager.isTransactionV2Enabled());
prepareEndTxnResponse(Errors.NONE, TransactionResult.COMMIT, producerId, epoch);
Review Comment:
this ensures that the epoch of when the begin complete transaction started was used right?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
Review Comment:
To understand the impact of the race condition:
First, we want to complete the transaction. however, if the end txn completes first then we may additionally start a new transaction here. is this correct?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +443,25 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
- public synchronized void maybeUpdateTransactionV2Enabled() {
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets clientSideEpochBumpRequired if upgrading to V2 since we need to bump the epoch.
+ * This is because V2 no longer adds partitions explicitly and there are some edge cases on upgrade
+ * that can be avoided by fencing the old V1 transaction epoch. For example, we won't consider
+ * partitions from the previous transaction as already added to the new V2 transaction if the epoch is fenced.
+ */
+
+ public synchronized void maybeUpdateTransactionV2Enabled(boolean onInitiatialization) {
Review Comment:
to understand the main ideas:
- we want to handle cases when v1 producer is completing a transaction + while upgrading to v2
- for the above case, we need to bump the epoch. otherwise, the next end txn will be fenced (v2 same epoch)
please correct me if i'm missing anything
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #18176:
URL: https://github.com/apache/kafka/pull/18176
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1888854431
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -437,15 +442,21 @@ public boolean isTransactional() {
return transactionalId != null;
}
- // Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ /**
+ * Check all the finalized features from apiVersions to whether the transaction V2 is enabled.
+ * Sets isUpgradingToV2 if the previous value was false and now it is true.
+ */
+
public synchronized void maybeUpdateTransactionV2Enabled() {
if (latestFinalizedFeaturesEpoch >= apiVersions.getMaxFinalizedFeaturesEpoch()) {
return;
}
ApiVersions.FinalizedFeaturesInfo info = apiVersions.getFinalizedFeaturesInfo();
latestFinalizedFeaturesEpoch = info.finalizedFeaturesEpoch;
Short transactionVersion = info.finalizedFeatures.get("transaction.version");
+ boolean previousValue = isTransactionV2Enabled;
isTransactionV2Enabled = transactionVersion != null && transactionVersion >= 2;
+ isUpgradingToV2 = currentState != State.READY && !previousValue && isTransactionV2Enabled;
Review Comment:
That makes sense 👍 will change that.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18176:
URL: https://github.com/apache/kafka/pull/18176#discussion_r1890866056
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -352,7 +355,9 @@ private TransactionalRequestResult beginCompletingTransaction(TransactionResult
enqueueRequest(handler);
// If an epoch bump is required for recovery, initialize the transaction after completing the EndTxn request.
- if (clientSideEpochBumpRequired) {
+ // If we are upgrading to TV2 transactions on the next transaction, also bump the epoch.
Review Comment:
The race condition involved the previous "ready" check that is now removed from the PR. But the idea was that in the race, the EndTxn request would queue and complete before we called the method to maybe upgrade to TV2. In the case that the transaction completes first, we would transition to READY (rather than committing/aborting transaction) and we wouldn't trigger the "upgrade to TV2" (now just `clientSideEpochBumpRequired`) condition.
Changing the method to not check on ready and check on the parameter eliminated that part of issue, but we also still have the general transition to READY we want to be careful about since we can't easily bump epoch if we are in READY state.
Right now, since we piggyback on `clientSideEpochBumpRequired` we will always transition to INITIALIZING after completing the transaction since we set that value BEFORE enqueueing the EndTxn request. See https://github.com/apache/kafka/blob/0055ef0a49ca09a350b8b99b0b758d88a9b6555c/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L1244.
TLDR; the issue was not starting a new transaction but ensuring we correctly identify the TV2 upgrade case and bumping the epoch (which could be missed in a race due to idiosyncrasies in the state machine).
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18227: Ensure v2 partitions are not added to last transaction during upgrade [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on PR #18176:
URL: https://github.com/apache/kafka/pull/18176#issuecomment-2542485192
Thanks for the change.
It may cause a delay if TV2 is enabled after the previous txn commit. The immediate next txn will still use TV1.
I wonder if we can modify `beginTransaction` to init producer id instead.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org