vvcephei commented on a change in pull request #8994:
URL: https://github.com/apache/kafka/pull/8994#discussion_r453088861
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -193,6 +196,26 @@ private void closeAndRevive(final Map<Task,
Collection<TopicPartition>> taskWith
log.error("Error suspending corrupted task {} ", task.id(),
swallow);
}
task.closeDirty();
+ if (task.isActive()) {
+ // Pause so we won't poll any more records for this task until
it has been re-initialized
+ // Note, closeDirty already clears the partitiongroup for the
task.
+ final Set<TopicPartition> currentAssignment =
mainConsumer().assignment();
+ final Set<TopicPartition> assignedToPauseAndReset =
+ Utils.intersection(HashSet::new, currentAssignment,
task.inputPartitions());
Review comment:
Thanks all, it may well have just been a problem with the test setup,
but @ableegoldman is making me nervous that there may be some edge cases where
it's expected in production code.
Since we would never have had a dependency on the task inputs actually being
assigned here, and now we do, I'm concerned that if we try to be strict right
now, we may be introducing an IllegalStateException. Normally, this wouldn't be
a problem, but this PR is a last-minute hotfix before a release, so I'm feeling
cautious.
Since we seem to agree that the code is correct, although it may not be
strict, I've added a warning log instead of increasing strictness.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
