shashankhs11 commented on code in PR #20544:
URL: https://github.com/apache/kafka/pull/20544#discussion_r2414114859
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -2281,38 +2276,43 @@ public void suspend() {
@Test
public void shouldCommitNonCorruptedTasksOnTaskCorruptedException() {
- final ProcessorStateManager stateManager =
mock(ProcessorStateManager.class);
-
- final StateMachineTask corruptedTask = new StateMachineTask(taskId00,
taskId00Partitions, true, stateManager);
- final StateMachineTask nonCorruptedTask = new
StateMachineTask(taskId01, taskId01Partitions, true, stateManager);
+ final StreamTask corruptedTask = statefulTask(taskId00,
taskId00ChangelogPartitions)
+ .withInputPartitions(taskId00Partitions)
+ .inState(State.RUNNING)
+ .build();
- final Map<TaskId, Set<TopicPartition>> firstAssignment = new
HashMap<>(taskId00Assignment);
- firstAssignment.putAll(taskId01Assignment);
+ final StreamTask nonCorruptedTask = statefulTask(taskId01,
taskId01ChangelogPartitions)
+ .withInputPartitions(taskId01Partitions)
+ .inState(State.RUNNING)
+ .build();
- // `handleAssignment`
- when(activeTaskCreator.createTasks(any(), eq(firstAssignment)))
- .thenReturn(asList(corruptedTask, nonCorruptedTask));
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ when(tasks.task(taskId00)).thenReturn(corruptedTask);
+ when(tasks.allTasksPerId()).thenReturn(mkMap(
+ mkEntry(taskId00, corruptedTask),
+ mkEntry(taskId01, nonCorruptedTask)
+ ));
+ when(tasks.activeTaskIds()).thenReturn(Set.of(taskId00, taskId01));
- when(consumer.assignment())
- .thenReturn(assignment)
- .thenReturn(taskId00Partitions);
+ when(nonCorruptedTask.commitNeeded()).thenReturn(true);
+ when(nonCorruptedTask.prepareCommit(true)).thenReturn(emptyMap());
+ when(corruptedTask.prepareCommit(false)).thenReturn(emptyMap());
+ doNothing().when(corruptedTask).postCommit(anyBoolean());
- taskManager.handleAssignment(firstAssignment, emptyMap());
- assertThat(taskManager.tryToCompleteRestoration(time.milliseconds(),
tp -> assertThat(tp, is(empty()))), is(true));
+ when(consumer.assignment()).thenReturn(taskId00Partitions);
- assertThat(nonCorruptedTask.state(), is(Task.State.RUNNING));
- nonCorruptedTask.setCommitNeeded();
+ final TaskManager taskManager =
setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
- corruptedTask.setChangelogOffsets(singletonMap(t1p0, 0L));
Review Comment:
```java
corruptedTask.setChangelogOffsets(singletonMap(t1p0, 0L));
```
`setChangelogOffsets()` is a method of the test utility class
`StateMachineTask`. Since we now use `StreamTask`, the method does not exist
anymore.
Just to confirm If I understood correctly, the changelog partitions are
defined at the creation time when we build the task, right?
--
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]
