cadonna commented on code in PR #13873:
URL: https://github.com/apache/kafka/pull/13873#discussion_r1241990023
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4521,11 +4385,18 @@ public void shouldListNotPausedTasks() {
assertEquals(taskManager.notPausedTasks().size(), 0);
}
- private static void expectRestoreToBeCompleted(final Consumer<byte[],
byte[]> consumer) {
+ private static void expectAssignmentToBeCalled(final Consumer<byte[],
byte[]> consumer) {
final Set<TopicPartition> assignment = singleton(new
TopicPartition("assignment", 0));
- expect(consumer.assignment()).andReturn(assignment);
- consumer.resume(assignment);
- expectLastCall();
+ when(consumer.assignment()).thenReturn(assignment);
+ }
Review Comment:
I would inline this function since it became an one-liner.
```java
when(consumer.assignment()).thenReturn(singleton(new
TopicPartition("assignment", 0)));
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2034,12 +1993,9 @@ public void
shouldCommitNonCorruptedTasksOnTaskCorruptedException() {
// `handleAssignment`
when(activeTaskCreator.createTasks(any(), Mockito.eq(assignment)))
.thenReturn(asList(corruptedTask, nonCorruptedTask));
- expectRestoreToBeCompleted(consumer);
- expect(consumer.assignment()).andReturn(taskId00Partitions);
- // check that we should not commit empty map either
- consumer.commitSync(eq(emptyMap()));
- expectLastCall().andStubThrow(new AssertionError("should not invoke
commitSync when offset map is empty"));
- replay(consumer);
+ expectAssignmentToBeCalled(consumer);
+ final Set<TopicPartition> partitions = union(HashSet::new,
taskId00Partitions);
Review Comment:
Same here
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2449,17 +2398,15 @@ public void markChangelogAsCorrupted(final
Collection<TopicPartition> partitions
assertThat(revokedActiveTask.state(), is(State.SUSPENDED));
assertThat(unrevokedActiveTask.state(), is(State.CREATED));
assertThat(unrevokedActiveTaskWithoutCommitNeeded.state(),
is(State.RUNNING));
+ verifyResumeWasCalledWith(consumer, partitions);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2377,6 +2326,7 @@ public void markChangelogAsCorrupted(final
Collection<TopicPartition> partitions
assertThat(revokedActiveTask.state(), is(State.SUSPENDED));
assertThat(unrevokedActiveTaskWithCommitNeeded.state(),
is(State.CREATED));
assertThat(unrevokedActiveTaskWithoutCommitNeeded.state(),
is(State.RUNNING));
+ verifyResumeWasCalledWith(consumer, partition);
Review Comment:
Why do you verify? It was not verified in the original code.
I added a couple of this comments below. However, I am not sure whether the
original author did not want to verify the consumer or if they forgot about it.
I will leave it to you if you want to keep them or not.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2805,6 +2716,7 @@ public void
shouldNotCommitOnHandleAssignmentIfNoTaskClosed() {
assertThat(task00.commitNeeded, is(true));
assertThat(task10.commitPrepared, is(false));
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2834,14 +2744,14 @@ public void
shouldNotCommitOnHandleAssignmentIfOnlyStandbyTaskClosed() {
taskManager.handleAssignment(assignmentActive, Collections.emptyMap());
assertThat(task00.commitNeeded, is(true));
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2688,20 +2614,8 @@ public void
shouldCommitAllActiveTasksThatNeedCommittingOnHandleRevocationWithEo
.thenReturn(singletonList(task10));
final ConsumerGroupMetadata groupMetadata = new
ConsumerGroupMetadata("appId");
- expect(consumer.groupMetadata()).andReturn(groupMetadata);
- producer.commitTransaction(expectedCommittedOffsets, groupMetadata);
- expectLastCall();
-
- task00.committedOffsets();
- EasyMock.expectLastCall();
- task01.committedOffsets();
- EasyMock.expectLastCall();
- task02.committedOffsets();
- EasyMock.expectLastCall();
- task10.committedOffsets();
- EasyMock.expectLastCall();
Review Comment:
Yeah, that seems weird...
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4521,11 +4385,18 @@ public void shouldListNotPausedTasks() {
assertEquals(taskManager.notPausedTasks().size(), 0);
}
- private static void expectRestoreToBeCompleted(final Consumer<byte[],
byte[]> consumer) {
+ private static void expectAssignmentToBeCalled(final Consumer<byte[],
byte[]> consumer) {
final Set<TopicPartition> assignment = singleton(new
TopicPartition("assignment", 0));
- expect(consumer.assignment()).andReturn(assignment);
- consumer.resume(assignment);
- expectLastCall();
+ when(consumer.assignment()).thenReturn(assignment);
+ }
+
+ private static void verifyResumeWasCalled(final Consumer<byte[], byte[]>
consumer) {
+ final Set<TopicPartition> assignment = singleton(new
TopicPartition("assignment", 0));
+ Mockito.verify(consumer, atLeastOnce()).resume(assignment);
+ }
+
+ private static void verifyResumeWasCalledWith(final Consumer<byte[],
byte[]> consumer, Set<TopicPartition> assignment) {
Review Comment:
I would inline this function.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1998,10 +1957,10 @@ public void suspend() {
}
};
- expectRestoreToBeCompleted(consumer);
+ expectAssignmentToBeCalled(consumer);
when(activeTaskCreator.createTasks(any(),
Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expect(consumer.assignment()).andReturn(taskId00Partitions);
- replay(consumer);
+ final Set<TopicPartition> partitions = union(HashSet::new,
taskId00Partitions);
+ when(consumer.assignment()).thenReturn(partitions);
Review Comment:
Same as above also here
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2541,6 +2482,7 @@ public void shouldAddNewActiveTasks() {
assertThat(taskManager.activeTaskMap(),
Matchers.equalTo(singletonMap(taskId00, task00)));
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
Mockito.verify(changeLogReader).enforceRestoreActive();
+ Mockito.verify(consumer).resume(emptySet());
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -1961,10 +1920,10 @@ public void postCommit(final boolean enforceCheckpoint)
{
};
// `handleAssignment`
- expectRestoreToBeCompleted(consumer);
+ expectAssignmentToBeCalled(consumer);
when(activeTaskCreator.createTasks(any(),
Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expect(consumer.assignment()).andReturn(taskId00Partitions);
- replay(consumer);
+ final Set<TopicPartition> partitions = union(HashSet::new,
taskId00Partitions);
+ when(consumer.assignment()).thenReturn(partitions);
Review Comment:
Why do you union a single set?
```suggestion
when(consumer.assignment()).thenReturn(taskId00Partitions);
```
... and isn't this a duplicate of line 1923 (except the differing
partitions).
```java
expectAssignmentToBeCalled(consumer);
```
Could it be that you can remove line 1923?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2469,19 +2416,18 @@ public void
shouldCloseStandbyUnassignedTasksWhenCreatingNewTasks() {
assertThat(task00.state(), is(Task.State.CLOSED));
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2494,17 +2440,16 @@ public void shouldAddNonResumedSuspendedTasks() {
assertThat(task01.state(), is(Task.State.RUNNING));
Mockito.verify(activeTaskCreator).createTasks(any(),
Mockito.eq(emptyMap()));
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2418,19 +2368,18 @@ public void markChangelogAsCorrupted(final
Collection<TopicPartition> partitions
mkEntry(taskId02, taskId02Partitions)
);
- expectRestoreToBeCompleted(consumer);
+ expectAssignmentToBeCalled(consumer);
when(activeTaskCreator.createTasks(any(),
Mockito.eq(assignmentActive)))
.thenReturn(asList(revokedActiveTask, unrevokedActiveTask,
unrevokedActiveTaskWithoutCommitNeeded));
final ConsumerGroupMetadata groupMetadata = new
ConsumerGroupMetadata("appId");
- expect(consumer.groupMetadata()).andReturn(groupMetadata);
+ when(consumer.groupMetadata()).thenReturn(groupMetadata);
doThrow(new
TimeoutException()).when(producer).commitTransaction(expectedCommittedOffsets,
groupMetadata);
- expect(consumer.assignment()).andStubReturn(union(HashSet::new,
taskId00Partitions, taskId01Partitions, taskId02Partitions));
-
- replay(consumer, stateManager);
Review Comment:
Did you miss to replay the state manager?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2688,20 +2614,8 @@ public void
shouldCommitAllActiveTasksThatNeedCommittingOnHandleRevocationWithEo
.thenReturn(singletonList(task10));
final ConsumerGroupMetadata groupMetadata = new
ConsumerGroupMetadata("appId");
- expect(consumer.groupMetadata()).andReturn(groupMetadata);
- producer.commitTransaction(expectedCommittedOffsets, groupMetadata);
- expectLastCall();
-
- task00.committedOffsets();
- EasyMock.expectLastCall();
- task01.committedOffsets();
- EasyMock.expectLastCall();
- task02.committedOffsets();
- EasyMock.expectLastCall();
- task10.committedOffsets();
- EasyMock.expectLastCall();
-
- replay(consumer);
+ when(consumer.groupMetadata()).thenReturn(groupMetadata);
+ doNothing().when(producer).commitTransaction(expectedCommittedOffsets,
groupMetadata);
Review Comment:
I think something strange happened here. Not your fault. The producer is
used like an EasyMock mock but it is specified as a Mockito mock. The producer
does not do anything in the test as far as I can see.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4034,14 +3913,13 @@ public Set<TopicPartition> changelogPartitions() {
};
when(activeTaskCreator.createTasks(any(),
Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- replay(consumer);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(),
null), is(false));
assertThat(task00.state(), is(Task.State.RESTORING));
// this could be a bit mysterious; we're verifying _no_ interactions
on the consumer,
// since the taskManager should _not_ resume the assignment while
we're still in RESTORING
- verify(consumer);
+ Mockito.verifyNoMoreInteractions(consumer);
Review Comment:
```suggestion
Mockito.verifyNoInteractions(consumer);
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2775,6 +2686,8 @@ public void
shouldCommitAllNeededTasksOnHandleRevocation() {
assertThat(task01.commitPrepared, is(true));
assertThat(task02.commitPrepared, is(false));
assertThat(task10.commitPrepared, is(false));
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2629,19 +2558,16 @@ public void shouldSuspendActiveTasksDuringRevocation() {
final Map<TopicPartition, OffsetAndMetadata> offsets =
singletonMap(t1p0, new OffsetAndMetadata(0L, null));
task00.setCommittableOffsetsAndMetadata(offsets);
- expectRestoreToBeCompleted(consumer);
+ expectAssignmentToBeCalled(consumer);
when(activeTaskCreator.createTasks(any(),
Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- consumer.commitSync(offsets);
- expectLastCall();
-
- replay(consumer);
taskManager.handleAssignment(taskId00Assignment, emptyMap());
assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(),
null), is(true));
assertThat(task00.state(), is(Task.State.RUNNING));
taskManager.handleRevocation(taskId00Partitions);
assertThat(task00.state(), is(Task.State.SUSPENDED));
+ verifyResumeWasCalled(consumer);
Review Comment:
Why do you verify? It was not verified in the original code.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4521,11 +4385,18 @@ public void shouldListNotPausedTasks() {
assertEquals(taskManager.notPausedTasks().size(), 0);
}
- private static void expectRestoreToBeCompleted(final Consumer<byte[],
byte[]> consumer) {
+ private static void expectAssignmentToBeCalled(final Consumer<byte[],
byte[]> consumer) {
final Set<TopicPartition> assignment = singleton(new
TopicPartition("assignment", 0));
- expect(consumer.assignment()).andReturn(assignment);
- consumer.resume(assignment);
- expectLastCall();
+ when(consumer.assignment()).thenReturn(assignment);
+ }
+
+ private static void verifyResumeWasCalled(final Consumer<byte[], byte[]>
consumer) {
+ final Set<TopicPartition> assignment = singleton(new
TopicPartition("assignment", 0));
+ Mockito.verify(consumer, atLeastOnce()).resume(assignment);
Review Comment:
I would inline this function.
--
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
