You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/09/08 18:36:25 UTC
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12607: KAFKA-14133: Replace EasyMock with Mockito in streams tests
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org