lucasbru commented on code in PR #20858:
URL: https://github.com/apache/kafka/pull/20858#discussion_r2513925810
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3323,157 +3322,86 @@ public void
shouldCloseActiveTasksAndPropagateExceptionsOnCleanShutdownWithExact
}
private void
shouldCloseActiveTasksAndPropagateExceptionsOnCleanShutdown(final
ProcessingMode processingMode) {
- final TaskManager taskManager =
setUpTaskManagerWithoutStateUpdater(processingMode, null, false);
-
- final TopicPartition changelog = new TopicPartition("changelog", 0);
- final Map<TaskId, Set<TopicPartition>> assignment = mkMap(
- mkEntry(taskId00, taskId00Partitions),
- mkEntry(taskId01, taskId01Partitions),
- mkEntry(taskId02, taskId02Partitions),
- mkEntry(taskId03, taskId03Partitions)
- );
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions,
true, stateManager) {
- @Override
- public Set<TopicPartition> changelogPartitions() {
- return singleton(changelog);
- }
- };
- final AtomicBoolean closedDirtyTask01 = new AtomicBoolean(false);
- final AtomicBoolean closedDirtyTask02 = new AtomicBoolean(false);
- final AtomicBoolean closedDirtyTask03 = new AtomicBoolean(false);
- final Task task01 = new StateMachineTask(taskId01, taskId01Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("migrated", new
RuntimeException("cause"));
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask01.set(true);
- }
- };
- final Task task02 = new StateMachineTask(taskId02, taskId02Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new RuntimeException("oops");
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask02.set(true);
- }
- };
- final Task task03 = new StateMachineTask(taskId03, taskId03Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new RuntimeException("oops");
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask03.set(true);
- }
- };
-
- when(activeTaskCreator.createTasks(any(), eq(assignment)))
- .thenReturn(asList(task00, task01, task02, task03));
- taskManager.handleAssignment(assignment, emptyMap());
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
Review Comment:
This test is a bit ewird. IIUC, we have an active restoring task that is not
owned by the state updater in this test. But this should be impossible. I think
we shouldn't have `task00` like this.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3739,36 +3678,29 @@ public void
shouldShutDownStateUpdaterAndCloseDirtyTasksFailedDuringRemoval() {
}
@Test
- public void shouldInitializeNewActiveTasks() {
Review Comment:
Yes, I agree
--
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]
