Copilot commented on code in PR #23549:
URL: https://github.com/apache/kafka/pull/23549#discussion_r4410838532
##########
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:
Keeping the interrupt flag set lets executor shutdown abandon its join; if
an executor still holds a task, `maybeLockTasks` can issue a duplicate release
request and throw before initialized tasks are closed, after which shutdown
clears the registry. The flag can also make `shutdown(true)` fail when producer
close encounters the same interrupt. Record and clear the interruption before
continuing cleanup, restore it in an outer `finally` after all shutdown stages,
and add regression coverage for a busy processing executor and
interrupt-sensitive producer cleanup.
--
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]
