[GitHub] [kafka] lucasbru commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling

["lucasbru (via GitHub)" <gi...@apache.org>]

lucasbru commented on code in PR #13829:
URL: https://github.com/apache/kafka/pull/13829#discussion_r1222946279


##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -294,7 +297,13 @@ private void closeDirtyAndRevive(final Collection<Task> taskWithChangelogs, fina
 
                 task.addPartitionsForOffsetReset(assignedToPauseAndReset);
             }
+            if (stateUpdater != null) {
+                tasks.removeTask(task);

Review Comment:
   Any particular reason why we remove the task from `tasks` before reviving it? It would seem cleaner to me to basically remove from stream thread / add back to state updater directly in one block, which makes it easier to track ownership.



##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1970,6 +1970,29 @@ public void shouldThrowWhenHandlingClosingTasksOnProducerCloseError() {
         assertThat(thrown.getCause().getMessage(), is("KABOOM!"));
     }
 
+    @Test
+    public void shouldReAddRevivedTasksToStateUpdater() {
+        final StreamTask corruptedActiveTask = statefulTask(taskId03, taskId03ChangelogPartitions)
+            .inState(State.RESTORING)
+            .withInputPartitions(taskId03Partitions).build();
+        final StandbyTask corruptedStandbyTask = standbyTask(taskId02, taskId02ChangelogPartitions)
+            .inState(State.RUNNING)
+            .withInputPartitions(taskId02Partitions).build();
+        final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
+        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+        when(tasks.task(taskId03)).thenReturn( corruptedActiveTask);
+        when(tasks.task(taskId02)).thenReturn( corruptedStandbyTask);
+        expect(consumer.assignment()).andReturn(emptySet());
+        replay(consumer);
+
+        taskManager.handleCorruption(mkSet(corruptedActiveTask.id(), corruptedStandbyTask.id()));
+
+        Mockito.verify(tasks).removeTask(corruptedActiveTask);

Review Comment:
   Do we want to validate the Task state as well?



##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1970,6 +1970,29 @@ public void shouldThrowWhenHandlingClosingTasksOnProducerCloseError() {
         assertThat(thrown.getCause().getMessage(), is("KABOOM!"));
     }
 
+    @Test
+    public void shouldReAddRevivedTasksToStateUpdater() {
+        final StreamTask corruptedActiveTask = statefulTask(taskId03, taskId03ChangelogPartitions)
+            .inState(State.RESTORING)
+            .withInputPartitions(taskId03Partitions).build();
+        final StandbyTask corruptedStandbyTask = standbyTask(taskId02, taskId02ChangelogPartitions)
+            .inState(State.RUNNING)
+            .withInputPartitions(taskId02Partitions).build();
+        final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
+        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
+        when(tasks.task(taskId03)).thenReturn( corruptedActiveTask);

Review Comment:
   Awkward space after `(`, also in the line below, maybe use automatic formatting for the changed code



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org