You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "clolov (via GitHub)" <gi...@apache.org> on 2024/01/25 13:47:41 UTC
[PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
clolov opened a new pull request, #15261:
URL: https://github.com/apache/kafka/pull/15261
The previous pull request in this series was https://github.com/apache/kafka/pull/15112.
This pull request continues the migration of the consumer mock in TaskManagerTest test by test for easier reviews.
I envision there will be at least 1 more pull request to clean things up. For example, all calls to `taskManager.setMainConsumer` should be removed.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1981388046
Heya @cadonna, apologies for the delay! Hopefully the latest version addresses your comments 😊, but if not let me know and I will pick it up tomorrow
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481556951
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2332,19 +2335,12 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
task00.setCommittableOffsetsAndMetadata(offsets);
// first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
- when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
-
- // `handleRevocation`
- consumer.commitSync(offsets);
- expectLastCall();
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
- // second `handleAssignment`
- consumer.commitSync(offsets);
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- replay(consumer);
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481555805
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1466405733
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2358,6 +2354,7 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
There are quite a few tests where this verification did not hold true. By spot-checking this in EasyMock the tests appeared to pass without this change there as well.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481561392
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2332,19 +2335,12 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
task00.setCommittableOffsetsAndMetadata(offsets);
// first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
- when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
-
- // `handleRevocation`
- consumer.commitSync(offsets);
- expectLastCall();
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481560896
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2358,6 +2354,7 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481560896
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481557433
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2369,12 +2366,9 @@ public void closeClean() {
}
};
- // first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
- replay(consumer);
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481555805
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1466405733
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2358,6 +2354,7 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
There are quite a few tests where this verification did not hold true. By spot-checking this in EasyMock the tests appeared to pass without this change there as well.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4821,8 +4911,10 @@ private Map<TaskId, StateMachineTask> handleAssignment(final Map<TaskId, Set<Top
when(standbyTaskCreator.createTasks(standbyAssignment)).thenReturn(standbyTasks);
when(activeTaskCreator.createTasks(any(), Mockito.eq(allActiveTasksAssignment))).thenReturn(allActiveTasks);
- expectRestoreToBeCompleted(consumer);
- replay(consumer);
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ lenient().when(mockitoConsumer.assignment()).thenReturn(assignment);
Review Comment:
`lenient()` was needed here because this method is a dependency on quite a few tests some of which do not actually call this. The tests which do not call it are
* `shouldClassifyExistingTasksWithoutStateUpdater`
* `shouldComputeOffsetSumForNonRunningActiveTask`
* `shouldSkipUnknownOffsetsWhenComputingOffsetSum`
* `shouldReportLatestOffsetAsOffsetSumForRunningTask`
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1979391796
Thanks for the review @cadonna! I will provide an updated version tomorrow morning!
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481566534
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2016,13 +2015,14 @@ public void shouldReleaseLockForUnassignedTasksAfterRebalance() throws Exception
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01, taskId02)));
handleAssignment(taskId00Assignment, taskId01Assignment, emptyMap());
- reset(consumer);
- expectConsumerAssignmentPaused(consumer);
- replay(consumer);
taskManager.handleRebalanceComplete();
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01)));
verify(stateDirectory);
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
Review Comment:
I am not certain I follow, do you just want `singleton(new TopicPartition("assignment", 0));` to be encapsulated in its own method?
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1512305606
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2016,13 +2015,14 @@ public void shouldReleaseLockForUnassignedTasksAfterRebalance() throws Exception
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01, taskId02)));
handleAssignment(taskId00Assignment, taskId01Assignment, emptyMap());
- reset(consumer);
- expectConsumerAssignmentPaused(consumer);
- replay(consumer);
taskManager.handleRebalanceComplete();
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01)));
verify(stateDirectory);
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
Review Comment:
I see `singleton(new TopicPartition("assignment", 0))` added around 45 times in this PR. I assume this is because `handleAssignment()` uses it internally. I propose to define a field final static variable `assignment` or similar that can be reused. Alternatively, the assignment can also be passed to `handleAssignment()`. Maybe to limit the changes in this PR, you should go for the former option of the field variable.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2210,6 +2210,9 @@ public void shouldComputeOffsetSumForStandbyTask() throws Exception {
restoringTask.setChangelogOffsets(changelogOffsets);
assertThat(taskManager.getTaskOffsetSums(), is(expectedOffsetSums));
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
With EasyMock you need to call `verify()` to verify a call. I do not see a call to `verify()` in the old code. That tells me that verifying the call to `consumer.resume()` was not relevant in the old code of this test. Additionally, the goal of this test, is to verify the result of `taskManager.getTaskOffsetSums()`. For that you need some tasks assigned. So part of the setup of this test is to install an assignment so that you get tasks assigned. With `Mockito.verify(mockitoConsumer).resume(assignment)` you would test rather an unimportant part of the setup of this test.
To sum up, I have the impression that you verify more in the migration that it was verified in the old code and you added a rather unimportant verification.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2332,19 +2335,12 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
task00.setCommittableOffsetsAndMetadata(offsets);
// first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
- when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
-
- // `handleRevocation`
- consumer.commitSync(offsets);
- expectLastCall();
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
- // second `handleAssignment`
- consumer.commitSync(offsets);
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- replay(consumer);
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
Fair enough! Waiting for your follow-up PR then.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1910268608
Heya @lucasbru @divijvaidya and @cadonna - a polite call for review 😊! There are only 2 classes left to move to Mockito!
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1918964713
Thank you, I will aim to address the comments as soon as possible!
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1471396786
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2332,19 +2335,12 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
task00.setCommittableOffsetsAndMetadata(offsets);
// first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
- when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
-
- // `handleRevocation`
- consumer.commitSync(offsets);
- expectLastCall();
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
- // second `handleAssignment`
- consumer.commitSync(offsets);
- expectLastCall();
+ when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- replay(consumer);
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
That could be set in `setupTaskManager()`.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2016,13 +2015,14 @@ public void shouldReleaseLockForUnassignedTasksAfterRebalance() throws Exception
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01, taskId02)));
handleAssignment(taskId00Assignment, taskId01Assignment, emptyMap());
- reset(consumer);
- expectConsumerAssignmentPaused(consumer);
- replay(consumer);
taskManager.handleRebalanceComplete();
assertThat(taskManager.lockedTaskDirectories(), is(mkSet(taskId00, taskId01)));
verify(stateDirectory);
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
Review Comment:
This assignment is duplicated a couple of times. Could you try to avoid that duplication?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2369,12 +2366,9 @@ public void closeClean() {
}
};
- // first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
- replay(consumer);
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
Set in `setupTaskManager()`.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2210,6 +2210,9 @@ public void shouldComputeOffsetSumForStandbyTask() throws Exception {
restoringTask.setChangelogOffsets(changelogOffsets);
assertThat(taskManager.getTaskOffsetSums(), is(expectedOffsetSums));
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
I do not think we need this verification. Method `resume()` is called in some setup code of the test.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4821,8 +4911,10 @@ private Map<TaskId, StateMachineTask> handleAssignment(final Map<TaskId, Set<Top
when(standbyTaskCreator.createTasks(standbyAssignment)).thenReturn(standbyTasks);
when(activeTaskCreator.createTasks(any(), Mockito.eq(allActiveTasksAssignment))).thenReturn(allActiveTasks);
- expectRestoreToBeCompleted(consumer);
- replay(consumer);
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ lenient().when(mockitoConsumer.assignment()).thenReturn(assignment);
+
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
Why did you not modify this call in `setUpTaskManager()`?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2332,19 +2335,12 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
task00.setCommittableOffsetsAndMetadata(offsets);
// first `handleAssignment`
- expectRestoreToBeCompleted(consumer);
- when(activeTaskCreator.createTasks(any(), Mockito.eq(taskId00Assignment))).thenReturn(singletonList(task00));
- expectLastCall();
-
- // `handleRevocation`
- consumer.commitSync(offsets);
- expectLastCall();
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
Review Comment:
I do not think you need this in this test. This call is not relevant for this test.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2358,6 +2354,7 @@ public void shouldCloseActiveUnassignedSuspendedTasksWhenClosingRevokedTasks() {
assertThat(taskManager.activeTaskMap(), Matchers.anEmptyMap());
assertThat(taskManager.standbyTaskMap(), Matchers.anEmptyMap());
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
I do not think you need this in this test. This call is not relevant for this test.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2437,6 +2432,7 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
assertThat(taskManager.lockedTaskDirectories(), is(emptySet()));
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
Not needed because not relevant for this test.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2399,7 +2393,8 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager);
// `handleAssignment`
- expectRestoreToBeCompleted(consumer);
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
Review Comment:
Not needed because not relevant for this test.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481555805
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4821,8 +4911,10 @@ private Map<TaskId, StateMachineTask> handleAssignment(final Map<TaskId, Set<Top
when(standbyTaskCreator.createTasks(standbyAssignment)).thenReturn(standbyTasks);
when(activeTaskCreator.createTasks(any(), Mockito.eq(allActiveTasksAssignment))).thenReturn(allActiveTasks);
- expectRestoreToBeCompleted(consumer);
- replay(consumer);
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ lenient().when(mockitoConsumer.assignment()).thenReturn(assignment);
+
+ taskManager.setMainConsumer(mockitoConsumer);
Review Comment:
I can try to completely get rid of the `consumer` in this pull request. I wanted to minimise the changes needed to review this change (fix all tests which relied on `expectRestoreToBeCompleted`). I was imagining one more pull request after this one is merged which gets rid of the the remaining usage of `consumer` and moves this into the setup. If you want to see all changes in this pull request I can try to achieve them in a subsequent commit.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1932183121
Heya @cadonna, apologies for the delay. I am not certain I fully understand the comments, so I wanted to confirm before making changes. The purpose of this pull request is to just migrate the tests depending on calls to `expectRestoreToBeCompleted`. That method included both a `when` and a `verify` step. This is why when we move it to Mockito-world it becomes a bit more verbose. I understand the `resume` verification might not be the purpose of the test, but it was verified in EasyMock-world and I didn't want to make the assertions more lenient. I have one more pull request after this one which should clean up all these intermediate steps and get rid of the `mockitoConsumer`, that's why I didn't make the changes in the setup method.
If you are happy to review the final changes directly I could try to merge the subsequent pull request into this one.
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on PR #15261:
URL: https://github.com/apache/kafka/pull/15261#issuecomment-1983094643
test failures are unrelated
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna merged PR #15261:
URL: https://github.com/apache/kafka/pull/15261
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481561763
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2399,7 +2393,8 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
final StateMachineTask task01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager);
// `handleAssignment`
- expectRestoreToBeCompleted(consumer);
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ when(mockitoConsumer.assignment()).thenReturn(assignment);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481560896
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2437,6 +2432,7 @@ public void shouldCloseActiveTasksWhenHandlingLostTasks() throws Exception {
assertThat(taskManager.lockedTaskDirectories(), is(emptySet()));
Mockito.verify(activeTaskCreator).closeAndRemoveTaskProducerIfNeeded(taskId00);
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
Same reply as in https://github.com/apache/kafka/pull/15261#discussion_r1481560896
--
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
Re: [PR] KAFKA-14133: Move consumer mock in TaskManagerTest to Mockito - part 2 [kafka]
Posted by "clolov (via GitHub)" <gi...@apache.org>.
clolov commented on code in PR #15261:
URL: https://github.com/apache/kafka/pull/15261#discussion_r1481560896
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2210,6 +2210,9 @@ public void shouldComputeOffsetSumForStandbyTask() throws Exception {
restoringTask.setChangelogOffsets(changelogOffsets);
assertThat(taskManager.getTaskOffsetSums(), is(expectedOffsetSums));
+
+ final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
+ Mockito.verify(mockitoConsumer).resume(assignment);
Review Comment:
I was trying to keep the same strength of verification as was already present from the method I was trying to get rid of i.e.
```
private static void expectRestoreToBeCompleted(final Consumer<byte[], byte[]> consumer) {
final Set<TopicPartition> assignment = singleton(new TopicPartition("assignment", 0));
expect(consumer.assignment()).andReturn(assignment);
consumer.resume(assignment); <-- THIS
expectLastCall();
}
```
If you think I can relax this, I am happy to remove it from the tests where you have made the same remark
--
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