diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java index 036e2ef7924e9..b2379795a14d9 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java @@ -1944,10 +1944,7 @@ public void shouldReAddRevivedTasksToStateUpdater() { @Test public void shouldReviveCorruptTasks() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - EasyMock.expectLastCall().once(); - replay(stateManager); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final AtomicBoolean enforcedCheckpoint = new AtomicBoolean(false); final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager) { @@ -1980,15 +1977,13 @@ public void postCommit(final boolean enforceCheckpoint) { assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00))); assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap()); - verify(stateManager); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test public void shouldReviveCorruptTasksEvenIfTheyCannotCloseClean() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - replay(stateManager); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager) { @Override @@ -2015,15 +2010,13 @@ public void suspend() { assertThat(taskManager.activeTaskMap(), is(singletonMap(taskId00, task00))); assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap()); - verify(stateManager); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - replay(stateManager); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask corruptedTask = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); final StateMachineTask nonCorruptedTask = new StateMachineTask(taskId01, taskId01Partitions, true, stateManager); @@ -2055,13 +2048,12 @@ public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() { assertThat(corruptedTask.partitionsForOffsetReset, equalTo(taskId00Partitions)); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test public void shouldNotCommitNonRunningNonCorruptedTasks() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - replay(stateManager); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask corruptedTask = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); final StateMachineTask nonRunningNonCorruptedTask = new StateMachineTask(taskId01, taskId01Partitions, true, stateManager); @@ -2088,13 +2080,12 @@ public void shouldNotCommitNonRunningNonCorruptedTasks() { assertFalse(nonRunningNonCorruptedTask.commitPrepared); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test public void shouldCleanAndReviveCorruptedStandbyTasksBeforeCommittingNonCorruptedTasks() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - replay(stateManager); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask corruptedStandby = new StateMachineTask(taskId00, taskId00Partitions, false, stateManager); final StateMachineTask runningNonCorruptedActive = new StateMachineTask(taskId01, taskId01Partitions, true, stateManager) { @@ -2129,11 +2120,12 @@ public Map prepareCommit() { assertThat(corruptedStandby.commitPrepared, is(true)); assertThat(corruptedStandby.state(), is(Task.State.CREATED)); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test public void shouldNotAttemptToCommitInHandleCorruptedDuringARebalance() { - final ProcessorStateManager stateManager = EasyMock.createNiceMock(ProcessorStateManager.class); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); expect(stateDirectory.listNonEmptyTaskDirectories()).andStubReturn(new ArrayList<>()); final StateMachineTask corruptedActive = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); @@ -2154,7 +2146,7 @@ public void shouldNotAttemptToCommitInHandleCorruptedDuringARebalance() { expect(consumer.assignment()).andStubReturn(union(HashSet::new, taskId00Partitions, taskId01Partitions)); - replay(consumer, stateDirectory, stateManager); + replay(consumer, stateDirectory); uncorruptedActive.setCommittableOffsetsAndMetadata(offsets); @@ -2180,10 +2172,8 @@ public void shouldNotAttemptToCommitInHandleCorruptedDuringARebalance() { } @Test - public void shouldCloseAndReviveUncorruptedTasksWhenTimeoutExceptionThrownFromCommitWithALSO() { - final ProcessorStateManager stateManager = EasyMock.createStrictMock(ProcessorStateManager.class); - stateManager.markChangelogAsCorrupted(taskId00Partitions); - replay(stateManager); + public void shouldCloseAndReviveUncorruptedTasksWhenTimeoutExceptionThrownFromCommitWithALOS() { + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask corruptedActive = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); final StateMachineTask uncorruptedActive = new StateMachineTask(taskId01, taskId01Partitions, true, stateManager) { @@ -2240,6 +2230,7 @@ public void markChangelogAsCorrupted(final Collection partitions assertThat(corruptedActive.state(), is(Task.State.CREATED)); assertThat(uncorruptedActive.state(), is(Task.State.CREATED)); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00Partitions); } @Test @@ -2247,7 +2238,7 @@ public void shouldCloseAndReviveUncorruptedTasksWhenTimeoutExceptionThrownFromCo final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_V2, false); final StreamsProducer producer = mock(StreamsProducer.class); when(activeTaskCreator.threadProducer()).thenReturn(producer); - final ProcessorStateManager stateManager = EasyMock.createMock(ProcessorStateManager.class); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final AtomicBoolean corruptedTaskChangelogMarkedAsCorrupted = new AtomicBoolean(false); final StateMachineTask corruptedActiveTask = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager) { @@ -2257,7 +2248,6 @@ public void markChangelogAsCorrupted(final Collection partitions corruptedTaskChangelogMarkedAsCorrupted.set(true); } }; - stateManager.markChangelogAsCorrupted(taskId00ChangelogPartitions); final AtomicBoolean uncorruptedTaskChangelogMarkedAsCorrupted = new AtomicBoolean(false); final StateMachineTask uncorruptedActiveTask = new StateMachineTask(taskId01, taskId01Partitions, true, stateManager) { @@ -2269,7 +2259,6 @@ public void markChangelogAsCorrupted(final Collection partitions }; final Map offsets = singletonMap(t1p1, new OffsetAndMetadata(0L, null)); uncorruptedActiveTask.setCommittableOffsetsAndMetadata(offsets); - stateManager.markChangelogAsCorrupted(taskId01ChangelogPartitions); // handleAssignment final Map> assignment = new HashMap<>(); @@ -2287,7 +2276,7 @@ public void markChangelogAsCorrupted(final Collection partitions expect(consumer.assignment()).andStubReturn(union(HashSet::new, taskId00Partitions, taskId01Partitions)); - replay(consumer, stateManager); + replay(consumer); taskManager.handleAssignment(assignment, emptyMap()); assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); @@ -2324,6 +2313,8 @@ public void markChangelogAsCorrupted(final Collection partitions assertThat(corruptedTaskChangelogMarkedAsCorrupted.get(), is(true)); assertThat(uncorruptedTaskChangelogMarkedAsCorrupted.get(), is(true)); verify(consumer); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00ChangelogPartitions); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId01ChangelogPartitions); } @Test @@ -2384,7 +2375,7 @@ public void shouldCloseAndReviveUncorruptedTasksWhenTimeoutExceptionThrownFromCo final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_V2, false); final StreamsProducer producer = mock(StreamsProducer.class); when(activeTaskCreator.threadProducer()).thenReturn(producer); - final ProcessorStateManager stateManager = EasyMock.createMock(ProcessorStateManager.class); + final ProcessorStateManager stateManager = Mockito.mock(ProcessorStateManager.class); final StateMachineTask revokedActiveTask = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); final Map revokedActiveTaskOffsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null)); @@ -2409,9 +2400,6 @@ public void markChangelogAsCorrupted(final Collection partitions expectedCommittedOffsets.putAll(revokedActiveTaskOffsets); expectedCommittedOffsets.putAll(unrevokedTaskOffsets); - stateManager.markChangelogAsCorrupted(taskId00ChangelogPartitions); - stateManager.markChangelogAsCorrupted(taskId01ChangelogPartitions); - final Map> assignmentActive = mkMap( mkEntry(taskId00, taskId00Partitions), mkEntry(taskId01, taskId01Partitions), @@ -2430,7 +2418,7 @@ public void markChangelogAsCorrupted(final Collection partitions expect(consumer.assignment()).andStubReturn(union(HashSet::new, taskId00Partitions, taskId01Partitions, taskId02Partitions)); - replay(consumer, stateManager); + replay(consumer); taskManager.handleAssignment(assignmentActive, emptyMap()); assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true)); @@ -2449,6 +2437,8 @@ public void markChangelogAsCorrupted(final Collection partitions assertThat(revokedActiveTask.state(), is(State.SUSPENDED)); assertThat(unrevokedActiveTask.state(), is(State.CREATED)); assertThat(unrevokedActiveTaskWithoutCommitNeeded.state(), is(State.RUNNING)); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId00ChangelogPartitions); + Mockito.verify(stateManager).markChangelogAsCorrupted(taskId01ChangelogPartitions); } @Test @@ -3524,24 +3514,23 @@ public void shouldCommitViaConsumerIfEosDisabled() { @Test public void shouldCommitViaProducerIfEosAlphaEnabled() { - final StreamsProducer producer = EasyMock.mock(StreamsProducer.class); + final StreamsProducer producer = Mockito.mock(StreamsProducer.class); when(activeTaskCreator.streamsProducerForTask(any(TaskId.class))) .thenReturn(producer); final Map offsetsT01 = singletonMap(t1p1, new OffsetAndMetadata(0L, null)); final Map offsetsT02 = singletonMap(t1p2, new OffsetAndMetadata(1L, null)); - producer.commitTransaction(offsetsT01, new ConsumerGroupMetadata("appId")); - expectLastCall(); - producer.commitTransaction(offsetsT02, new ConsumerGroupMetadata("appId")); - expectLastCall(); + shouldCommitViaProducerIfEosEnabled(ProcessingMode.EXACTLY_ONCE_ALPHA, offsetsT01, offsetsT02); - shouldCommitViaProducerIfEosEnabled(ProcessingMode.EXACTLY_ONCE_ALPHA, producer, offsetsT01, offsetsT02); + Mockito.verify(producer).commitTransaction(offsetsT01, new ConsumerGroupMetadata("appId")); + Mockito.verify(producer).commitTransaction(offsetsT02, new ConsumerGroupMetadata("appId")); + Mockito.verifyNoMoreInteractions(producer); } @Test public void shouldCommitViaProducerIfEosV2Enabled() { - final StreamsProducer producer = EasyMock.mock(StreamsProducer.class); + final StreamsProducer producer = Mockito.mock(StreamsProducer.class); when(activeTaskCreator.threadProducer()).thenReturn(producer); final Map offsetsT01 = singletonMap(t1p1, new OffsetAndMetadata(0L, null)); @@ -3550,14 +3539,13 @@ public void shouldCommitViaProducerIfEosV2Enabled() { allOffsets.putAll(offsetsT01); allOffsets.putAll(offsetsT02); - producer.commitTransaction(allOffsets, new ConsumerGroupMetadata("appId")); - expectLastCall(); + shouldCommitViaProducerIfEosEnabled(ProcessingMode.EXACTLY_ONCE_V2, offsetsT01, offsetsT02); - shouldCommitViaProducerIfEosEnabled(ProcessingMode.EXACTLY_ONCE_V2, producer, offsetsT01, offsetsT02); + Mockito.verify(producer).commitTransaction(allOffsets, new ConsumerGroupMetadata("appId")); + Mockito.verifyNoMoreInteractions(producer); } private void shouldCommitViaProducerIfEosEnabled(final ProcessingMode processingMode, - final StreamsProducer producer, final Map offsetsT01, final Map offsetsT02) { final TaskManager taskManager = setUpTaskManager(processingMode, false); @@ -3573,11 +3561,11 @@ private void shouldCommitViaProducerIfEosEnabled(final ProcessingMode processing reset(consumer); expect(consumer.groupMetadata()).andStubReturn(new ConsumerGroupMetadata("appId")); - replay(consumer, producer); + replay(consumer); taskManager.commitAll(); - verify(producer, consumer); + verify(consumer); } @Test @@ -4463,21 +4451,18 @@ public void suspend() { @Test public void shouldConvertActiveTaskToStandbyTask() { - final StreamTask activeTask = EasyMock.mock(StreamTask.class); - expect(activeTask.id()).andStubReturn(taskId00); - expect(activeTask.inputPartitions()).andStubReturn(taskId00Partitions); - expect(activeTask.isActive()).andStubReturn(true); - expect(activeTask.prepareCommit()).andStubReturn(Collections.emptyMap()); + final StreamTask activeTask = Mockito.mock(StreamTask.class); + when(activeTask.id()).thenReturn(taskId00); + when(activeTask.inputPartitions()).thenReturn(taskId00Partitions); + when(activeTask.isActive()).thenReturn(true); - final StandbyTask standbyTask = EasyMock.mock(StandbyTask.class); - expect(standbyTask.id()).andStubReturn(taskId00); + final StandbyTask standbyTask = Mockito.mock(StandbyTask.class); + when(standbyTask.id()).thenReturn(taskId00); when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(activeTask)); - activeTask.prepareRecycle(); - expectLastCall().once(); when(standbyTaskCreator.createStandbyTaskFromActive(Mockito.any(), Mockito.eq(taskId00Partitions))).thenReturn(standbyTask); - replay(activeTask, standbyTask, consumer); + replay(consumer); taskManager.handleAssignment(taskId00Assignment, Collections.emptyMap()); taskManager.handleAssignment(Collections.emptyMap(), taskId00Assignment);