You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrancisGodinho (via GitHub)" <gi...@apache.org> on 2025/12/16 03:27:24 UTC
[PR] KAFKA-19925: Fix transaction timeout handling during broker upgrades [kafka]
FrancisGodinho opened a new pull request, #21161:
URL: https://github.com/apache/kafka/pull/21161
# Problem
During broker upgrades, the `sendOffsetsToTransaction` call would sometimes hang. Logs showed that it continuously returned `errorCode=51` which is `CONCURRENT_TRANSACTION`. The test would eventually hit its timeout and fail. This happened for every single version upgrade and occurred in around 30% of the runs.
# Resolution
The problem above left the producer in a broken state and even after 5-10 minutes of waiting, it didn't resolve itself (even if we waited a few minutes past the transaction.max.ms time). I tried multiple solutions including waiting extended periods of time and re-trying the `sendOffsetsToTransaction` multiple times whenever timeout occurred.
Unfortunately, the producer was just permanently stuck and always receiving the `errorCode=51`. In this case, the recommended resolution in the Kafka docs is to close the previous producer and create a new producer. https://kafka.apache.org/documentation/#usingtransactions
<img width="652" height="59" alt="image" src="https://github.com/user-attachments/assets/e95500d6-f1b6-44fa-b6a2-5c1800448d32" />
Using the old transaction.id would continue to lead to a stuck state, so this fix creates a brand new producer with a new ID and then rewinds the consumer offset to ensure EOD.
# Testing and Validation
Previously, I was able to run the test for a single version upgrade and have it fail within the first 5-10 runs. After the fix, I was able to run it 40 times continuously with 0 failures. I also ran the full test (all versions) ~5 times with 9/9 cases passing.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2658683685
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -387,7 +389,17 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
long messagesSentWithinCurrentTxn = records.count();
ConsumerGroupMetadata groupMetadata = useGroupMetadata ? consumer.groupMetadata() : new ConsumerGroupMetadata(consumerGroup);
- producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ try {
+ producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ } catch (KafkaException e) {
Review Comment:
IMHO, shouldn't we focus on handling the `TimeoutException`?
AFAIK, `CONCURRENT_TRANSACTIONS` eventually manifests as a `TimeoutException` on the client side. I'm concerned that broad scope retries might mask other underlying errors. In those cases, we need to clearly identify the root cause to take appropriate action.
What do you think?
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -387,7 +389,17 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
long messagesSentWithinCurrentTxn = records.count();
ConsumerGroupMetadata groupMetadata = useGroupMetadata ? consumer.groupMetadata() : new ConsumerGroupMetadata(consumerGroup);
- producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ try {
+ producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ } catch (KafkaException e) {
+ // in case the producer gets stuck here, create a new one and continue the loop
+ try { producer.close(Duration.ofSeconds(0)); } catch (Exception ignore) {}
+ parsedArgs.getAttrs().put("transactionalId", parsedArgs.getString("transactionalId") + producerNumber++);
Review Comment:
Is there a safer way to generate a globally unique transactionalId to avoid collisions?
Simply appending an incremental number `(producerNumber++)` to the user-provided ID seems risky in a shared cluster environment. If the generated ID happens to match an existing transactionalId of a running production application, it could trigger the fencing mechanism, unintentionally aborting the active transactions of that application.
Perhaps appending a UUID or a random suffix would be a safer approach to ensure uniqueness?
What do you think?
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #21161: KAFKA-19925: Fix transaction timeout handling during broker upgrades
URL: https://github.com/apache/kafka/pull/21161
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "FrancisGodinho (via GitHub)" <gi...@apache.org>.
FrancisGodinho commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3658569174
@chia7712 can you take a look when you get a chance please?
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "FrancisGodinho (via GitHub)" <gi...@apache.org>.
FrancisGodinho commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2629863889
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -308,9 +308,11 @@ public static void runEventLoop(Namespace parsedArgs) {
String consumerGroup = parsedArgs.getString("consumerGroup");
- final KafkaProducer<String, String> producer = createProducer(parsedArgs);
+ KafkaProducer<String, String> producer = createProducer(parsedArgs);
final KafkaConsumer<String, String> consumer = createConsumer(parsedArgs);
+ int producerNumber = 0;
Review Comment:
this doesn't work for local variables, only class variables. get this error
```
error: variable producerNumber might not have been initialized parsedArgs.getAttrs().put("transactionalId", parsedArgs.getString("transactionalId") + producerNumber++);
```
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-4186302332
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "FrancisGodinho (via GitHub)" <gi...@apache.org>.
FrancisGodinho commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3668732814
@Pankraz76 thanks for the comments, can you re-review please?
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3706785858
@FrancisGodinho , thanks for the update!
Following @chia7712 ’s suggestion to try the server-side approach (retrying internally on the broker side), I wanted to check on ownership for the follow-up work.
Are you planning to implement the server-side fix as a separate PR? If so, I’m happy to help with reviews/testing. If you’d prefer to focus on the current client-side PR, I can pick up the follow-up server-side PR (or co-author it) to help move KAFKA-20000 forward.
Totally up to you!
I just wanted to express my willingness to help if needed.
Let me know what works best for you!
CC. @chia7712
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2630540388
##########
tests/kafkatest/tests/core/transactions_upgrade_test.py:
##########
@@ -179,7 +179,7 @@ def copy_messages_transactionally_during_upgrade(self, input_topic, output_topic
self.perform_upgrade(from_kafka_version)
- copier_timeout_sec = 180
+ copier_timeout_sec = 360
Review Comment:
As I described in https://issues.apache.org/jira/browse/KAFKA-20000, the performance regression is caused by the backoff logic. Therefore, I suggest fixing the underlying issue instead of increasing the timeout.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3662121409
@FrancisGodinho thanks for you patch. I have identified some underlying issues in e2e and TV2. Addressing them should allow us to achieve more stable transaction behavior. Please check https://issues.apache.org/jira/browse/KAFKA-19999 and https://issues.apache.org/jira/browse/KAFKA-20000 for more details.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "Pankraz76 (via GitHub)" <gi...@apache.org>.
Pankraz76 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2628487286
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -402,7 +414,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
} catch (KafkaException e) {
log.debug("Aborting transaction after catching exception", e);
abortTransactionAndResetPosition(producer, consumer);
- }
+ }
Review Comment:
- https://github.com/junit-team/junit-framework/blob/575f741c2c73dd4242b29b8d4642fda9dfdd8cce/gradle/plugins/common/src/main/kotlin/junitbuild.spotless-conventions.gradle.kts#L39
- https://docs.openrewrite.org/recipes/java/format/removetrailingwhitespace
- https://github.com/apache/kafka/pull/21168
kindly related example to improve setup, streamline development convention.
```suggestion
}
```
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -308,9 +308,11 @@ public static void runEventLoop(Namespace parsedArgs) {
String consumerGroup = parsedArgs.getString("consumerGroup");
- final KafkaProducer<String, String> producer = createProducer(parsedArgs);
+ KafkaProducer<String, String> producer = createProducer(parsedArgs);
final KafkaConsumer<String, String> consumer = createConsumer(parsedArgs);
+ int producerNumber = 0;
Review Comment:
```suggestion
int producerNumber;
```
imho, its kind of the same right? idk which is the default, but again just offtopic im sorry.
This could be streamlined as well by:
https://docs.openrewrite.org/recipes/staticanalysis/commonstaticanalysis
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "Pankraz76 (via GitHub)" <gi...@apache.org>.
Pankraz76 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2630224707
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -387,7 +389,17 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
long messagesSentWithinCurrentTxn = records.count();
ConsumerGroupMetadata groupMetadata = useGroupMetadata ? consumer.groupMetadata() : new ConsumerGroupMetadata(consumerGroup);
- producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ try {
+ producer.sendOffsetsToTransaction(consumerPositions(consumer), groupMetadata);
+ } catch (KafkaException e) {
+ // in case the producer gets stuck here, create a new one and continue the loop
+ try { producer.close(Duration.ofSeconds(0)); } catch (Exception ignore) {}
+ parsedArgs.getAttrs().put("transactionalId", parsedArgs.getString("transactionalId") + producerNumber++);
+ producer = createProducer(parsedArgs);
+ producer.initTransactions();
+ resetToLastCommittedPositions(consumer);
Review Comment:
```suggestion
circutBreaker(); // in case the producer gets stuck here, create a new one and continue the loop
```
could give dedicated to this concern apply single responsibility principle, giving more focus to each own. Here its just about breaking the circut, how this is actually done seems to be some kind of (randomly) changing impl. detail.
##########
tests/kafkatest/tests/core/transactions_upgrade_test.py:
##########
@@ -179,7 +179,7 @@ def copy_messages_transactionally_during_upgrade(self, input_topic, output_topic
self.perform_upgrade(from_kafka_version)
- copier_timeout_sec = 180
+ copier_timeout_sec = 360
Review Comment:
```suggestion
copier_timeout_sec = 360
```
sorry again this something for SCA. Taking away the off-topics upfront.
spotless and rewrite both ready to fix on their own.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "Pankraz76 (via GitHub)" <gi...@apache.org>.
Pankraz76 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2630210388
##########
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java:
##########
@@ -308,9 +308,11 @@ public static void runEventLoop(Namespace parsedArgs) {
String consumerGroup = parsedArgs.getString("consumerGroup");
- final KafkaProducer<String, String> producer = createProducer(parsedArgs);
+ KafkaProducer<String, String> producer = createProducer(parsedArgs);
final KafkaConsumer<String, String> consumer = createConsumer(parsedArgs);
+ int producerNumber = 0;
Review Comment:
ok thank you, my bad.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "FrancisGodinho (via GitHub)" <gi...@apache.org>.
FrancisGodinho commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2621619245
##########
tests/kafkatest/tests/core/transactions_upgrade_test.py:
##########
@@ -179,7 +179,7 @@ def copy_messages_transactionally_during_upgrade(self, input_topic, output_topic
self.perform_upgrade(from_kafka_version)
- copier_timeout_sec = 180
+ copier_timeout_sec = 360
Review Comment:
Note: due to timeouts and re-creation of producer, this copier_timeout needed to be increased. I experimented a bit and found that 360s was a consistently reliable value.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "Pankraz76 (via GitHub)" <gi...@apache.org>.
Pankraz76 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2630614263
##########
tests/kafkatest/tests/core/transactions_upgrade_test.py:
##########
@@ -179,7 +179,7 @@ def copy_messages_transactionally_during_upgrade(self, input_topic, output_topic
self.perform_upgrade(from_kafka_version)
- copier_timeout_sec = 180
+ copier_timeout_sec = 360
Review Comment:
kindly asking, if this is something to consider? If so, would add some test for this adjustment.
Thanks.
- https://github.com/apache/kafka/pull/21172
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #21161:
URL: https://github.com/apache/kafka/pull/21161#discussion_r2630718124
##########
tests/kafkatest/tests/core/transactions_upgrade_test.py:
##########
@@ -179,7 +179,7 @@ def copy_messages_transactionally_during_upgrade(self, input_topic, output_topic
self.perform_upgrade(from_kafka_version)
- copier_timeout_sec = 180
+ copier_timeout_sec = 360
Review Comment:
@Pankraz76 thanks for the effort. As Justine suggested, hardcoding the timeout is a bit coarse-grained. Please refer to KAFKA-20000 for more discussion.
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3706686351
> Also, it seems that the cause of TimeoutException via TV2 is resolved by this PR
Agreed, the root cause is the livelock, but fixing the retry is still valuable. +1 to trying the server-side approach. Let’s keeping this PR and handle the server-side fix in a separate PR
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "FrancisGodinho (via GitHub)" <gi...@apache.org>.
FrancisGodinho commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-3706817700
@chickenchickenlove @chia7712 yeah I think retrying on the server side would be better as well since less logic for the client. I'd like to take a stab at it since I will have some time this week, but I'll let you know if I need help/guidance (thanks for offering!)
I'll implement the server side changes as a separate PR (since it's also a separate ticket) and ping you once I have some progress made!
--
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-19925: Fix transaction timeout handling during broker upgrades [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21161:
URL: https://github.com/apache/kafka/pull/21161#issuecomment-4368308467
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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