[GitHub] [kafka] cadonna commented on a diff in pull request #12583: KAFKA-10199: Separate state updater from old restore

[GitBox <gi...@apache.org>]

cadonna commented on code in PR #12583:
URL: https://github.com/apache/kafka/pull/12583#discussion_r960754245


##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -280,25 +280,17 @@ public void shouldClassifyExistingTasksWithStateUpdater() {
 
     @Test
     public void shouldAddTasksToStateUpdater() {
-        final StreamTask task00 = statefulTask(taskId00, taskId00Partitions)
+        final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
             .withInputPartitions(taskId00Partitions)
-            .inState(State.RESTORING)
-            .build();
-        final StandbyTask task01 = standbyTask(taskId01, taskId01Partitions)
+            .inState(State.RESTORING).build();
+        final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
             .withInputPartitions(taskId01Partitions)
-            .inState(State.RUNNING)
-            .build();
-        expect(changeLogReader.completedChangelogs()).andReturn(emptySet()).anyTimes();
-        expect(consumer.assignment()).andReturn(emptySet()).anyTimes();
-        consumer.resume(anyObject());
-        expectLastCall().anyTimes();
-        expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment))).andStubReturn(singletonList(task00));
-        expect(standbyTaskCreator.createTasks(eq(taskId01Assignment))).andStubReturn(singletonList(task01));
-        replay(activeTaskCreator, standbyTaskCreator, consumer, changeLogReader);
+            .inState(State.RUNNING).build();
+        final TasksRegistry tasks = mock(TasksRegistry.class);
+        when(tasks.drainPendingTaskToInit()).thenReturn(mkSet(task00, task01));
+        taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
 
-        taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, true);
-        taskManager.handleAssignment(taskId00Assignment, taskId01Assignment);
-        taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter -> { });
+        taskManager.checkStateUpdater(time.milliseconds(), noOpResetter -> { });
 

Review Comment:
   In this file, I did a lot of refactorings to use the utils to create task mocks and to mock the tasks registry. I had to touch those tests anyways so I thought I will bring them up-to-date.



-- 
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