You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "cadonna (via GitHub)" <gi...@apache.org> on 2023/06/08 11:55:57 UTC
[GitHub] [kafka] cadonna opened a new pull request, #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
cadonna opened a new pull request, #13829:
URL: https://github.com/apache/kafka/pull/13829
Fixes a bug regarding the state updater where tasks that experience corruption during restoration are passed from the state updater to the stream thread for closing and reviving but then the revived tasks are not re-added to the state updater.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] cadonna commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13829:
URL: https://github.com/apache/kafka/pull/13829#discussion_r1222962171
##########
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:
Yes! The reason is that `removeTask()` verifies that the state of the task is `CLOSED` and throws an exception when it is not. I did not want to change the check and lose that guard.
--
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
[GitHub] [kafka] cadonna commented on pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #13829:
URL: https://github.com/apache/kafka/pull/13829#issuecomment-1586163556
Build failures are unrelated:
```
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testReplicateSourceDefault()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetSyncsTopicsOnTarget()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testSyncTopicConfigs()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetSyncsTopicsOnTarget()
Build / JDK 11 and Scala 2.13 / kafka.controller.ControllerIntegrationTest.testTopicIdPersistsThroughControllerRestart()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testReplicationWithEmptyPartition()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testOffsetSyncsTopicsOnTarget()
Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest.testReplicationWithEmptyPartition()
Build / JDK 17 and Scala 2.13 / kafka.api.TransactionsTest.testBumpTransactionalEpoch(String).quorum=zk
Build / JDK 8 and Scala 2.12 / org.apache.kafka.clients.consumer.KafkaConsumerTest.testGracefulClose()
Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.mirror.integration.DedicatedMirrorIntegrationTest.testMultiNodeCluster()
Build / JDK 8 and Scala 2.12 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest.testReplicateSourceDefault()
Build / JDK 8 and Scala 2.12 / integration.kafka.server.FetchFromFollowerIntegrationTest.testRackAwareRangeAssignor()
Build / JDK 8 and Scala 2.12 / kafka.admin.TopicCommandIntegrationTest.testDescribeUnderMinIsrPartitions(String).quorum=zk
```
--
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
[GitHub] [kafka] lucasbru commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "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
[GitHub] [kafka] cadonna commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13829:
URL: https://github.com/apache/kafka/pull/13829#discussion_r1222976522
##########
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:
The task is a mock. The state of the task is specified by the test. However, I added verifications for the method calls that change the state of the task.
--
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
[GitHub] [kafka] cadonna merged pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #13829:
URL: https://github.com/apache/kafka/pull/13829
--
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
[GitHub] [kafka] lucasbru commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13829:
URL: https://github.com/apache/kafka/pull/13829#discussion_r1222992405
##########
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:
Yes, that's what I meant. Thanks
--
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
[GitHub] [kafka] lucasbru commented on a diff in pull request #13829: KAFKA-10199: Re-add revived tasks to the state updater after handling
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #13829:
URL: https://github.com/apache/kafka/pull/13829#discussion_r1222994406
##########
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:
Got it. Could make sense to add a little comment, but I won't insist
--
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