vvcephei commented on a change in pull request #8994:
URL: https://github.com/apache/kafka/pull/8994#discussion_r452949230
##########
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());
+
+ mainConsumer().pause(assignedToPauseAndReset);
+ final Map<TopicPartition, OffsetAndMetadata> committed =
mainConsumer().committed(assignedToPauseAndReset);
+ for (final Map.Entry<TopicPartition, OffsetAndMetadata>
committedEntry : committed.entrySet()) {
+ final OffsetAndMetadata offsetAndMetadata =
committedEntry.getValue();
+ if (offsetAndMetadata != null) {
+ mainConsumer().seek(committedEntry.getKey(),
offsetAndMetadata);
+
assignedToPauseAndReset.remove(committedEntry.getKey());
+ }
+ }
+ final Set<TopicPartition> remainder =
resetter.apply(assignedToPauseAndReset);
+ // If anything didn't have a configured policy, reset to
beginning
+ mainConsumer().seekToBeginning(remainder);
Review comment:
Thanks. I was confused by the behaviour of the StreamThreadTest. I think
there was something funny in the StreamThread factory, which just never
surfaced before. I'll call it out for your review.
----------------------------------------------------------------
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:
us...@infra.apache.org
