ableegoldman commented on a change in pull request #8248:
URL: https://github.com/apache/kafka/pull/8248#discussion_r418842202
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -155,9 +155,6 @@ void handleCorruption(final Map<TaskId,
Collection<TopicPartition>> taskWithChan
final TaskId taskId = entry.getKey();
final Task task = tasks.get(taskId);
- // this call is idempotent so even if the task is only CREATED we
can still call it
- changelogReader.remove(task.changelogPartitions());
Review comment:
Yeah, so `remove` is basically the opposite of `register` in the
ChangelogReader and previously the registration was done by the
ProcessorStateManager while the TaskManager was responsible for unregistering
changelogs (via `remove`).
I renamed `remove` to `unregister` and moved the unregistration to the
ProcessorStateManager as well, so the changelog lifecycle is entirely managed
by the state manager now.
----------------------------------------------------------------
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
