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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
