guozhangwang commented on code in PR #12607:
URL: https://github.com/apache/kafka/pull/12607#discussion_r966289656
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1300,28 +1229,16 @@ public void
shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
// `handleAssignment`
expectRestoreToBeCompleted(consumer, changeLogReader);
- expect(activeTaskCreator.createTasks(anyObject(),
eq(taskId00Assignment))).andStubReturn(singletonList(task00));
-
expect(standbyTaskCreator.createTasks(eq(taskId01Assignment))).andStubReturn(singletonList(task01));
- topologyBuilder.addSubscribedTopicsFromAssignment(anyObject(),
anyString());
- expectLastCall().anyTimes();
+ when(activeTaskCreator.createTasks(any(),
eq(taskId00Assignment))).thenReturn(singletonList(task00));
+
when(standbyTaskCreator.createTasks(eq(taskId01Assignment))).thenReturn(singletonList(task01));
+ // The second attempt will return empty tasks.
makeTaskFolders(taskId00.toString(), taskId01.toString());
expectLockObtainedFor(taskId00, taskId01);
- // The second attempt will return empty tasks.
- makeTaskFolders();
Review Comment:
Why we can remove those calls?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -325,25 +318,21 @@ public void
shouldHandleRemovedTasksToRecycleFromStateUpdater() {
when(stateUpdater.drainRemovedTasks()).thenReturn(mkSet(task00,
task01));
taskManager =
setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01),
eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
- expect(standbyTaskCreator.createStandbyTaskFromActive(eq(task00),
eq(taskId00Partitions)))
- .andStubReturn(task00Converted);
- expect(consumer.assignment()).andReturn(emptySet()).anyTimes();
- consumer.resume(anyObject());
- expectLastCall().anyTimes();
- replay(activeTaskCreator, standbyTaskCreator, topologyBuilder,
consumer);
+ when(activeTaskCreator.createActiveTaskFromStandby(eq(task01),
eq(taskId01Partitions), eq(consumer)))
+ .thenReturn(task01Converted);
+ when(standbyTaskCreator.createStandbyTaskFromActive(eq(task00),
eq(taskId00Partitions)))
+ .thenReturn(task00Converted);
+ when(consumer.assignment()).thenReturn(emptySet());
taskManager.tasks().addPendingTaskToRecycle(taskId00,
taskId00Partitions);
taskManager.tasks().addPendingTaskToRecycle(taskId01,
taskId01Partitions);
taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter
-> { });
- Mockito.verify(task00Converted).initializeIfNeeded();
- Mockito.verify(task01Converted).initializeIfNeeded();
- Mockito.verify(stateUpdater).add(task00Converted);
- Mockito.verify(stateUpdater).add(task01Converted);
+ verify(task00Converted).initializeIfNeeded();
+ verify(task01Converted).initializeIfNeeded();
+ verify(stateUpdater).add(task00Converted);
+ verify(stateUpdater).add(task01Converted);
+ verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(any());
Review Comment:
How to specify that we expect this function call only once? Should we use
`verify(activeTaskCreator, times(1)).func();` instead?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4091,21 +3815,20 @@ public void shouldListNotPausedTasks() {
topologyMetadata.pauseTopology(UNNAMED_TOPOLOGY);
assertEquals(taskManager.notPausedTasks().size(), 0);
+
+ verifyConsumerResumedWithAssignment(consumer);
Review Comment:
Why add this additional verification?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -271,11 +265,13 @@ public void shouldClassifyExistingTasksWithStateUpdater()
{
taskManager.handleAssignment(standbyTasks, restoringActiveTasks);
- Mockito.verify(stateUpdater).getTasks();
- Mockito.verify(stateUpdater).remove(standbyTaskToRecycle.id());
- Mockito.verify(stateUpdater).remove(standbyTaskToClose.id());
- Mockito.verify(stateUpdater).remove(restoringActiveTaskToRecycle.id());
- Mockito.verify(stateUpdater).remove(restoringActiveTaskToClose.id());
+ verify(stateUpdater).getTasks();
+ verify(stateUpdater).remove(standbyTaskToRecycle.id());
+ verify(stateUpdater).remove(standbyTaskToClose.id());
+ verify(stateUpdater).remove(restoringActiveTaskToRecycle.id());
+ verify(stateUpdater).remove(restoringActiveTaskToClose.id());
+
+ verifyConsumerResumedWithAssignment(consumer);
Review Comment:
There are several tests where we are adding this verification, could you
elaborate a bit why?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4068,18 +3796,14 @@ public void shouldConvertStandbyTaskToActiveTask() {
when(activeTask.id()).thenReturn(taskId00);
when(activeTask.inputPartitions()).thenReturn(taskId00Partitions);
- expect(activeTaskCreator.createTasks(anyObject(),
eq(Collections.emptyMap()))).andReturn(Collections.emptySet());
-
expect(standbyTaskCreator.createTasks(eq(taskId00Assignment))).andReturn(singletonList(standbyTask));
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(standbyTask),
eq(taskId00Partitions), anyObject())).andReturn(activeTask);
- expect(activeTaskCreator.createTasks(anyObject(),
eq(Collections.emptyMap()))).andReturn(Collections.emptySet());
-
expect(standbyTaskCreator.createTasks(eq(Collections.emptyMap()))).andReturn(Collections.emptySet());
-
- replay(standbyTaskCreator, activeTaskCreator, consumer);
+ when(activeTaskCreator.createTasks(any(),
eq(Collections.emptyMap()))).thenReturn(Collections.emptySet());
Review Comment:
Here the goals are both verifying those functions should be called, and also
return the results when they are called (you can see we did not use
`andStubReturn` here). When replacing them with `when` we would not verify
anymore right? Ditto elsewhere.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -325,25 +318,21 @@ public void
shouldHandleRemovedTasksToRecycleFromStateUpdater() {
when(stateUpdater.drainRemovedTasks()).thenReturn(mkSet(task00,
task01));
taskManager =
setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01),
eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
- expect(standbyTaskCreator.createStandbyTaskFromActive(eq(task00),
eq(taskId00Partitions)))
- .andStubReturn(task00Converted);
- expect(consumer.assignment()).andReturn(emptySet()).anyTimes();
- consumer.resume(anyObject());
- expectLastCall().anyTimes();
- replay(activeTaskCreator, standbyTaskCreator, topologyBuilder,
consumer);
+ when(activeTaskCreator.createActiveTaskFromStandby(eq(task01),
eq(taskId01Partitions), eq(consumer)))
+ .thenReturn(task01Converted);
+ when(standbyTaskCreator.createStandbyTaskFromActive(eq(task00),
eq(taskId00Partitions)))
+ .thenReturn(task00Converted);
+ when(consumer.assignment()).thenReturn(emptySet());
taskManager.tasks().addPendingTaskToRecycle(taskId00,
taskId00Partitions);
taskManager.tasks().addPendingTaskToRecycle(taskId01,
taskId01Partitions);
taskManager.tryToCompleteRestoration(time.milliseconds(), noOpResetter
-> { });
- Mockito.verify(task00Converted).initializeIfNeeded();
- Mockito.verify(task01Converted).initializeIfNeeded();
- Mockito.verify(stateUpdater).add(task00Converted);
- Mockito.verify(stateUpdater).add(task01Converted);
+ verify(task00Converted).initializeIfNeeded();
+ verify(task01Converted).initializeIfNeeded();
+ verify(stateUpdater).add(task00Converted);
+ verify(stateUpdater).add(task01Converted);
+ verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(any());
Review Comment:
Ditto elsewhere for replacing `expectLastCall().once()`?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2180,29 +2035,16 @@ public void
shouldCommitAllActiveTasksThatNeedCommittingOnHandleRevocationWithEo
);
expectRestoreToBeCompleted(consumer, changeLogReader);
- expect(activeTaskCreator.createTasks(anyObject(),
eq(assignmentActive)))
- .andReturn(asList(task00, task01, task02));
+ when(activeTaskCreator.createTasks(any(), eq(assignmentActive)))
+ .thenReturn(asList(task00, task01, task02));
- expect(activeTaskCreator.threadProducer()).andReturn(producer);
+ when(activeTaskCreator.threadProducer()).thenReturn(producer);
activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
- expect(standbyTaskCreator.createTasks(eq(assignmentStandby)))
- .andReturn(singletonList(task10));
+ when(standbyTaskCreator.createTasks(eq(assignmentStandby)))
+ .thenReturn(singletonList(task10));
final ConsumerGroupMetadata groupMetadata = new
ConsumerGroupMetadata("appId");
- expect(consumer.groupMetadata()).andReturn(groupMetadata);
- producer.commitTransaction(expectedCommittedOffsets, groupMetadata);
- expectLastCall();
-
- task00.committedOffsets();
Review Comment:
Why we can remove those verifications now?
--
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
