You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mehbey (via GitHub)" <gi...@apache.org> on 2023/05/08 01:07:38 UTC
[GitHub] [kafka] mehbey opened a new pull request, #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
mehbey opened a new pull request, #13681:
URL: https://github.com/apache/kafka/pull/13681
This pull requests migrates the ActiveTaskCreator mock in TaskManagerTest from EasyMock to Mockito
The change is restricted to a single mock to minimize the scope and make it easier for review.
Please see two examples that follows the same pattern below:
https://github.com/apache/kafka/pull/13529
https://github.com/apache/kafka/pull/13621
### Summary of testing strategy (including rationale)
Verified that all tests are passing
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1212275329
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3137,13 +3102,11 @@ public Set<TopicPartition> changelogPartitions() {
}
};
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(singletonList(task00));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId00));
- expectLastCall();
- activeTaskCreator.closeThreadProducerIfNeeded();
- expectLastCall().andThrow(new RuntimeException("whatever"));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment))).thenReturn(singletonList(task00));
+ doThrow(new RuntimeException("whatever")).when(activeTaskCreator).closeThreadProducerIfNeeded();
expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
- replay(activeTaskCreator, standbyTaskCreator);
+ expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
Review Comment:
removed duplicate
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1546424538
addressed all comments from @clolov.
@cadonna - when you get a chance would you be able to review?
--
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
[GitHub] [kafka] divijvaidya commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1187177289
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
Review Comment:
we still seem to have easymock imports. Can you please remove them and replace with Mockito as well?
```
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.resetToStrict;
import static org.easymock.EasyMock.verify;
````
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1212257493
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4448,10 +4396,10 @@ public void shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_ALPHA, tasks, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer)
Review Comment:
updated, thank you
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1539231000
Verified that failing Testing are not related to this change
```
[2023-05-08T22:22:09.340Z] 1: Task failed with an exception.
[2023-05-08T22:22:09.340Z] -----------
[2023-05-08T22:22:09.340Z] * What went wrong:
[2023-05-08T22:22:09.340Z] Execution failed for task ':streams:upgrade-system-tests-0102:integrationTest'.
[2023-05-08T22:22:09.340Z] > Process 'Gradle Test Executor 151' finished with non-zero exit value 1
[2023-05-08T22:22:09.340Z] This problem might be caused by incorrect test process configuration.
[2023-05-08T22:22:09.340Z] Please refer to the test execution section in the User Manual at https://docs.gradle.org/8.1.1/userguide/java_testing.html#sec:test_execution
[2023-05-08T22:22:09.340Z]
[2023-05-08T22:22:09.340Z] * Try:
[2023-05-08T22:22:09.340Z] > Run with --stacktrace option to get the stack trace.
[2023-05-08T22:22:09.340Z] > Run with --info or --debug option to get more log output.
[2023-05-08T22:22:09.340Z] > Run with --scan to get full insights.
[2023-05-08T22:22:09.340Z] ==============================================================================
[2023-05-08T22:22:09.340Z]
[2023-05-08T22:22:09.340Z] 2: Task failed with an exception.
[2023-05-08T22:22:09.340Z] -----------
[2023-05-08T22:22:09.340Z] * What went wrong:
[2023-05-08T22:22:09.340Z] Execution failed for task ':connect:mirror:integrationTest'.
[2023-05-08T22:22:09.340Z] > Process 'Gradle Test Executor 130' finished with non-zero exit value 137
[2023-05-08T22:22:09.340Z] This problem might be caused by incorrect test process configuration.
[2023-05-08T22:22:09.340Z] Please refer to the test execution section in the User Manual at https://docs.gradle.org/8.1.1/userguide/java_testing.html#sec:test_execution
[2023-05-08T22:22:09.340Z]
[2023-05-08T22:22:09.340Z] * Try:
[2023-05-08T22:22:09.340Z] > Run with --stacktrace option to get the stack trace.
[2023-05-08T22:22:09.340Z] > Run with --info or --debug option to get more log output.
[2023-05-08T22:22:09.340Z] > Run with --scan to get full insights.
```
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1548206878
@machi1990, thank you for the review. I have addressed your comment.
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1192860199
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
+ Mockito.eq(consumer))).thenReturn(task01Converted);
expect(standbyTaskCreator.createStandbyTaskFromActive(eq(task00), eq(taskId00Partitions)))
.andStubReturn(task00Converted);
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator, times(1)).closeAndRemoveTaskProducerIfNeeded(any());
Review Comment:
Yes make sense - updated
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1212265470
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
updated, thank you.
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1187611572
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2579,7 +2563,9 @@ public void shouldUpdateInputPartitionsAfterRebalance() {
assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
assertEquals(newPartitionsSet, task00.inputPartitions());
- verify(activeTaskCreator, consumer, changeLogReader);
+ verify(consumer, changeLogReader);
+ Mockito.verify(activeTaskCreator).createTasks(any(), Mockito.eq(taskId00Assignment));
Review Comment:
yeah good recommendation, I will push a new version with the update
--
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
[GitHub] [kafka] clolov commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1191109013
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
Review Comment:
```suggestion
when(activeTaskCreator.createActiveTaskFromStandby(task01, taskId01Partitions,
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -879,13 +880,10 @@ public void shouldCloseTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToCloseClean(task00.id())).thenReturn(true);
when(tasks.removePendingTaskToCloseClean(task01.id())).thenReturn(true);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
- replay(activeTaskCreator);
taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
- verify(activeTaskCreator);
+ Mockito.verify(activeTaskCreator, times(1)).closeAndRemoveTaskProducerIfNeeded(any());
Review Comment:
```suggestion
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(any());
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -960,10 +958,8 @@ public void shouldHandleMultipleRemovedTasksFromStateUpdater() {
when(stateUpdater.drainRemovedTasks())
.thenReturn(mkSet(taskToRecycle0, taskToRecycle1, taskToClose, taskToUpdateInputPartitions));
when(stateUpdater.restoresActiveTasks()).thenReturn(true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(taskToRecycle1), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(convertedTask1);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().times(2);
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(taskToRecycle1), Mockito.eq(taskId01Partitions),
Review Comment:
```suggestion
when(activeTaskCreator.createActiveTaskFromStandby(taskToRecycle1, taskId01Partitions,
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
+ Mockito.eq(consumer))).thenReturn(task01Converted);
expect(standbyTaskCreator.createStandbyTaskFromActive(eq(task00), eq(taskId00Partitions)))
.andStubReturn(task00Converted);
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.checkStateUpdater(time.milliseconds(), noOpResetter);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator, times(1)).closeAndRemoveTaskProducerIfNeeded(any());
Review Comment:
```suggestion
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(any());
```
The default check of verify is times(1) as far as I know.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2162,11 +2140,12 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit() {
// handleAssignment
expect(standbyTaskCreator.createTasks(eq(taskId00Assignment))).andStubReturn(singleton(corruptedStandby));
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId01Assignment))).andStubReturn(singleton(runningNonCorruptedActive));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId01Assignment)))
+ .thenReturn(singleton(runningNonCorruptedActive));
Review Comment:
```suggestion
.thenReturn(singleton(runningNonCorruptedActive));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2082,17 +2062,16 @@ public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() {
assignment.putAll(taskId01Assignment);
// `handleAssignment`
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
- .andStubReturn(asList(corruptedTask, nonCorruptedTask));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedTask, nonCorruptedTask));
Review Comment:
```suggestion
.thenReturn(asList(corruptedTask, nonCorruptedTask));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2202,14 +2181,15 @@ public void shouldNotAttemptToCommitInHandleCorruptedDuringARebalance() {
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.putAll(taskId00Assignment);
assignment.putAll(taskId01Assignment);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(asList(corruptedActive, uncorruptedActive));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedActive, uncorruptedActive));
Review Comment:
```suggestion
.thenReturn(asList(corruptedActive, uncorruptedActive));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2082,17 +2062,16 @@ public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() {
assignment.putAll(taskId01Assignment);
// `handleAssignment`
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
- .andStubReturn(asList(corruptedTask, nonCorruptedTask));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedTask, nonCorruptedTask));
Review Comment:
```suggestion
.thenReturn(asList(corruptedTask, nonCorruptedTask));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2412,15 +2394,15 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive))).andReturn(asList(revokedActiveTask, unrevokedActiveTaskWithCommitNeeded, unrevokedActiveTaskWithoutCommitNeeded));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(asList(revokedActiveTask, unrevokedActiveTaskWithCommitNeeded, unrevokedActiveTaskWithoutCommitNeeded));
Review Comment:
```suggestion
.thenReturn(asList(revokedActiveTask, unrevokedActiveTaskWithCommitNeeded, unrevokedActiveTaskWithoutCommitNeeded));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2750,11 +2726,10 @@ public void shouldCommitAllActiveTasksThatNeedCommittingOnHandleRevocationWithEo
);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive)))
- .andReturn(asList(task00, task01, task02));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(asList(task00, task01, task02));
Review Comment:
```suggestion
.thenReturn(asList(task00, task01, task02));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3087,13 +3053,11 @@ public Set<TopicPartition> changelogPartitions() {
final Map<TopicPartition, OffsetAndMetadata> offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));
task00.setCommittableOffsetsAndMetadata(offsets);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(singletonList(task00));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId00));
- expectLastCall().andThrow(new RuntimeException("whatever"));
- activeTaskCreator.closeThreadProducerIfNeeded();
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment))).thenReturn(singletonList(task00));
+ doThrow(new RuntimeException("whatever"))
+ .when(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(Mockito.eq(taskId00));
Review Comment:
```suggestion
.when(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3016,14 +2985,10 @@ public void closeDirty() {
}
};
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
- .andStubReturn(asList(task00, task01, task02, task03));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().times(4);
- activeTaskCreator.closeThreadProducerIfNeeded();
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(task00, task01, task02, task03));
Review Comment:
```suggestion
.thenReturn(asList(task00, task01, task02, task03));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2822,16 +2797,14 @@ public void shouldCommitAllNeededTasksOnHandleRevocation() {
);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive)))
- .andReturn(asList(task00, task01, task02));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(taskId00);
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(asList(task00, task01, task02));
Review Comment:
```suggestion
.thenReturn(asList(task00, task01, task02));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3120,8 +3084,9 @@ public Set<TopicPartition> changelogPartitions() {
assertThat(exception.getCause().getMessage(), is("whatever"));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
+ Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(Mockito.eq(taskId00));
Review Comment:
```suggestion
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3484,14 +3435,14 @@ public void shouldCommitActiveAndStandbyTasks() {
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, false);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment)))
- .andStubReturn(singletonList(task00));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment)))
+ .thenReturn(singletonList(task00));
Review Comment:
```suggestion
.thenReturn(singletonList(task00));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2082,17 +2062,16 @@ public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() {
assignment.putAll(taskId01Assignment);
// `handleAssignment`
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
- .andStubReturn(asList(corruptedTask, nonCorruptedTask));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedTask, nonCorruptedTask));
Review Comment:
Could you keep the indentation consistent with the rest of the file?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
Review Comment:
```suggestion
.thenReturn(producer)
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -960,10 +958,8 @@ public void shouldHandleMultipleRemovedTasksFromStateUpdater() {
when(stateUpdater.drainRemovedTasks())
.thenReturn(mkSet(taskToRecycle0, taskToRecycle1, taskToClose, taskToUpdateInputPartitions));
when(stateUpdater.restoresActiveTasks()).thenReturn(true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(taskToRecycle1), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(convertedTask1);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().times(2);
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(taskToRecycle1), Mockito.eq(taskId01Partitions),
+ Mockito.eq(consumer))).thenReturn(convertedTask1);
Review Comment:
```suggestion
consumer)).thenReturn(convertedTask1);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
+ Mockito.eq(consumer))).thenReturn(task01Converted);
Review Comment:
```suggestion
consumer)).thenReturn(task01Converted);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3528,14 +3479,14 @@ public void shouldCommitProvidedTasksIfNeeded() {
);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive)))
- .andStubReturn(Arrays.asList(task00, task01, task02));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(Arrays.asList(task00, task01, task02));
Review Comment:
```suggestion
.thenReturn(Arrays.asList(task00, task01, task02));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3587,12 +3537,12 @@ public void shouldNotCommitActiveAndStandbyTasksWhileRebalanceInProgress() throw
makeTaskFolders(taskId00.toString(), task01.toString());
expectLockObtainedFor(taskId00, taskId01);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment)))
- .andStubReturn(singletonList(task00));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment)))
+ .thenReturn(singletonList(task00));
Review Comment:
```suggestion
.thenReturn(singletonList(task00));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
```suggestion
```
You do not need to specify this multiple times. If you specify it once Mockito should take care of continuing to return the same mock on every invocation.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2125,12 +2104,12 @@ public void shouldNotCommitNonRunningNonCorruptedTasks() {
assignment.putAll(taskId01Assignment);
// `handleAssignment`
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment)))
- .andStubReturn(asList(corruptedTask, nonRunningNonCorruptedTask));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedTask, nonRunningNonCorruptedTask));
Review Comment:
```suggestion
.thenReturn(asList(corruptedTask, nonRunningNonCorruptedTask));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
Review Comment:
```suggestion
.thenReturn(producer)
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2330,7 +2311,8 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.putAll(taskId00Assignment);
assignment.putAll(taskId01Assignment);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(asList(corruptedActiveTask, uncorruptedActiveTask));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedActiveTask, uncorruptedActiveTask));
Review Comment:
```suggestion
.thenReturn(asList(corruptedActiveTask, uncorruptedActiveTask));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2254,7 +2234,8 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
final Map<TaskId, Set<TopicPartition>> assignment = new HashMap<>();
assignment.putAll(taskId00Assignment);
assignment.putAll(taskId01Assignment);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(asList(corruptedActive, uncorruptedActive));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(asList(corruptedActive, uncorruptedActive));
Review Comment:
```suggestion
.thenReturn(asList(corruptedActive, uncorruptedActive));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4493,9 +4441,9 @@ public void shouldThrowTaskCorruptedExceptionForTimeoutExceptionOnCommitWithEosV
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_V2, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.threadProducer())
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.threadProducer())
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3920,9 +3869,10 @@ public void shouldProcessActiveTasks() {
assignment.put(taskId01, taskId01Partitions);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(Arrays.asList(task00, task01));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
+ .thenReturn(Arrays.asList(task00, task01));
Review Comment:
```suggestion
.thenReturn(Arrays.asList(task00, task01));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4448,10 +4396,10 @@ public void shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_ALPHA, tasks, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer)
Review Comment:
You do not need to specify this 3 times. If you specify it once Mockito takes care of returning the same result on every invocation.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
Review Comment:
```suggestion
.thenReturn(producer)
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3171,7 +3134,7 @@ public Set<TopicPartition> changelogPartitions() {
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
// the active task creator should also get closed (so that it closes the thread producer if applicable)
- verify(activeTaskCreator);
+ Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(Mockito.eq(taskId00));
Review Comment:
```suggestion
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4448,10 +4396,10 @@ public void shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_ALPHA, tasks, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer)
Review Comment:
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -539,18 +542,15 @@ public void shouldCreateActiveTaskDuringAssignment() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
final Set<Task> createdTasks = mkSet(activeTaskToBeCreated);
- expect(activeTaskCreator.createTasks(consumer, mkMap(
- mkEntry(activeTaskToBeCreated.id(), activeTaskToBeCreated.inputPartitions())))
- ).andReturn(createdTasks);
+ final Map<TaskId, Set<TopicPartition>> tasksToBeCreated = mkMap(
+ mkEntry(activeTaskToBeCreated.id(), activeTaskToBeCreated.inputPartitions()));
Review Comment:
```suggestion
mkEntry(activeTaskToBeCreated.id(), activeTaskToBeCreated.inputPartitions()));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2476,10 +2458,9 @@ public void markChangelogAsCorrupted(final Collection<TopicPartition> partitions
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive))).andReturn(asList(revokedActiveTask, unrevokedActiveTask, unrevokedActiveTaskWithoutCommitNeeded));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(asList(revokedActiveTask, unrevokedActiveTask, unrevokedActiveTaskWithoutCommitNeeded));
Review Comment:
```suggestion
.thenReturn(asList(revokedActiveTask, unrevokedActiveTask, unrevokedActiveTaskWithoutCommitNeeded));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4635,19 +4582,18 @@ public void shouldConvertStandbyTaskToActiveTask() {
final StreamTask activeTask = mock(StreamTask.class);
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());
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(standbyTask), Mockito.eq(taskId00Partitions), any()))
+ .thenReturn(activeTask);
Review Comment:
```suggestion
.thenReturn(activeTask);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1266,16 +1259,15 @@ public void shouldHandleExceptionThrownDuringClosingTaskProducerInCloseCleanRest
.withInputPartitions(taskId00Partitions).build();
final TasksRegistry tasks = mock(TasksRegistry.class);
final TaskManager taskManager = setUpCloseCleanRestoredTask(statefulTask, tasks);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(statefulTask.id());
- expectLastCall().andThrow(new RuntimeException("Something happened"));
- replay(activeTaskCreator);
+ final TaskId taskId = statefulTask.id();
+ doThrow(new RuntimeException("Something happened"))
+ .when(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId);
Review Comment:
```suggestion
.when(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4448,10 +4396,10 @@ public void shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_ALPHA, tasks, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
```suggestion
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3137,13 +3102,11 @@ public Set<TopicPartition> changelogPartitions() {
}
};
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(singletonList(task00));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId00));
- expectLastCall();
- activeTaskCreator.closeThreadProducerIfNeeded();
- expectLastCall().andThrow(new RuntimeException("whatever"));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment))).thenReturn(singletonList(task00));
+ doThrow(new RuntimeException("whatever")).when(activeTaskCreator).closeThreadProducerIfNeeded();
expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
- replay(activeTaskCreator, standbyTaskCreator);
+ expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
Review Comment:
Isn't this the same as line 3107? If it is, is there a reason we need a copy of it? If not, can we get rid of it?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -767,20 +770,19 @@ public void shouldAssignMultipleTasksInTasksRegistryWithStateUpdaterEnabled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(tasks.allTasks()).thenReturn(mkSet(activeTaskToClose));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(activeTaskToClose.id());
- expect(activeTaskCreator.createTasks(
- consumer,
- mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
- )).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions())),
Collections.emptyMap()
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator).createTasks(
+ consumer,
Review Comment:
```suggestion
consumer,
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -767,20 +770,19 @@ public void shouldAssignMultipleTasksInTasksRegistryWithStateUpdaterEnabled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(tasks.allTasks()).thenReturn(mkSet(activeTaskToClose));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(activeTaskToClose.id());
- expect(activeTaskCreator.createTasks(
- consumer,
- mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
- )).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions())),
Collections.emptyMap()
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator).createTasks(
+ consumer,
Review Comment:
```suggestion
consumer,
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -767,20 +770,19 @@ public void shouldAssignMultipleTasksInTasksRegistryWithStateUpdaterEnabled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(tasks.allTasks()).thenReturn(mkSet(activeTaskToClose));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(activeTaskToClose.id());
- expect(activeTaskCreator.createTasks(
- consumer,
- mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
- )).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions())),
Collections.emptyMap()
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator).createTasks(
+ consumer,
+ mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
Review Comment:
```suggestion
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3428,10 +3380,10 @@ public void shouldShutDownStateUpdaterAndAddRemovedTasksToTaskRegistry() {
public void shouldInitializeNewActiveTasks() {
final StateMachineTask task00 = new StateMachineTask(taskId00, taskId00Partitions, true);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment)))
- .andStubReturn(singletonList(task00));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment)))
+ .thenReturn(singletonList(task00));
Review Comment:
```suggestion
.thenReturn(singletonList(task00));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
Review Comment:
If none of the things are captors or things which allow you to generalise you don't need to use matchers.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
```suggestion
.thenReturn(producer);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3876,14 +3825,14 @@ public void shouldMaybeCommitAllActiveTasksThatNeedCommit() {
);
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignmentActive)))
- .andStubReturn(asList(task00, task01, task02, task03));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignmentActive)))
+ .thenReturn(asList(task00, task01, task02, task03));
Review Comment:
```suggestion
.thenReturn(asList(task00, task01, task02, task03));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4058,10 +4008,10 @@ public boolean process(final long wallClockTime) {
};
expectRestoreToBeCompleted(consumer);
- expect(activeTaskCreator.createTasks(anyObject(), eq(taskId00Assignment)))
- .andStubReturn(singletonList(task00));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment)))
+ .thenReturn(singletonList(task00));
Review Comment:
```suggestion
.thenReturn(singletonList(task00));
```
--
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
[GitHub] [kafka] cadonna commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1571455056
Build failures are unrelated:
```
Build / JDK 17 and Scala 2.13 / kafka.admin.TopicCommandIntegrationTest.testDescribeAndListTopicsWithoutInternalTopics(String).quorum=kraft
Build / JDK 17 and Scala 2.13 / org.apache.kafka.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest.testReplicateSourceDefault()
Build / JDK 11 and Scala 2.13 / integration.kafka.server.FetchFromFollowerIntegrationTest.testRackAwareRangeAssignor()
Build / JDK 11 and Scala 2.13 / org.apache.kafka.controller.QuorumControllerTest.testBalancePartitionLeaders()
```
--
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
[GitHub] [kafka] cadonna merged pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #13681:
URL: https://github.com/apache/kafka/pull/13681
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1192859892
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -767,20 +770,19 @@ public void shouldAssignMultipleTasksInTasksRegistryWithStateUpdaterEnabled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(tasks.allTasks()).thenReturn(mkSet(activeTaskToClose));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(activeTaskToClose.id());
- expect(activeTaskCreator.createTasks(
- consumer,
- mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
- )).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions())),
Collections.emptyMap()
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator).createTasks(
+ consumer,
Review Comment:
updated all the formatting suggestions. As a side note, such formatting changes can be handled by code style schemes that are imported to the IDE. Is there a schema like that that can be imported to Intellij?
--
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
[GitHub] [kafka] divijvaidya commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1187177289
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
Review Comment:
we still seem to have easymock imports. Can you please remove them and replace with Mockito as well?
```
import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.expectLastCall;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.resetToStrict;
import static org.easymock.EasyMock.verify;
````
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -291,16 +292,16 @@ public void shouldPrepareActiveTaskInStateUpdaterToBeRecycled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(stateUpdater.getTasks()).thenReturn(mkSet(activeTaskToRecycle));
- expect(activeTaskCreator.createTasks(consumer, Collections.emptyMap())).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
Review Comment:
Mockito doesn't require to call `replay` (unlike EasyMock)
(same comment for rest of the places in this file where we are using replay)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -291,16 +292,16 @@ public void shouldPrepareActiveTaskInStateUpdaterToBeRecycled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(stateUpdater.getTasks()).thenReturn(mkSet(activeTaskToRecycle));
- expect(activeTaskCreator.createTasks(consumer, Collections.emptyMap())).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
Collections.emptyMap(),
mkMap(mkEntry(activeTaskToRecycle.id(), activeTaskToRecycle.inputPartitions()))
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
Review Comment:
We probably want to verify a method invocation here. If you use Mockito's verify() here instead of EasyMock, you might see a compilation error.
(same comment for rest of the usage of verify(mock) in this PR)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2579,7 +2563,9 @@ public void shouldUpdateInputPartitionsAfterRebalance() {
assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(), null), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
assertEquals(newPartitionsSet, task00.inputPartitions());
- verify(activeTaskCreator, consumer, changeLogReader);
+ verify(consumer, changeLogReader);
+ Mockito.verify(activeTaskCreator).createTasks(any(), Mockito.eq(taskId00Assignment));
Review Comment:
Mockito has a mode called STRICT_STUBS which fails the test if a defined stub is not invoked. We can greatly simplify code using that annotation since we don't have to do both `when()` and `verify()`. Using `when()` would suffice since the test will fail if the stub is not used (or using verify() would suffice for cases with void return). We use STRICT_STUBS in a bunch of places in Kafka code such as [this](https://github.com/apache/kafka/blob/347238948b86882a47faee4a2916d1b01333d95f/streams/src/test/java/org/apache/kafka/streams/state/internals/metrics/RocksDBMetricsRecorderTest.java#L60).
Please consider using it in this file as it will greatly remove boilerplate code verification.
--
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
[GitHub] [kafka] divijvaidya commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1187180990
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -291,16 +292,16 @@ public void shouldPrepareActiveTaskInStateUpdaterToBeRecycled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(stateUpdater.getTasks()).thenReturn(mkSet(activeTaskToRecycle));
- expect(activeTaskCreator.createTasks(consumer, Collections.emptyMap())).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
Review Comment:
Mockito doesn't require to call `replay` (unlike EasyMock)
(same comment for rest of the places in this file where we are using replay)
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -291,16 +292,16 @@ public void shouldPrepareActiveTaskInStateUpdaterToBeRecycled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(stateUpdater.getTasks()).thenReturn(mkSet(activeTaskToRecycle));
- expect(activeTaskCreator.createTasks(consumer, Collections.emptyMap())).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
Collections.emptyMap(),
mkMap(mkEntry(activeTaskToRecycle.id(), activeTaskToRecycle.inputPartitions()))
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
Review Comment:
We probably want to verify a method invocation here. If you use Mockito's verify() here instead of EasyMock, you might see a compilation error.
(same comment for rest of the usage of verify(mock) in this PR)
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1538991259
Addressed Divij's comment and re-based with the latest changes.
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1544909428
>
Thank you for the review @clolov - I will address the comments shortly
--
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
[GitHub] [kafka] mehbey commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1192858937
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -845,17 +847,16 @@ public void shouldRecycleTasksRemovedFromStateUpdater() {
when(tasks.removePendingTaskToRecycle(task00.id())).thenReturn(taskId00Partitions);
when(tasks.removePendingTaskToRecycle(task01.id())).thenReturn(taskId01Partitions);
taskManager = setUpTaskManager(StreamsConfigUtils.ProcessingMode.AT_LEAST_ONCE, tasks, true);
- expect(activeTaskCreator.createActiveTaskFromStandby(eq(task01), eq(taskId01Partitions), eq(consumer)))
- .andStubReturn(task01Converted);
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(anyObject());
- expectLastCall().once();
+ when(activeTaskCreator.createActiveTaskFromStandby(Mockito.eq(task01), Mockito.eq(taskId01Partitions),
Review Comment:
I simply converted the previous EasyMock version that was using the matchers to Mockito but what you are saying make sense. Updated it - Thank you
--
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
[GitHub] [kafka] clolov commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1193602799
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -767,20 +770,19 @@ public void shouldAssignMultipleTasksInTasksRegistryWithStateUpdaterEnabled() {
final TasksRegistry tasks = Mockito.mock(TasksRegistry.class);
final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
when(tasks.allTasks()).thenReturn(mkSet(activeTaskToClose));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(activeTaskToClose.id());
- expect(activeTaskCreator.createTasks(
- consumer,
- mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions()))
- )).andReturn(emptySet());
expect(standbyTaskCreator.createTasks(Collections.emptyMap())).andReturn(emptySet());
- replay(activeTaskCreator, standbyTaskCreator);
+ replay(standbyTaskCreator);
taskManager.handleAssignment(
mkMap(mkEntry(activeTaskToCreate.id(), activeTaskToCreate.inputPartitions())),
Collections.emptyMap()
);
- verify(activeTaskCreator, standbyTaskCreator);
+ verify(standbyTaskCreator);
+ Mockito.verify(activeTaskCreator).createTasks(
+ consumer,
Review Comment:
None that I know of 😞
--
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
[GitHub] [kafka] mehbey commented on pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "mehbey (via GitHub)" <gi...@apache.org>.
mehbey commented on PR #13681:
URL: https://github.com/apache/kafka/pull/13681#issuecomment-1570933391
@cadonna Thank you for the review. Addressed all outstanding comments. Please take another look when you get some time. Thank you.
--
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
[GitHub] [kafka] cadonna commented on a diff in pull request #13681: KAFKA-14133: Migrate ActiveTaskCreator mock in TaskManagerTest to Mockito
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #13681:
URL: https://github.com/apache/kafka/pull/13681#discussion_r1210581130
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3137,13 +3102,11 @@ public Set<TopicPartition> changelogPartitions() {
}
};
- expect(activeTaskCreator.createTasks(anyObject(), eq(assignment))).andStubReturn(singletonList(task00));
- activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(eq(taskId00));
- expectLastCall();
- activeTaskCreator.closeThreadProducerIfNeeded();
- expectLastCall().andThrow(new RuntimeException("whatever"));
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment))).thenReturn(singletonList(task00));
+ doThrow(new RuntimeException("whatever")).when(activeTaskCreator).closeThreadProducerIfNeeded();
expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
- replay(activeTaskCreator, standbyTaskCreator);
+ expect(standbyTaskCreator.createTasks(eq(emptyMap()))).andStubReturn(emptyList());
Review Comment:
I agree with @clolov .
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3636,9 +3586,9 @@ public void shouldCommitViaConsumerIfEosDisabled() {
@Test
public void shouldCommitViaProducerIfEosAlphaEnabled() {
final StreamsProducer producer = EasyMock.mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer);
Review Comment:
I agree
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4448,10 +4396,10 @@ public void shouldNotFailForTimeoutExceptionOnCommitWithEosAlpha() {
final TaskManager taskManager = setUpTaskManager(ProcessingMode.EXACTLY_ONCE_ALPHA, tasks, false);
final StreamsProducer producer = mock(StreamsProducer.class);
- expect(activeTaskCreator.streamsProducerForTask(anyObject(TaskId.class)))
- .andReturn(producer)
- .andReturn(producer)
- .andReturn(producer);
+ when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
+ .thenReturn(producer)
+ .thenReturn(producer)
Review Comment:
I agree
--
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