cadonna commented on code in PR #14716:
URL: https://github.com/apache/kafka/pull/14716#discussion_r1552378325
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ stateDirectory = mock(StateDirectory.class);
- stateManager.registerGlobalStateStores(emptyList());
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).registerGlobalStateStores(emptyList());
Review Comment:
Do we really need this stub?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ stateDirectory = mock(StateDirectory.class);
- stateManager.registerGlobalStateStores(emptyList());
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).registerGlobalStateStores(emptyList());
- EasyMock.expect(stateManager.taskId()).andReturn(taskId);
+ when(stateManager.taskId()).thenReturn(taskId);
- EasyMock.expect(stateDirectory.lock(taskId)).andReturn(true);
+ when(stateDirectory.lock(taskId)).thenReturn(true);
- stateManager.close();
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).close();
Review Comment:
Do we really need this stub?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ stateDirectory = mock(StateDirectory.class);
- stateManager.registerGlobalStateStores(emptyList());
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).registerGlobalStateStores(emptyList());
- EasyMock.expect(stateManager.taskId()).andReturn(taskId);
+ when(stateManager.taskId()).thenReturn(taskId);
- EasyMock.expect(stateDirectory.lock(taskId)).andReturn(true);
+ when(stateDirectory.lock(taskId)).thenReturn(true);
- stateManager.close();
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).close();
// The `baseDir` will be accessed when attempting to delete the state
store.
-
EasyMock.expect(stateManager.baseDir()).andReturn(TestUtils.tempDirectory("state_store"));
+
when(stateManager.baseDir()).thenReturn(TestUtils.tempDirectory("state_store"));
- stateDirectory.unlock(taskId);
- EasyMock.expectLastCall();
+ doNothing().when(stateDirectory).unlock(taskId);
Review Comment:
Do we really need this stub?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1903,42 +1844,36 @@ public void shouldThrowIfPostCommittingOnIllegalState()
{
@Test
public void shouldSkipCheckpointingSuspendedCreatedTask() {
- stateManager.checkpoint();
- EasyMock.expectLastCall().andThrow(new AssertionError("Should not have
tried to checkpoint"));
-
EasyMock.expect(recordCollector.offsets()).andReturn(emptyMap()).anyTimes();
- EasyMock.replay(stateManager, recordCollector);
+ when(recordCollector.offsets()).thenReturn(emptyMap());
Review Comment:
Is this stub needed since it returns an empty map?
Please also check other occurrences of such stubs.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -365,11 +356,10 @@ public void
shouldResetOffsetsToLastCommittedForSpecifiedPartitions() {
consumer.seek(partition1, 10L);
consumer.seek(partition2, 15L);
+ @SuppressWarnings("unchecked")
final java.util.function.Consumer<Set<TopicPartition>> resetter =
- EasyMock.mock(java.util.function.Consumer.class);
- resetter.accept(Collections.emptySet());
- EasyMock.expectLastCall();
- EasyMock.replay(resetter);
+ mock(java.util.function.Consumer.class);
+ doNothing().when(resetter).accept(Collections.emptySet());
Review Comment:
Again, I do not think we need this stub with Mockito.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1568,10 +1548,8 @@ public void
shouldNotShareHeadersBetweenPunctuateIterations() {
@Test
public void shouldWrapKafkaExceptionWithStreamsExceptionWhenProcess() {
-
EasyMock.expect(stateManager.changelogPartitions()).andReturn(Collections.emptySet()).anyTimes();
-
EasyMock.expect(stateManager.changelogOffsets()).andReturn(emptyMap()).anyTimes();
-
EasyMock.expect(recordCollector.offsets()).andReturn(emptyMap()).anyTimes();
- EasyMock.replay(stateManager, recordCollector);
+ when(stateManager.changelogOffsets()).thenReturn(emptyMap());
+ when(recordCollector.offsets()).thenReturn(emptyMap());
Review Comment:
Do we even need those? What is the default return value for a map? I guess
an empty map. Could you please verify?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ stateDirectory = mock(StateDirectory.class);
- stateManager.registerGlobalStateStores(emptyList());
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).registerGlobalStateStores(emptyList());
- EasyMock.expect(stateManager.taskId()).andReturn(taskId);
+ when(stateManager.taskId()).thenReturn(taskId);
Review Comment:
Do we really need to redefine this stub?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
Review Comment:
Do we really need to redefine this mock?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -309,49 +300,49 @@ public void
shouldThrowLockExceptionIfFailedToLockStateDirectory() throws IOExce
@Test
public void shouldNotAttemptToLockIfNoStores() {
- stateDirectory = EasyMock.createNiceMock(StateDirectory.class);
- EasyMock.replay(stateDirectory);
+ stateDirectory = mock(StateDirectory.class);
task = createStatelessTask(createConfig("100"));
task.initializeIfNeeded();
// should fail if lock is called
- EasyMock.verify(stateDirectory);
+ verify(stateDirectory, never()).lock(any());
}
@Test
public void
shouldAttemptToDeleteStateDirectoryWhenCloseDirtyAndEosEnabled() throws
IOException {
- final IMocksControl ctrl = EasyMock.createStrictControl();
- final ProcessorStateManager stateManager =
ctrl.createMock(ProcessorStateManager.class);
-
EasyMock.expect(stateManager.taskType()).andStubReturn(TaskType.ACTIVE);
- stateDirectory = ctrl.createMock(StateDirectory.class);
+ final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
+ when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
+ stateDirectory = mock(StateDirectory.class);
- stateManager.registerGlobalStateStores(emptyList());
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).registerGlobalStateStores(emptyList());
- EasyMock.expect(stateManager.taskId()).andReturn(taskId);
+ when(stateManager.taskId()).thenReturn(taskId);
- EasyMock.expect(stateDirectory.lock(taskId)).andReturn(true);
+ when(stateDirectory.lock(taskId)).thenReturn(true);
- stateManager.close();
- EasyMock.expectLastCall();
+ doNothing().when(stateManager).close();
// The `baseDir` will be accessed when attempting to delete the state
store.
-
EasyMock.expect(stateManager.baseDir()).andReturn(TestUtils.tempDirectory("state_store"));
+
when(stateManager.baseDir()).thenReturn(TestUtils.tempDirectory("state_store"));
- stateDirectory.unlock(taskId);
- EasyMock.expectLastCall();
+ doNothing().when(stateDirectory).unlock(taskId);
- ctrl.checkOrder(true);
- ctrl.replay();
+ final InOrder inOrder = inOrder(stateManager, stateDirectory);
task = createStatefulTask(createConfig(StreamsConfig.EXACTLY_ONCE_V2,
"100"), true, stateManager);
task.suspend();
task.closeDirty();
task = null;
- ctrl.verify();
+ inOrder.verify(stateManager).taskType();
+ inOrder.verify(stateManager).registerGlobalStateStores(emptyList());
+ inOrder.verify(stateManager).taskId();
+ inOrder.verify(stateDirectory).lock(taskId);
+ inOrder.verify(stateManager).close();
+ inOrder.verify(stateManager).baseDir();
+ inOrder.verify(stateDirectory).unlock(taskId);
Review Comment:
I think these verifications cover the `doNothing()...` stubs from above and
we do not really need those stubs. Could you verify?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -1798,8 +1746,6 @@ public void
shouldCloseStateManagerEvenDuringFailureOnUncleanTaskClose() {
assertThrows(RuntimeException.class, () -> task.suspend());
task.closeDirty();
-
- EasyMock.verify(stateManager);
Review Comment:
Verifications are missing in the new code. For example,
`stateManager.close()`.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -514,8 +496,6 @@ public void
shouldTransitToRestoringThenRunningAfterCreation() throws IOExceptio
assertEquals(RUNNING, task.state());
assertTrue(source1.initialized);
assertTrue(source2.initialized);
-
- EasyMock.verify(stateDirectory);
Review Comment:
I think some verifications are missing here, aren't they? For example,
`stateManager.registerStore()`.
--
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
