shashankhs11 commented on code in PR #20858:
URL: https://github.com/apache/kafka/pull/20858#discussion_r2525582670
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3323,157 +3322,86 @@ public void
shouldCloseActiveTasksAndPropagateExceptionsOnCleanShutdownWithExact
}
private void
shouldCloseActiveTasksAndPropagateExceptionsOnCleanShutdown(final
ProcessingMode processingMode) {
- final TaskManager taskManager =
setUpTaskManagerWithoutStateUpdater(processingMode, null, false);
-
- final TopicPartition changelog = new TopicPartition("changelog", 0);
- final Map<TaskId, Set<TopicPartition>> assignment = mkMap(
- mkEntry(taskId00, taskId00Partitions),
- mkEntry(taskId01, taskId01Partitions),
- mkEntry(taskId02, taskId02Partitions),
- mkEntry(taskId03, taskId03Partitions)
- );
- final Task task00 = new StateMachineTask(taskId00, taskId00Partitions,
true, stateManager) {
- @Override
- public Set<TopicPartition> changelogPartitions() {
- return singleton(changelog);
- }
- };
- final AtomicBoolean closedDirtyTask01 = new AtomicBoolean(false);
- final AtomicBoolean closedDirtyTask02 = new AtomicBoolean(false);
- final AtomicBoolean closedDirtyTask03 = new AtomicBoolean(false);
- final Task task01 = new StateMachineTask(taskId01, taskId01Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("migrated", new
RuntimeException("cause"));
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask01.set(true);
- }
- };
- final Task task02 = new StateMachineTask(taskId02, taskId02Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new RuntimeException("oops");
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask02.set(true);
- }
- };
- final Task task03 = new StateMachineTask(taskId03, taskId03Partitions,
true, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new RuntimeException("oops");
- }
-
- @Override
- public void closeDirty() {
- super.closeDirty();
- closedDirtyTask03.set(true);
- }
- };
-
- when(activeTaskCreator.createTasks(any(), eq(assignment)))
- .thenReturn(asList(task00, task01, task02, task03));
- taskManager.handleAssignment(assignment, emptyMap());
+ final StreamTask task00 = statefulTask(taskId00,
taskId00ChangelogPartitions)
Review Comment:
addressed in 5e36ed5
--
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]
