KAFKA-14133: Replace EasyMock with Mockito in streams tests #12607

clolov · 2022-09-08T13:52:28Z

Batch 5 of the tests detailed in https://issues.apache.org/jira/browse/KAFKA-14133 which use EasyMock and need to be moved to Mockito.

clolov · 2022-09-08T13:56:45Z

There were stubbings which were no longer on the call path so I have removed them. The way I checked that I wasn't changing the test behaviour was to use EasyMock.verify on the mocks and confirming that the stubbings were indeed unused prior to my change. There are multiple possibilities for refactoring, but I chose to keep the changes as close to the EasyMock implementation as the PR is already big.

clolov · 2022-09-08T13:58:17Z

I am aware that there are merge conflicts and I will aim to address them over the coming days.

clolov · 2022-09-08T13:59:08Z

@guozhangwang and @cadonna for visibility

guozhangwang

Thanks @clolov , just a few quick meta thoughts:

I think we can remove the StateMachineTask class as well since it was used as a mock really to compensate the incapabilities of easymock. While now with Mockito I think we would no longer need it. The only thing I'd need to ask is how to stubbing those state manipulations (e.g. if setCommitNeeded() is called, then followed commitNeeded should return true).
There are a few places where we used expect.andReturn which was for both stubbing the return values but as well as expecting those functions to be called, when they are replaced with when.thenReturn we would not do verifications afaik, is that right? If yes could we fix them?

cc @cadonna for a second pass.

guozhangwang · 2022-09-08T18:17:38Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+        verify(task01Converted).initializeIfNeeded();
+        verify(stateUpdater).add(task00Converted);
+        verify(stateUpdater).add(task01Converted);
+        verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(any());


How to specify that we expect this function call only once? Should we use verify(activeTaskCreator, times(1)).func(); instead?

Ditto elsewhere for replacing expectLastCall().once()?

By default it expects the method call to happen only once:

public static <T> T verify(T mock) Verifies certain behavior happened once.

https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#verify(T)

Ack, thanks!

guozhangwang · 2022-09-08T18:27:44Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

        makeTaskFolders(taskId00.toString(), taskId01.toString());
        expectLockObtainedFor(taskId00, taskId01);

-        // The second attempt will return empty tasks.
-        makeTaskFolders();


Why we can remove those calls?

I have altered the makeTaskFolders function to take the second call into account.

Before:

expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(taskFolders).once();

After:

when(stateDirectory.listNonEmptyTaskDirectories()).thenReturn(taskFolders).thenReturn(Collections.emptyList());

As far as I understand subsequent calls for stubbing in Mockito overwrite previous ones, so I cannot do exactly what was done with EasyMock.

guozhangwang · 2022-09-08T18:29:15Z

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


Why add this additional verification?

The below method set an expectation for a method call which was not stubbed:

private static void expectRestoreToBeCompleted(final Consumer<byte[], byte[]> consumer, final ChangelogReader changeLogReader, final boolean changeLogUpdateRequired) { final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0)); expect(consumer.assignment()).andReturn(assignment); consumer.resume(assignment); <--- THIS IS THE EXPECTATION expectLastCall(); expect(changeLogReader.completedChangelogs()).andReturn(emptySet()).times(changeLogUpdateRequired ? 1 : 0, 1); }

I wanted to keep this verification so I moved it to a separate function.

Ah I see, thanks!

guozhangwang · 2022-09-08T18:29:56Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+        verify(stateUpdater).remove(restoringActiveTaskToRecycle.id());
+        verify(stateUpdater).remove(restoringActiveTaskToClose.id());
+
+        verifyConsumerResumedWithAssignment(consumer);


There are several tests where we are adding this verification, could you elaborate a bit why?

Same as above.

guozhangwang · 2022-09-08T18:30:55Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

-        producer.commitTransaction(expectedCommittedOffsets, groupMetadata);
-        expectLastCall();
-
-        task00.committedOffsets();


Why we can remove those verifications now?

As far as I am aware Mockito cannot verify interactions with things which are not mocks. This being said, given you suggested I remove the StateMachineTask this might change in subsequent commits.

Sounds great, thanks.

guozhangwang · 2022-09-08T18:34:57Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

-        expect(standbyTaskCreator.createTasks(eq(Collections.emptyMap()))).andReturn(Collections.emptySet());
-
-        replay(standbyTaskCreator, activeTaskCreator, consumer);
+        when(activeTaskCreator.createTasks(any(), eq(Collections.emptyMap()))).thenReturn(Collections.emptySet());


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.

If we use @RunWith(MockitoJUnitRunner.StrictStubs.class) and we run the whole test class Mockito verifies that all stubbings have indeed been used.

clolov · 2022-09-09T09:58:57Z

Hey @guozhangwang and thank you for the insightful comments! I will aim to give explanations and amend the PR in the upcoming days.

clolov · 2022-11-24T10:48:27Z

Polite bump for a review here @cadonna. I have rebased on the latest trunk and I believe I have answered the majority of the comments already on the pull request. Given the size of the change I would prefer to get rid of StateMachineTask in a seprate pull request rather than as part of this one.

clolov · 2023-02-02T13:19:26Z

@cadonna I have rebased this on the latest trunk. Is it possible to get a review? This pull requests is one of the last remaining ones for completing the Mockito migration for streams.

cadonna

@clolov Thanks for your patience!

Here my feedback for the the first 1500 lines. I will continue to review the PR in 1.5 weeks after my vacation.

cadonna · 2023-04-04T12:04:05Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

-            .inState(State.RESTORING)
-            .withInputPartitions(taskId00Partitions).build();
+                .inState(State.RESTORING)
+                .withInputPartitions(taskId00Partitions).build();
        final StandbyTask taskToRecycle1 = standbyTask(taskId01, taskId01ChangelogPartitions)
-            .inState(State.RUNNING)
-            .withInputPartitions(taskId01Partitions).build();
+                .inState(State.RUNNING)
+                .withInputPartitions(taskId01Partitions).build();
        final StandbyTask convertedTask0 = standbyTask(taskId00, taskId00ChangelogPartitions).build();
        final StreamTask convertedTask1 = statefulTask(taskId01, taskId01ChangelogPartitions).build();
        final StreamTask taskToClose = statefulTask(taskId02, taskId02ChangelogPartitions)
-            .inState(State.RESTORING)
-            .withInputPartitions(taskId02Partitions).build();
+                .inState(State.RESTORING)
+                .withInputPartitions(taskId02Partitions).build();
        final StreamTask taskToUpdateInputPartitions = statefulTask(taskId03, taskId03ChangelogPartitions)
-            .inState(State.RESTORING)
-            .withInputPartitions(taskId03Partitions).build();
+                .inState(State.RESTORING)
+                .withInputPartitions(taskId03Partitions).build();
        when(stateUpdater.hasRemovedTasks()).thenReturn(true);
        when(stateUpdater.drainRemovedTasks())
-            .thenReturn(mkSet(taskToRecycle0, taskToRecycle1, taskToClose, taskToUpdateInputPartitions));
+                .thenReturn(mkSet(taskToRecycle0, taskToRecycle1, taskToClose, taskToUpdateInputPartitions));


nit: adding the indentation is not necessary.

cadonna · 2023-04-04T12:05:06Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

-            argThat(taskId -> !taskId.equals(taskToRecycle0.id()) && !taskId.equals(taskToRecycle1.id())))
+                argThat(taskId -> !taskId.equals(taskToRecycle0.id()) && !taskId.equals(taskToRecycle1.id())))


nit: adding indentation not necessary

cadonna · 2023-04-04T12:05:42Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+        when(activeTaskCreator.createActiveTaskFromStandby(taskToRecycle1, taskId01Partitions, consumer))
+                .thenReturn(convertedTask1);
+        when(standbyTaskCreator.createStandbyTaskFromActive(taskToRecycle0, taskId00Partitions))
+                .thenReturn(convertedTask0);


nit: we usually indent 4 spaces.

cadonna · 2023-04-04T12:16:36Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

@@ -1060,14 +996,12 @@ public void shouldTransitRestoredTaskToRunning() {
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTransitionToRunningOfRestoredTask(task, tasks);
        consumer.resume(task.inputPartitions());


This needs to become a verification

verify(consumer).resume(task.inputPartitions());

further down in the verification block.

cadonna · 2023-04-04T12:19:09Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

-        expect(standbyTaskCreator.createStandbyTaskFromActive(statefulTask, statefulTask.inputPartitions()))
-            .andStubReturn(standbyTask);
+        when(standbyTaskCreator.createStandbyTaskFromActive(statefulTask, statefulTask.inputPartitions()))
+            .thenReturn(standbyTask);
        activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(statefulTask.id());


This needs to become verification

verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(statefulTask.id());

in the verification block further down.

cadonna · 2023-04-04T12:22:11Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

@@ -1268,16 +1188,14 @@ public void shouldCloseDirtyRestoredTask() {
        when(stateUpdater.restoresActiveTasks()).thenReturn(true);
        activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(statefulTask.id());


This needs to become verification

verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(statefulTask.id());

in the verification block.

cadonna · 2023-04-04T12:22:58Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

@@ -1292,15 +1210,13 @@ public void shouldUpdateInputPartitionsOfRestoredTask() {
        when(stateUpdater.restoresActiveTasks()).thenReturn(true);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
        consumer.resume(statefulTask.inputPartitions());


This needs to become verification

verify(consumer).resume(statefulTask.inputPartitions());

in the verification block.

cadonna · 2023-04-04T13:33:55Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

    private static void expectConsumerAssignmentPaused(final Consumer<byte[], byte[]> consumer) {
        final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
-        expect(consumer.assignment()).andReturn(assignment);
-        consumer.pause(assignment);
+        when(consumer.assignment()).thenReturn(assignment);
+        doNothing().when(consumer).pause(assignment);
    }


This method is not needed anymore.

cadonna · 2023-04-04T13:38:52Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

+        verify(consumer).pause(assignment);
+        verify(stateDirectory).unlock(taskId02);
+
+        verifyConsumerResumedWithAssignment(consumer);


This method should also take the assignment as an argument. Maybe it is even better to call

verify(consumer, atLeastOnce()).resume(assignment);

instead of verifyConsumerResumedWithAssignment(consumer) here.

cadonna · 2023-04-04T13:49:42Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java

@@ -1667,6 +1566,8 @@ public void shouldComputeOffsetSumForStandbyTask() throws Exception {
        restoringTask.setChangelogOffsets(changelogOffsets);

        assertThat(taskManager.getTaskOffsetSums(), is(expectedOffsetSums));
+
+        verifyConsumerResumedWithAssignment(consumer);


Why did you add this verification? It was not there before the migration, right?

cadonna · 2023-04-08T09:00:33Z

@clolov Since this PR is rather long and hard to review, I took a stab to subdivide it into smaller PRs by migrating single mocks. I opened a PR for the migration of the topology builder mock: #13529.
Let me know what you think about it.
If you think that is a good idea feel free to migrate the other mocks. I think we also need a bit to rethink the testing strategy in this test class. Sorry that this migration gets so complicated.

clolov · 2023-04-12T09:31:30Z

Heya @cadonna, thank you for the review! The approach you have taken in the other PR makes sense to me, so I will aim to open a few pull requests in a similar fashion. What worries you about the testing strategy in the class?

cadonna · 2023-04-13T14:01:31Z

What worries you about the testing strategy in the class?

My worry is that we verify interactions with mocks (mainly strict mocks) in unit tests in which we test unrelated aspects. That makes the test code more complicated than necessary. One example, is this from the other PR I linked above. In other words, we should probably try not only to migrate to Mockito but also to refactor a bit the unit tests where it makes sense to decrease the verifications on the mocks and thus the complexity of the tests. However, we should try to keep the refactorings at a minimum.

clolov · 2023-04-18T14:36:42Z

Okay, this makes sense to me. I will aim to start opening PRs in the same manner as yours in the upcoming days and let's see where we go!

clolov · 2023-06-27T13:05:33Z

I will close this PR as we have taken a different approach. Please refer to PRs linking to this one in order to find out how we have implemented the changes.

cadonna added the streams label Sep 8, 2022

guozhangwang reviewed Sep 8, 2022

View reviewed changes

cadonna mentioned this pull request Sep 14, 2022

KAFKA-12689: Remove exactly_once / exactly_once_beta lucasbru/kafka#1

Merged

3 tasks

clolov force-pushed the KAFKA-14133-PART4 branch from b62306f to aad7ae5 Compare November 24, 2022 10:46

clolov force-pushed the KAFKA-14133-PART4 branch from aad7ae5 to b19d229 Compare December 1, 2022 15:30

mjsax added the tests Test fixes (including flaky tests) label Dec 28, 2022

clolov added 2 commits February 2, 2023 13:16

Migrating newest changes to Mockito

084190c

Migrating older changes to Mockito

24f9db0

clolov force-pushed the KAFKA-14133-PART4 branch from b19d229 to 24f9db0 Compare February 2, 2023 13:18

cadonna reviewed Apr 4, 2023

View reviewed changes

This was referenced Jun 5, 2023

KAFKA-14133: Migrate StandbyTaskCreator mock in TaskManagerTest to Mockito #13711

Merged

KAFKA-14133: Migrate Admin mock in TaskManagerTest to Mockito #13712

Merged

This was referenced Jun 19, 2023

KAFKA-14133: Migrate Consumer mock in TaskManagerTest to Mockito #13873

Open

KAFKA-14133: Migrate various mocks in TaskManagerTest to Mockito #13874

Merged

TaklaGerguis mentioned this pull request Jun 21, 2023

KAFKA-14133: Migrate StateDirectory mock in TaskManagerTest to Mockito #13897

Open

clolov closed this Jun 27, 2023

clolov mentioned this pull request Oct 10, 2023

KAFKA-14684 Replace EasyMock and PowerMock with Mockito in WorkerSinkTaskThreadedTest #14505

Merged

3 tasks

		argThat(taskId -> !taskId.equals(taskToRecycle0.id()) && !taskId.equals(taskToRecycle1.id())))
		argThat(taskId -> !taskId.equals(taskToRecycle0.id()) && !taskId.equals(taskToRecycle1.id())))

		@@ -1268,16 +1188,14 @@ public void shouldCloseDirtyRestoredTask() {
		when(stateUpdater.restoresActiveTasks()).thenReturn(true);
		activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(statefulTask.id());

KAFKA-14133: Replace EasyMock with Mockito in streams tests #12607

KAFKA-14133: Replace EasyMock with Mockito in streams tests #12607

Conversation

clolov commented Sep 8, 2022

clolov commented Sep 8, 2022

clolov commented Sep 8, 2022

clolov commented Sep 8, 2022

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clolov Sep 12, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clolov Sep 12, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

clolov commented Sep 9, 2022

clolov commented Nov 24, 2022

clolov commented Feb 2, 2023

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna commented Apr 8, 2023

clolov commented Apr 12, 2023

cadonna commented Apr 13, 2023 • edited

clolov commented Apr 18, 2023

clolov commented Jun 27, 2023

clolov Sep 12, 2022 •

edited

clolov Sep 12, 2022 •

edited

cadonna commented Apr 13, 2023 •

edited