Copilot commented on code in PR #20858:
URL: https://github.com/apache/kafka/pull/20858#discussion_r2513796951
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3739,36 +3678,29 @@ public void
shouldShutDownStateUpdaterAndCloseDirtyTasksFailedDuringRemoval() {
}
@Test
- public void shouldInitializeNewActiveTasks() {
- final StateMachineTask task00 = new StateMachineTask(taskId00,
taskId00Partitions, true, stateManager);
- when(consumer.assignment()).thenReturn(assignment);
+ public void shouldInitialiseNewStandbyTasks() {
Review Comment:
The method name uses British spelling 'initialise' instead of American
spelling 'initialize', which is inconsistent with the rest of the codebase that
uses American spelling (e.g., 'initializeIfNeeded' on line 3701).
```suggestion
public void shouldInitializeNewStandbyTasks() {
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4095,29 +4033,27 @@ public void shouldNotSendPurgeDataIfPreviousNotDone() {
when(adminClient.deleteRecords(singletonMap(t1p1,
RecordsToDelete.beforeOffset(5L))))
.thenReturn(new DeleteRecordsResult(singletonMap(t1p1,
futureDeletedRecords)));
- final Map<TopicPartition, Long> purgableOffsets = new HashMap<>();
- final StateMachineTask task00 = new StateMachineTask(taskId00,
taskId00Partitions, true, stateManager) {
- @Override
- public Map<TopicPartition, Long> purgeableOffsets() {
- return purgableOffsets;
- }
- };
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RUNNING)
+ .build();
- when(consumer.assignment()).thenReturn(assignment);
- when(activeTaskCreator.createTasks(any(),
eq(taskId00Assignment))).thenReturn(singletonList(task00));
+ when(task00.purgeableOffsets())
+ .thenReturn(new HashMap<>())
+ .thenReturn(singletonMap(t1p1, 5L))
+ .thenReturn(singletonMap(t1p1, 17L));
- taskManager.handleAssignment(taskId00Assignment, emptyMap());
- assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(),
null), is(true));
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.allTasks()).thenReturn(Set.of(task00));
- assertThat(task00.state(), is(Task.State.RUNNING));
+ final TaskManager taskManager =
setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
- purgableOffsets.put(t1p1, 5L);
+ taskManager.maybePurgeCommittedRecords();
taskManager.maybePurgeCommittedRecords();
// this call should be a no-op.
- // this is verified, as there is no expectation on adminClient for
this second call,
+ // this is verified, as there is no expectation on adminClient for
this third call,
// so it would fail verification if we invoke the admin client again.
Review Comment:
The comment says "this is verified, as there is no expectation on
adminClient for this third call" but the code only calls
`maybePurgeCommittedRecords()` three times total (lines 4051-4057). The comment
should clarify that the third call should be a no-op because the second call's
purge operation hasn't completed yet, not because of missing expectations on
adminClient.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
