vamossagar12 commented on a change in pull request #11433:
URL: https://github.com/apache/kafka/pull/11433#discussion_r739152945
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java
##########
@@ -292,12 +327,20 @@ public void handleAssignment(final Map<TaskId,
Set<TopicPartition>> activeTasks,
final Set<Task> tasksToRecycle = new TreeSet<>(byId);
final Set<Task> tasksToCloseClean = new TreeSet<>(byId);
final Set<Task> tasksToCloseDirty = new TreeSet<>(byId);
+ boolean commitAssignedActiveTasks = false;
+ final Set<Task> activeTasksNeedCommit = new HashSet<>();
// first rectify all existing tasks
for (final Task task : tasks.allTasks()) {
if (activeTasks.containsKey(task.id()) && task.isActive()) {
tasks.updateInputPartitionsAndResume(task,
activeTasks.get(task.id()));
activeTasksToCreate.remove(task.id());
+ if (task.state() == State.RESTORING) {
Review comment:
Thanks for the info. I will go through the `StoreChangelogReader` class
to understand the relevant logic.
However, I am slightly confused as there are a couple of first passes in
different places in your explanation :D . So, for now, if we find that
activeTasksToCreate is non-empty, then we can set commitAssignedActiveTasks to
true.
But, for the long term, we also need to check if there are some tasks still
in RESTORING at the end of TaskManager#tryToCompleteRestoration. Is that right?
This particular method is called from runOnce in StreamThread. So, are you
suggesting that we also add the check in this method and again check if there
are active tasks needing commit and the other condition - along with the one
added in handleAssignment?
--
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]
