KAFKA-14830: Illegal state error in transactional producer #17022

kirktrue · 2024-08-27T22:35:04Z

When the producer's transaction manager receives a notification that an
error has occurred during a transaction, it takes steps to abort the
transaction and reset its internal state.

Users have reported the following case where a producer experiences
timeouts while in a transaction:

The TransactionManager (TM) starts with state READY and epoch
set to 0
A transaction (T1) begins and TM sets its internal state to
IN_TRANSACTION
Batches are created and sent off to their respective brokers
A timeout threshold is hit
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
A moment later, in the Sender thread, one of the failed batches
calls handleFailedBatch()
handleFailedBatch() sets the TM state to ABORTABLE_ERROR which
is an invalid state transition from READY, hence the exception

This change compares the transaction manager's current epoch (1)
with the batch's epoch (0). If they're different, the batch is
considered "stale" and can be ignored (though a DEBUG message is
logged).

jolshan · 2024-08-28T19:58:31Z

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;


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)

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!

jolshan · 2024-08-28T19:59:44Z

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.

github-actions · 2025-01-16T03:35:31Z

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.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

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.

kirktrue · 2025-01-16T18:07:23Z

Still needed, just lower priority 😞

kirktrue · 2025-03-26T19:50:24Z

cc @k-raina

kirktrue · 2025-03-31T20:49:50Z

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.

@jolshan—The unit tests mimic the use cases that were seen in the wild. What other test cases should we consider? Thanks!

…ansitions

mjsax · 2025-04-17T04:19:21Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+     * 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) {


Does this method need to by synchronized?

Also seems it could be private?

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.

I inlined the 'is stale batch' logic into handleFailedBatch(), so these issues are no longer present.

mjsax · 2025-04-17T04:20:31Z

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())


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?

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

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)

kirktrue · 2025-04-19T02:46:33Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+        if (!isStaleBatch && !hasFatalError())
+            maybeTransitionToErrorState(exception);


Is this comment helpful or distracting?

Suggested change

if (!isStaleBatch && !hasFatalError())

maybeTransitionToErrorState(exception);

// 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);

kirktrue added 4 commits August 15, 2024 17:59

WIP

7522421

Updates to tests and a first attempt at "fixing" the problem

26c6eae

Clean up

4be2627

Removed temporary logging

9cc5688

kirktrue added the producer label Aug 28, 2024

jolshan reviewed Aug 28, 2024

View reviewed changes

kirktrue added 2 commits August 30, 2024 17:40

Pulled out isStaleBatch for clarity

11d98a3

Merge branch 'trunk' into KAFKA-14830-prevent-illegal-state-transitions

8f73d09

github-actions bot added clients small Small PRs labels Oct 17, 2024

kirktrue added transactions Transactions and EOS ci-approved labels Oct 17, 2024

github-actions bot added the stale Stale PRs label Jan 16, 2025

kirktrue removed the stale Stale PRs label Jan 16, 2025

Merge branch 'trunk' into KAFKA-14830-prevent-illegal-state-transitions

729f5dc

Merge branch 'trunk' into KAFKA-14830-prevent-illegal-state-transitions

7d82a63

kirktrue marked this pull request as ready for review March 31, 2025 20:47

Merge branch 'apache:trunk' into KAFKA-14830-prevent-illegal-state-tr…

31e3a82

…ansitions

kirktrue changed the title ~~[WIP] KAFKA-14830: Illegal state error in transactional producer~~ KAFKA-14830: Illegal state error in transactional producer Apr 7, 2025

kirktrue requested a review from jolshan April 7, 2025 18:48

mjsax reviewed Apr 17, 2025

View reviewed changes

kirktrue added 2 commits April 18, 2025 19:16

Inlining isStaleBatch

569467f

Update TransactionManager.java

0567155

kirktrue commented Apr 19, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14830: Illegal state error in transactional producer #17022

KAFKA-14830: Illegal state error in transactional producer #17022

kirktrue commented Aug 27, 2024 •

edited by github-actions bot

Loading

jolshan Aug 28, 2024 •

edited

Loading

kirktrue Mar 26, 2025

jolshan commented Aug 28, 2024

github-actions bot commented Jan 16, 2025

kirktrue commented Jan 16, 2025

kirktrue commented Mar 26, 2025

kirktrue commented Mar 31, 2025 •

edited

Loading

mjsax Apr 17, 2025

kirktrue Apr 19, 2025

kirktrue Apr 19, 2025

mjsax Apr 17, 2025

kirktrue Apr 19, 2025

kirktrue Apr 19, 2025

kirktrue Apr 19, 2025

		if (!isStaleBatch && !hasFatalError())
		maybeTransitionToErrorState(exception);

-        if (!isStaleBatch && !hasFatalError())
-            maybeTransitionToErrorState(exception);
+        // 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);

KAFKA-14830: Illegal state error in transactional producer #17022

Are you sure you want to change the base?

KAFKA-14830: Illegal state error in transactional producer #17022

Conversation

kirktrue commented Aug 27, 2024 • edited by github-actions bot Loading

jolshan Aug 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jolshan commented Aug 28, 2024

github-actions bot commented Jan 16, 2025

kirktrue commented Jan 16, 2025

kirktrue commented Mar 26, 2025

kirktrue commented Mar 31, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirktrue commented Aug 27, 2024 •

edited by github-actions bot

Loading

jolshan Aug 28, 2024 •

edited

Loading

kirktrue commented Mar 31, 2025 •

edited

Loading