You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kirktrue (via GitHub)" <gi...@apache.org> on 2024/08/27 22:35:07 UTC
[PR] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
kirktrue opened a new pull request, #17022:
URL: https://github.com/apache/kafka/pull/17022
WIP
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2051350718
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -772,6 +779,14 @@ synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException except
}
}
+ /**
+ * Returns {@code true} if the given {@link ProducerBatch} has the same producer ID but a different epoch than the
+ * {@link #producerIdAndEpoch cached producer ID and epoch}.
+ */
+ synchronized boolean isStaleBatch(ProducerBatch batch) {
Review Comment:
I inlined the 'is stale batch' logic into `handleFailedBatch()`, so these issues are no longer present.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2133073450
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Here are the places I found in which a `ProducerIdAndEpoch` is compared:
* [`TransactionManager.setProducerIdAndEpoch()`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L604-L615) checks the `producerId`, but it appears to only affect logging, though.
* [`TransactionManager.maybeUpdateProducerIdAndEpoch()`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L589-L602) calls to `hasStaleProducerIdAndEpoch()` to compare its current `ProducerIdAndEpoch` with the one in its `txnPartitionMap`. In the case we're seeing, the producer ID in the `ProducerBatch` is out of sync. I don't know if the `txnPartitionMap` is also out of sync in that case.
* [`ProducerBatch.resetProducerState()`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java#L476-L481) seems interesting to consider in that maybe it could be called out of sync with the transaction manager? That method is called by [`TxnPartitionEntry.adjustSequencesDueToFailedBatch()`](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/TxnPartitionEntry.java#L139-L152), but resets the batch with the same `ProducerIdAndEpoch` from the batch. Should it be consulting the `TransactionManager` for the _current_ 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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2755596812
cc @k-raina
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2594409211
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2138442177
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
@jolshan—There are a number of tests that fail with the latest change:
* `TransactionManagerTest.testTransitionToAbortableErrorOnBatchExpiry()`
* `TransactionManagerTest.testTransitionToAbortableErrorOnMultipleBatchExpiry()`
* `TransactionManagerTest.testBackgroundInvalidStateTransitionIsFatal()`
* `TransactionsTest.testBumpTransactionalEpochWithTV2Disabled()`
* `TransactionsWithTieredStoreTest.testBumpTransactionalEpochWithTV2Disabled()`
From the perspective of those tests, the state should be set to `FATAL_ERROR` in the case where the batch’s producer ID is different from the transaction manager’s.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2152880201
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -790,14 +790,34 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale. This means this batch will not cause a change to either the
+ // state or the sequence numbers; the stale batch is only removed from the inflight set.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ // It's possible the transaction manager is already in the FATAL_ERROR state at this point. Depending on the
+ // incoming exception type, maybeTransitionToErrorState() could attempt to set the state to ABORTABLE_ERROR.
+ // For example, suppose a fatal error occurred during a transaction, and then moments later one of the batches
+ // in that transaction failed with a TimeoutException. maybeTransitionToErrorState() would then (blindly)
+ // attempt to transition to ABORTABLE_ERROR, which is invalid and would result in an IllegalStateException.
+ //
+ // Therefore, only attempt to transition to the FATAL_ERROR state if the batch is "fresh" *and* the
+ // transaction manager is not already in the FATAL_ERROR state.
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
+
removeInFlightBatch(batch);
if (hasFatalError()) {
log.debug("Ignoring batch {} with producer id {}, epoch {}, and sequence number {} " +
"since the producer is already in fatal error state", batch, batch.producerId(),
batch.producerEpoch(), batch.baseSequence(), exception);
return;
+ } else if (isStaleBatch) {
+ log.debug("Ignoring stale batch {} with producer id {}, epoch {}, and sequence number {} " +
Review Comment:
Work for me to leave a DEBUG -- was just wondering -- if it's a volume problem (what I don't expect), people could disable DEBUG for this class until we fix it.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue closed pull request #17022: KAFKA-14830: Illegal state error in transactional producer
URL: https://github.com/apache/kafka/pull/17022
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2051351665
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,21 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ if (!isStaleBatch(batch) && !hasFatalError())
Review Comment:
> I will add another unit test or two to make sure this is a valid concern.
Turns out I already had a unit test for that: `testBatchesReceivedAfterFatalError()`. If I remove the `!hasFatalError()` condition that test fails:
```
TransactionalId foobar: Invalid transition attempted from state FATAL_ERROR to state ABORTABLE_ERROR
java.lang.IllegalStateException: TransactionalId foobar: Invalid transition attempted from state FATAL_ERROR to state ABORTABLE_ERROR
at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:1082)
at org.apache.kafka.clients.producer.internals.TransactionManager.transitionToAbortableError(TransactionManager.java:498)
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeTransitionToErrorState(TransactionManager.java:735)
at org.apache.kafka.clients.producer.internals.TransactionManager.handleFailedBatch(TransactionManager.java:745)
at org.apache.kafka.clients.producer.internals.TransactionManagerTest.testBatchesReceivedAfterFatalError(TransactionManagerTest.java:3987)
at java.base/java.lang.reflect.Method.invoke(Method.java:569)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
```
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2116253121
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
When we have epoch overflow, we get a new producer id and start epoch at 0.
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2767367218
> I think the overall approach makes sense. But I would like to see some tests to see if the issue is improved. If so the logging could also give us some more insight.
The unit tests mimic the use cases that were seen in the wild. What other test cases should we consider? Thanks!
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2133083297
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
I think that means it is safe to mark as stale if the producer ID doesn't match
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r1735219780
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -667,14 +667,23 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() < producerIdAndEpoch.epoch;
Review Comment:
I'm wondering if there are any cases where producerIdAndEpoch could have a race -- or is case there the ID and epoch are the same but the issue still happens
btw -- maybe not super common, but could the overflow case be triggered here? (new producer id and epoch resets due to epoch reaching max 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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r1735219780
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -667,14 +667,23 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() < producerIdAndEpoch.epoch;
Review Comment:
I'm wondering if there are any cases where producerIdAndEpoch could have a race -- or is case there the ID and epoch are the same but the issue still happens
btw -- maybe not super common, but could the overflow case be missed here? (new producer id and epoch resets due to epoch reaching max 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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2112903427
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
one minor thing here, if we have an epoch overflow, we won't consider this check. Maybe if the producer ID doesn't match, we also ignore it? Not sure if that has unintended consequences. We could also tackle as a followup, since without this fix, it's the same state as today.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2981092997
Hey thanks. I took a step back and I'm wondering if this is the right approach. Let me try to get back to you on this -- I need to get a better idea of the whole flow. The discussions on the check made me realize I was missing something.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2978134097
@jolshan @mjsax—Another ping to check if we can merge... Thanks!
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2112901807
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
Comment seems fine to me.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2051328782
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,21 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ if (!isStaleBatch(batch) && !hasFatalError())
Review Comment:
Based on the type of the exception pass in, the logic in `maybeTransitionToErrorState()` may set the internal state to either `FATAL_ERROR` or `ABORTABLE_ERROR`. Assuming there's a race condition of failures, it's possible the transaction manager could be set to a `FATAL_ERROR` state, followed by a call to `handleFailedBatch()` that then attempts to set the state to `ABORTABLE_ERROR`. Transitioning from `FATAL_ERROR` to any other state results in an `IllegalStateException`.
This is an attempt to prevent that case. I will add another unit test or two to make sure this is a valid concern.
cc @jolshan
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2051330115
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -772,6 +779,14 @@ synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException except
}
}
+ /**
+ * Returns {@code true} if the given {@link ProducerBatch} has the same producer ID but a different epoch than the
+ * {@link #producerIdAndEpoch cached producer ID and epoch}.
+ */
+ synchronized boolean isStaleBatch(ProducerBatch batch) {
Review Comment:
> Does this method need to by `synchronized`?
No, not at present.
> Also seems it could be `private`?
Yes, it could. I might even just inline it in `handleFailedBatch()` to simplify things.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "k-raina (via GitHub)" <gi...@apache.org>.
k-raina commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2054686862
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
```
// Therefore, only attempt to transition to the FATAL_ERROR state if
// the batch is "fresh" *and* the transaction manager is not already
// in the FATAL_ERROR state.
```
FATAL_ERROR to FATAL_ERROR[ is not an invalid transition](https://github.com/k-raina/kafka/blob/ab2cae14b93c13e21eda1d142d582bb7f42d7aec/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L211-L233). Do you mean `only attempt transition to the ABORTABLE_ERROR state`
EDIT: Okay i got that check is trying to say "Do not allow any transitions if we are in fatal state"
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2596382775
Still needed, just lower priority 😞
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2316147085
I think the overall approach makes sense. But I would like to see some tests to see if the issue is improved. If so the logging could also give us some more insight.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2141308186
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Is `batch.producerEpoch() != producerIdAndEpoch.epoch` ok? Or should it be `batch.producerEpoch() < producerIdAndEpoch.epoch`?
On the other hand, we don't have a check here for batch.producerEpoch() > producerIdAndEpoch.epoch` which should never happen? Should we add such a check (or does it maybe exist somewhere else)? Or do we not care?
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2116240713
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
I think I'm missing something conceptually—when an epoch overflows, does it restart at 0, roll over to negative, or something else?
In the case of an overflow, wouldn't the batch's producer epoch value still differ from the `producerIdAndEpoch`’s 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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2122375192
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
That could work -- I guess the other thing is do we check and handle if the producer ID doesn't match elsewhere? I don't remember off the top of my head.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2981743810
Ok -- I'm back. I think the thing that wasn't sitting right for me, and what I realized from our discussion with the producer ID overflow is whether this is the right place to make the change and the right mental model.
Specifically, we don't have a great way to distinguish benign stale requests from those that that could indicate a divergence of state or other real problem.
I'm wondering if we can get to the heart of this:
> 4. A timeout threshold is hit
> 5. _T1_ starts the abort process
1. _TM_ state is set to `ABORTING_TRANSACTION`
2. The batches involved with _T1_ are marked as expired
3. _TM_ is reinitialized, bumping the epoch from `0` to `1` and
setting its state to `READY`
> 6. A moment later, in the `Sender` thread, one of the failed batches
calls `handleFailedBatch()`
Do we know if there is any way we can ensure the exact inflight requests during the timeout are marked as stale vs just assuming this from the epoch in the batch? I think for some errors, we close the connection -- we may not want to do that here, but thinking about addressing the problem at the root and not adding small changes in ways that we may not realize the side effects for yet. (As we saw in the test failures from making the change, the current part of code may cause other issues)
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2143291551
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -790,14 +790,34 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale. This means this batch will not cause a change to either the
+ // state or the sequence numbers; the stale batch is only removed from the inflight set.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ // It's possible the transaction manager is already in the FATAL_ERROR state at this point. Depending on the
+ // incoming exception type, maybeTransitionToErrorState() could attempt to set the state to ABORTABLE_ERROR.
+ // For example, suppose a fatal error occurred during a transaction, and then moments later one of the batches
+ // in that transaction failed with a TimeoutException. maybeTransitionToErrorState() would then (blindly)
+ // attempt to transition to ABORTABLE_ERROR, which is invalid and would result in an IllegalStateException.
+ //
+ // Therefore, only attempt to transition to the FATAL_ERROR state if the batch is "fresh" *and* the
+ // transaction manager is not already in the FATAL_ERROR state.
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
+
removeInFlightBatch(batch);
if (hasFatalError()) {
log.debug("Ignoring batch {} with producer id {}, epoch {}, and sequence number {} " +
"since the producer is already in fatal error state", batch, batch.producerId(),
batch.producerEpoch(), batch.baseSequence(), exception);
return;
+ } else if (isStaleBatch) {
+ log.debug("Ignoring stale batch {} with producer id {}, epoch {}, and sequence number {} " +
Review Comment:
That's a good question, @mjsax.
IMO, `TRACE` is only for developers. No organization will set logging to `TRACE` in production, at least not long enough to hit this issue and see the log. But honestly, the same is true for `DEBUG` 🤔
Can we leave at `DEBUG` and let end users tell us to turn it down?
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2112721563
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,21 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ if (!isStaleBatch(batch) && !hasFatalError())
Review Comment:
I am asserting that we need the check for `!hasFatalError()` before calling `maybeTransitionToErrorState()` because without the check we could end up with the error listed above. My assertion is based off the unit test I wrote `testBatchesReceivedAfterFatalError()`.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2116253746
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
So the producer ID will not be the same.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2138447310
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Ok. So I suppose this was being check somewhere else 😅 I suppose we can revert that.
Perhaps it is sufficient to still fail in the edge case where we have epoch overflow for now as it doesn't seem like we will make much progress on finding an alternative.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17022:
URL: https://github.com/apache/kafka/pull/17022#issuecomment-2963868532
@jolshan @mjsax—Anything else we need to check before we can merge? Thanks!
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2136374381
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
I've made the change to check the producer ID too. PTAL. Thanks!
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2052656744
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,21 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ if (!isStaleBatch(batch) && !hasFatalError())
Review Comment:
What is the question here? I think we want to avoid extra errors if we are already in fatal.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2151075013
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
I think it is possible that these stale overflow cases unfortunately may lead to having to close the producer, but I would prefer to look into that as a potential followup. Seems like it would be very rare.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "k-raina (via GitHub)" <gi...@apache.org>.
k-raina commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2054686862
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
```
// Therefore, only attempt to transition to the FATAL_ERROR state if
// the batch is "fresh" *and* the transaction manager is not already
// in the FATAL_ERROR state.
```
FATAL_ERROR to FATAL_ERROR[ is not an invalid transition](https://github.com/k-raina/kafka/blob/ab2cae14b93c13e21eda1d142d582bb7f42d7aec/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L211-L233). Do you mean `only attempt transition to the ABORTABLE_ERROR state`
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2143175834
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Receiving a future epoch would seem to imply something was really wrong.
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2143261652
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
So it sounds like `batch.producerEpoch() < producerIdAndEpoch.epoch` is really more appropriate as a "staleness" check and let the `TransactionManager` handle the other cases?
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2133081835
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Anyway, this is not quite what I meant. I meant, do we check in the failedBatch path anywhere. Doing a quick search now, it appears not.
--
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] [WIP] KAFKA-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2014904591
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -667,14 +667,23 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() < producerIdAndEpoch.epoch;
Review Comment:
Thanks for the feedback @jolshan!
> I'm wondering if there are any cases where producerIdAndEpoch could have a race -- or is case there the ID and epoch are the same but the issue still happens
There are a couple of bug reports with logs. I'll dig through those to see if it's happened in the wild.
> btw -- maybe not super common, but could the overflow case be missed here? (new producer id and epoch resets due to epoch reaching max value)
Sounds super rare ;)
If an epoch overflowed, wouldn't that just be interpreted as 'not equal' to the last known epoch, and thus trigger the "stale batch" logic? Perhaps my understanding of staleness is too naive?
Thanks!
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2112722933
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
@k-raina Did you have a suggestion for better wording? Happy to hear it 😄
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2122305073
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
So if _either_ the producer ID _or_ the epoch of the batch is different than that of the transaction manager, it's "stale?"
```suggestion
boolean isStaleBatch = batch.producerId() != producerIdAndEpoch.producerId || batch.producerEpoch() != producerIdAndEpoch.epoch;
```
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2133080898
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -779,14 +779,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
Review Comment:
Not sure I follow the last point. That method is always called from the TransactionManager's `txnPartitionMap`
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2141309397
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -790,14 +790,34 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale. This means this batch will not cause a change to either the
+ // state or the sequence numbers; the stale batch is only removed from the inflight set.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ // It's possible the transaction manager is already in the FATAL_ERROR state at this point. Depending on the
+ // incoming exception type, maybeTransitionToErrorState() could attempt to set the state to ABORTABLE_ERROR.
+ // For example, suppose a fatal error occurred during a transaction, and then moments later one of the batches
+ // in that transaction failed with a TimeoutException. maybeTransitionToErrorState() would then (blindly)
+ // attempt to transition to ABORTABLE_ERROR, which is invalid and would result in an IllegalStateException.
+ //
+ // Therefore, only attempt to transition to the FATAL_ERROR state if the batch is "fresh" *and* the
+ // transaction manager is not already in the FATAL_ERROR state.
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
+
removeInFlightBatch(batch);
if (hasFatalError()) {
log.debug("Ignoring batch {} with producer id {}, epoch {}, and sequence number {} " +
"since the producer is already in fatal error state", batch, batch.producerId(),
batch.producerEpoch(), batch.baseSequence(), exception);
return;
+ } else if (isStaleBatch) {
+ log.debug("Ignoring stale batch {} with producer id {}, epoch {}, and sequence number {} " +
Review Comment:
Just wonder if this should be TRACE? It seems not to be important enough for DEBUG (not worried about volume, as it should only happen rarely -- just wondering about importance).
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2051355805
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
Is this comment helpful or distracting?
```suggestion
// It's possible the transaction manager is already in the FATAL_ERROR
// state at this point. Depending on the incoming exception type,
// maybeTransitionToErrorState() could attempt to set the state to
// ABORTABLE_ERROR. For example, suppose a fatal error occurred during a
// transaction, and then moments later one of the batches in that
// transaction failed with a TimeoutException.
// maybeTransitionToErrorState() would then (blindly) attempt to
// transition to ABORTABLE_ERROR, which is invalid and would result
// in an IllegalStateException.
//
// Therefore, only attempt to transition to the FATAL_ERROR state if
// the batch is "fresh" *and* the transaction manager is not already
// in the FATAL_ERROR state.
if (!isStaleBatch && !hasFatalError())
maybeTransitionToErrorState(exception);
```
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2048203504
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -772,6 +779,14 @@ synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException except
}
}
+ /**
+ * Returns {@code true} if the given {@link ProducerBatch} has the same producer ID but a different epoch than the
+ * {@link #producerIdAndEpoch cached producer ID and epoch}.
+ */
+ synchronized boolean isStaleBatch(ProducerBatch batch) {
Review Comment:
Does this method need to by `synchronized`?
Also seems it could be `private`?
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,21 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ if (!isStaleBatch(batch) && !hasFatalError())
Review Comment:
Not sure if I understand the `!hasFatalError()` condition. Can you elaborate? -- I thought we want to call `maybeTransitionToErrorState(exception);` for any non-stale batch, independent of the current error state?
--
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-14830: Illegal state error in transactional producer [kafka]
Posted by "k-raina (via GitHub)" <gi...@apache.org>.
k-raina commented on code in PR #17022:
URL: https://github.com/apache/kafka/pull/17022#discussion_r2054686862
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -737,14 +737,25 @@ public synchronized void maybeTransitionToErrorState(RuntimeException exception)
}
synchronized void handleFailedBatch(ProducerBatch batch, RuntimeException exception, boolean adjustSequenceNumbers) {
- maybeTransitionToErrorState(exception);
+ // Compare the batch with the current ProducerIdAndEpoch. If the producer IDs are the *same* but the epochs
+ // are *different*, consider the batch as stale.
+ boolean isStaleBatch = batch.producerId() == producerIdAndEpoch.producerId && batch.producerEpoch() != producerIdAndEpoch.epoch;
+
+ if (!isStaleBatch && !hasFatalError())
+ maybeTransitionToErrorState(exception);
Review Comment:
```
// Therefore, only attempt to transition to the FATAL_ERROR state if
// the batch is "fresh" *and* the transaction manager is not already
// in the FATAL_ERROR state.
```
FATAL_ERROR to FATAL_ERROR[ is not an invalid transition](https://github.com/k-raina/kafka/blob/ab2cae14b93c13e21eda1d142d582bb7f42d7aec/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java#L211-L233). Do you mean `only attempt transition to the ABORTABLE_ERROR state`
EDIT: Okay i got that check is trying to say "Do not allow any transitions if transaction is in fatal state"
--
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