KAFKA-19864: Handle TimeoutException from initializeIfNeeded() in StateUpdater Code #20829
Conversation
shashankhs11
changed the title
| for (final StreamTask restoredTask : restoredTasks) { | ||
| verify(restoredTask).completeRestoration(noOpResetter); | ||
| verify(restoredTask).clearTaskTimeout(); | ||
| verify(restoredTask, atLeastOnce()).clearTaskTimeout(); |
There was a problem hiding this comment.
I added this because, one of the tests shouldAddNewActiveTasks was failing as it expected the call to clearTaskTimeout() more than once.
There was a problem hiding this comment.
Pull Request Overview
This PR adds handling for TimeoutException during task initialization in the state updater, aligning its behavior with existing LockException handling. It also ensures that task timeouts are properly cleared after successful initialization.
Key changes:
- Added a
TimeoutExceptioncatch block inaddTaskToStateUpdaterthat retries initialization with backoff - Added
clearTaskTimeout()call after successful task initialization - Updated tests to verify timeout handling behavior and clearTaskTimeout invocation
Reviewed Changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| TaskManager.java | Added TimeoutException handling with retry logic and clearTaskTimeout call after successful initialization |
| TaskManagerTest.java | Added test for TimeoutException during initialization and updated existing tests to verify clearTaskTimeout behavior |
Comments suppressed due to low confidence (1)
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:1160
- The test
shouldAddTasksToStateUpdatershould verify thatclearTaskTimeout()is called on both tasks after successful initialization, consistent with the new behavior added in line 1085 of TaskManager.java and verified in other tests likeshouldRetryInitializationWhenLockExceptionInStateUpdater.
verify(task00).initializeIfNeeded();
verify(task01).initializeIfNeeded();
verify(stateUpdater).add(task00);
verify(stateUpdater).add(task01);
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
lucasbru
left a comment
lucasbru
left a comment
There was a problem hiding this comment.
THanks! Looks good to me with minor comments.
| task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException); | ||
| tasks.addPendingTasksToInit(Collections.singleton(task)); | ||
| updateOrCreateBackoffRecord(task.id(), nowMs); | ||
| log.debug("Task {} timed out during initialization; will retry", task.id(), timeoutException); |
| tasks.addPendingTasksToInit(Collections.singleton(task)); | ||
| updateOrCreateBackoffRecord(task.id(), nowMs); | ||
| } catch (final TimeoutException timeoutException) { | ||
| task.maybeInitTaskTimeoutOrThrow(nowMs, timeoutException); |
There was a problem hiding this comment.
Maybe add a comment when this can happen: Either during producer initialization, or while fetching committed offset.
shashankhs11
force-pushed
the
KAFKA-19684
branch
from
a0e55f0 to
7c1bfc7
Compare
…teUpdater Code (apache#20829) Handle TimeoutException in StateUpdater code path in `TaskManager.java` Reviewers: Lucas Brutschy <[email protected]>
2 participants
Handle TimeoutException in StateUpdater code path in
TaskManager.javaReviewers: Lucas Brutschy [email protected]