clolov commented on code in PR #15112:
URL: https://github.com/apache/kafka/pull/15112#discussion_r1440569384
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -197,6 +196,8 @@ public class TaskManagerTest {
@Mock(type = MockType.STRICT)
private Consumer<byte[], byte[]> consumer;
@org.mockito.Mock
+ private Consumer<byte[], byte[]> mockitoConsumer;
Review Comment:
I ran into quite a lot of problems when I tried migrating the whole mock, so
I decided to do the migration test-by-test. This way problems could be flushed
out. By introducing this mock and using `setMainConsumer` on a test-by-test
basis things are easier to go through (at least in my opinion 😊)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4809,11 +4768,14 @@ public void
shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
exception.corruptedTasks(),
equalTo(Collections.singleton(taskId00))
);
+
+ Mockito.verify(mockitoConsumer, times(2)).groupMetadata();
Review Comment:
Mockito claims the method is called twice. This wasn't specified in EasyMock
world, but I decided to make it explicit now.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3010,13 +2993,9 @@ public void initializeIfNeeded() {
}
};
- consumer.commitSync(Collections.emptyMap());
- expectLastCall();
- expect(consumer.assignment()).andReturn(emptySet());
- consumer.resume(eq(emptySet()));
- expectLastCall();
Review Comment:
Same as above, Mockito reported these are unused. I added an assertion in
Mockito world at the end (...I should probably add one in the first test as
well, but I can do this in part 2 of this pull request)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1168,13 +1166,10 @@ public void
shouldHandleMultipleRemovedTasksFromStateUpdater() {
when(stateUpdater.drainRemovedTasks())
.thenReturn(mkSet(taskToRecycle0, taskToRecycle1, taskToClose,
taskToUpdateInputPartitions));
when(stateUpdater.restoresActiveTasks()).thenReturn(true);
- when(activeTaskCreator.createActiveTaskFromStandby(taskToRecycle1,
taskId01Partitions, consumer))
+ when(activeTaskCreator.createActiveTaskFromStandby(taskToRecycle1,
taskId01Partitions, mockitoConsumer))
.thenReturn(convertedTask1);
when(standbyTaskCreator.createStandbyTaskFromActive(taskToRecycle0,
taskId00Partitions))
.thenReturn(convertedTask0);
- expect(consumer.assignment()).andReturn(emptySet()).anyTimes();
- consumer.resume(anyObject());
- expectLastCall().anyTimes();
Review Comment:
According to Mockito these were unused. I deleted them in EasyMock world and
the test still passed. Hence I removed them.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3047,14 +3027,9 @@ public void completeRestoration(final
java.util.function.Consumer<Set<TopicParti
}
};
- consumer.commitSync(Collections.emptyMap());
- expectLastCall();
- expect(consumer.assignment()).andReturn(emptySet());
- consumer.resume(eq(emptySet()));
- expectLastCall();
- expectLastCall();
Review Comment:
Ditto
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3664,20 +3639,11 @@ public void shouldCloseStandbyTasksOnShutdown() {
final Map<TaskId, Set<TopicPartition>> assignment =
singletonMap(taskId00, taskId00Partitions);
final Task task00 = new StateMachineTask(taskId00, taskId00Partitions,
false, stateManager);
+ taskManager.setMainConsumer(mockitoConsumer);
+
// `handleAssignment`
when(standbyTaskCreator.createTasks(assignment)).thenReturn(singletonList(task00));
- // `tryToCompleteRestoration`
- expect(consumer.assignment()).andReturn(emptySet());
- consumer.resume(eq(emptySet()));
- expectLastCall();
-
- // `shutdown`
- consumer.commitSync(Collections.emptyMap());
- expectLastCall();
Review Comment:
Ditto
--
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
