sunm2n commented on code in PR #23549:
URL: https://github.com/apache/kafka/pull/23549#discussion_r4411172086
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1538,10 +1538,48 @@ private void shutdownStateUpdater() {
}
final Set<Task> tasksToCloseClean = new
TreeSet<>(Comparator.comparing(Task::id));
final Set<Task> tasksToCloseDirty = new
TreeSet<>(Comparator.comparing(Task::id));
- addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty);
- // at this point we removed all tasks, so the shutdown should not take
a lot of time
+ try {
+ addToTasksToClose(futures, tasksToCloseClean, tasksToCloseDirty);
+ } catch (final IllegalStateException interruptedDuringRemoval) {
+ if (!(interruptedDuringRemoval.getCause() instanceof
InterruptedException)) {
Review Comment:
Thanks, addressed in 907d4dc0b7.
The interrupt status is left untouched during removal waits, since an
interrupt is the only way to stop waiting for a removal that never completes.
If a removal wait is interrupted, the interruption is recorded and the flag is
cleared before cleanup continues; the same happens if the state updater
restores the flag after its join. The caller's interrupt status is restored in
a `finally` block in `TaskManager#shutdown`, including when cleanup throws.
Added three tests in `TaskManagerTest`. The first two fail against the
previous PR revision. The third passes there too, since that revision never
cleared the flag; it guards against losing the interrupt status if shutdown
throws after the flag has been cleared.
-
`shouldClearInterruptWhileClosingRemainingResourcesAndRestoreItAfterShutdown`:
the flag is cleared when the thread producer is closed during `shutdown(true)`,
and restored afterwards.
- `shouldClearInterruptWhileShuttingDownProcessingThreadsAndLockingTasks`:
with processing threads enabled, the flag is cleared when `TaskManager` shuts
down the task executors and locks the tasks to close.
- `shouldRestoreInterruptStatusWhenShutdownThrowsAfterClearingIt`: the flag
is restored when the state updater shutdown throws after the interruption was
cleared.
The processing-thread test uses a mocked `DefaultTaskManager`, so it checks
the interrupt status at the calls from `TaskManager`; it does not exercise a
busy executor's join or the duplicate release request themselves.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
