ableegoldman commented on a change in pull request #8994:
URL: https://github.com/apache/kafka/pull/8994#discussion_r453027126
##########
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:
Could this happen due to some race condition with a deleted regex topic?
I assume that's not happening in your tests, just wondering if this check
actually is necessary or if it would only be masking a bug as Matthias said.
I'm _pretty_ sure we should always go through the process of updating the
task's input partitions immediately after updating the consumer's assignment,
so we would never find them to be out of sync. But can the consumer assignment
be updated outside of a rebalance, eg based on metadata refresh?
----------------------------------------------------------------
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]
