You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "shashankhs11 (via GitHub)" <gi...@apache.org> on 2025/11/21 05:53:08 UTC
[PR] KAFKA-19683: Final test replacements [8/N] [kafka]
shashankhs11 opened a new pull request, #20944:
URL: https://github.com/apache/kafka/pull/20944
Last set of test replacements in `TaskManagerTest.java`
Reviewers: Lucas Brutschy <lb...@confluent.io>
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on PR #20944:
URL: https://github.com/apache/kafka/pull/20944#issuecomment-3561506604
@lucasbru, tagging for review :)
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560186948
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3374,33 +3352,41 @@ public void shouldCloseActiveTasksAndPropagateStreamsProducerExceptionsOnCleanSh
@Test
public void shouldOnlyCommitRevokedStandbyTaskAndPropagatePrepareCommitException() {
Review Comment:
Yes, this is intentional. The stateupdater path does not call `prepareCommit` when closing revoked standby tasks. Instead, it directly calls `suspend()` and then `closeClean()`.
We would also need to rename the test in that case - `shouldPropagateSuspendExceptionWhenRevokingStandbyTask`
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560025442
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4360,55 +4326,85 @@ public void shouldHaveRemainingPartitionsUncleared() {
@Test
public void shouldThrowTaskMigratedWhenAllTaskCloseExceptionsAreTaskMigrated() {
- final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t1 close exception", new RuntimeException());
- }
- };
+ final StandbyTask migratedTask01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId01Partitions)
+ .build();
+ final StandbyTask migratedTask02 = standbyTask(taskId02, taskId02ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId02Partitions)
+ .build();
- final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t2 close exception", new RuntimeException());
- }
- };
- taskManager.addTask(migratedTask01);
- taskManager.addTask(migratedTask02);
+ doThrow(new TaskMigratedException("t1 close exception", new RuntimeException()))
+ .when(migratedTask01).suspend();
+ doThrow(new TaskMigratedException("t2 close exception", new RuntimeException()))
+ .when(migratedTask02).suspend();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ when(stateUpdater.tasks()).thenReturn(Set.of(migratedTask01, migratedTask02));
+
+ // mock futures for removing tasks from StateUpdater
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future01 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId01)).thenReturn(future01);
+ future01.complete(new StateUpdater.RemovedTaskResult(migratedTask01));
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future02 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId02)).thenReturn(future02);
+ future02.complete(new StateUpdater.RemovedTaskResult(migratedTask02));
final TaskMigratedException thrown = assertThrows(
TaskMigratedException.class,
() -> taskManager.handleAssignment(emptyMap(), emptyMap())
);
- // The task map orders tasks based on topic group id and partition, so here
- // t1 should always be the first.
+
+ // iteration order here is non-deterministic due to hashset.
+ // The last exception encountered wins, so we accept either
+ // t1 or t2's exception message.
assertThat(
thrown.getMessage(),
- equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ anyOf(
+ equalTo("t1 close exception; it means all tasks belonging to this thread should be migrated."),
+ equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ )
Review Comment:
Yes that sounds like a good improvement. Please update the production code in a follow-up PR and then get rid of this `anyOf`.
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560213701
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4599,28 +4615,43 @@ public void shouldNotFailForTimeoutExceptionOnConsumerCommit() {
verify(consumer, times(2)).commitSync(any(Map.class));
}
+ @SuppressWarnings("removal")
Review Comment:
The annotation was added because `ConsumerGroupMetadata` constructor was marked as `@Deprecated` and it triggered compiler warnings.
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560572568
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4599,28 +4615,43 @@ public void shouldNotFailForTimeoutExceptionOnConsumerCommit() {
verify(consumer, times(2)).commitSync(any(Map.class));
}
+ @SuppressWarnings("removal")
Review Comment:
I removed all instances of `SuppressWarnings("removal")` and mocked the class instead -- df4bfd1
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20944:
URL: https://github.com/apache/kafka/pull/20944
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560620857
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4360,55 +4326,85 @@ public void shouldHaveRemainingPartitionsUncleared() {
@Test
public void shouldThrowTaskMigratedWhenAllTaskCloseExceptionsAreTaskMigrated() {
- final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t1 close exception", new RuntimeException());
- }
- };
+ final StandbyTask migratedTask01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId01Partitions)
+ .build();
+ final StandbyTask migratedTask02 = standbyTask(taskId02, taskId02ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId02Partitions)
+ .build();
- final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t2 close exception", new RuntimeException());
- }
- };
- taskManager.addTask(migratedTask01);
- taskManager.addTask(migratedTask02);
+ doThrow(new TaskMigratedException("t1 close exception", new RuntimeException()))
+ .when(migratedTask01).suspend();
+ doThrow(new TaskMigratedException("t2 close exception", new RuntimeException()))
+ .when(migratedTask02).suspend();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ when(stateUpdater.tasks()).thenReturn(Set.of(migratedTask01, migratedTask02));
+
+ // mock futures for removing tasks from StateUpdater
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future01 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId01)).thenReturn(future01);
+ future01.complete(new StateUpdater.RemovedTaskResult(migratedTask01));
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future02 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId02)).thenReturn(future02);
+ future02.complete(new StateUpdater.RemovedTaskResult(migratedTask02));
final TaskMigratedException thrown = assertThrows(
TaskMigratedException.class,
() -> taskManager.handleAssignment(emptyMap(), emptyMap())
);
- // The task map orders tasks based on topic group id and partition, so here
- // t1 should always be the first.
+
+ // iteration order here is non-deterministic due to hashset.
+ // The last exception encountered wins, so we accept either
+ // t1 or t2's exception message.
assertThat(
thrown.getMessage(),
- equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ anyOf(
+ equalTo("t1 close exception; it means all tasks belonging to this thread should be migrated."),
+ equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ )
Review Comment:
- I reverted this test entirely and rewrote in new PR -- #20992
- I also added back `setupTaskManagerWithoutStateUpdater` in 76311ce
- I will do a cleanup of this file in another PR.
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560031613
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4599,28 +4615,43 @@ public void shouldNotFailForTimeoutExceptionOnConsumerCommit() {
verify(consumer, times(2)).commitSync(any(Map.class));
}
+ @SuppressWarnings("removal")
Review Comment:
Why did we have to add this? Maybe add a comment.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3374,33 +3352,41 @@ public void shouldCloseActiveTasksAndPropagateStreamsProducerExceptionsOnCleanSh
@Test
public void shouldOnlyCommitRevokedStandbyTaskAndPropagatePrepareCommitException() {
Review Comment:
The method name says "prepare commit exception" but then we throw in `suspend`. Is this intended?
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2548625110
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4360,55 +4326,85 @@ public void shouldHaveRemainingPartitionsUncleared() {
@Test
public void shouldThrowTaskMigratedWhenAllTaskCloseExceptionsAreTaskMigrated() {
- final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t1 close exception", new RuntimeException());
- }
- };
+ final StandbyTask migratedTask01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId01Partitions)
+ .build();
+ final StandbyTask migratedTask02 = standbyTask(taskId02, taskId02ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId02Partitions)
+ .build();
- final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t2 close exception", new RuntimeException());
- }
- };
- taskManager.addTask(migratedTask01);
- taskManager.addTask(migratedTask02);
+ doThrow(new TaskMigratedException("t1 close exception", new RuntimeException()))
+ .when(migratedTask01).suspend();
+ doThrow(new TaskMigratedException("t2 close exception", new RuntimeException()))
+ .when(migratedTask02).suspend();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ when(stateUpdater.tasks()).thenReturn(Set.of(migratedTask01, migratedTask02));
+
+ // mock futures for removing tasks from StateUpdater
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future01 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId01)).thenReturn(future01);
+ future01.complete(new StateUpdater.RemovedTaskResult(migratedTask01));
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future02 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId02)).thenReturn(future02);
+ future02.complete(new StateUpdater.RemovedTaskResult(migratedTask02));
final TaskMigratedException thrown = assertThrows(
TaskMigratedException.class,
() -> taskManager.handleAssignment(emptyMap(), emptyMap())
);
- // The task map orders tasks based on topic group id and partition, so here
- // t1 should always be the first.
+
+ // iteration order here is non-deterministic due to hashset.
+ // The last exception encountered wins, so we accept either
+ // t1 or t2's exception message.
assertThat(
thrown.getMessage(),
- equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ anyOf(
+ equalTo("t1 close exception; it means all tasks belonging to this thread should be migrated."),
+ equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ )
Review Comment:
The iteration order here with the new stateupdater code is non-deterministic, whereas earlier it was deterministic due to the use of `TreeSet`. I am not 100% sure if this was intentional and I think it may have been an oversight while refactoring?
Should we update to use `TreeSet` instead? Please advise.
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2548625110
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -4360,55 +4326,85 @@ public void shouldHaveRemainingPartitionsUncleared() {
@Test
public void shouldThrowTaskMigratedWhenAllTaskCloseExceptionsAreTaskMigrated() {
- final StateMachineTask migratedTask01 = new StateMachineTask(taskId01, taskId01Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t1 close exception", new RuntimeException());
- }
- };
+ final StandbyTask migratedTask01 = standbyTask(taskId01, taskId01ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId01Partitions)
+ .build();
+ final StandbyTask migratedTask02 = standbyTask(taskId02, taskId02ChangelogPartitions)
+ .inState(State.RUNNING)
+ .withInputPartitions(taskId02Partitions)
+ .build();
- final StateMachineTask migratedTask02 = new StateMachineTask(taskId02, taskId02Partitions, false, stateManager) {
- @Override
- public void suspend() {
- super.suspend();
- throw new TaskMigratedException("t2 close exception", new RuntimeException());
- }
- };
- taskManager.addTask(migratedTask01);
- taskManager.addTask(migratedTask02);
+ doThrow(new TaskMigratedException("t1 close exception", new RuntimeException()))
+ .when(migratedTask01).suspend();
+ doThrow(new TaskMigratedException("t2 close exception", new RuntimeException()))
+ .when(migratedTask02).suspend();
+
+ final TasksRegistry tasks = mock(TasksRegistry.class);
+ final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);
+
+ when(stateUpdater.tasks()).thenReturn(Set.of(migratedTask01, migratedTask02));
+
+ // mock futures for removing tasks from StateUpdater
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future01 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId01)).thenReturn(future01);
+ future01.complete(new StateUpdater.RemovedTaskResult(migratedTask01));
+
+ final CompletableFuture<StateUpdater.RemovedTaskResult> future02 = new CompletableFuture<>();
+ when(stateUpdater.remove(taskId02)).thenReturn(future02);
+ future02.complete(new StateUpdater.RemovedTaskResult(migratedTask02));
final TaskMigratedException thrown = assertThrows(
TaskMigratedException.class,
() -> taskManager.handleAssignment(emptyMap(), emptyMap())
);
- // The task map orders tasks based on topic group id and partition, so here
- // t1 should always be the first.
+
+ // iteration order here is non-deterministic due to hashset.
+ // The last exception encountered wins, so we accept either
+ // t1 or t2's exception message.
assertThat(
thrown.getMessage(),
- equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ anyOf(
+ equalTo("t1 close exception; it means all tasks belonging to this thread should be migrated."),
+ equalTo("t2 close exception; it means all tasks belonging to this thread should be migrated.")
+ )
Review Comment:
The iteration order here with the new stateupdater code is non-deterministic, whereas earlier it was deterministic due to the use of `TreeSet`. I am not 100% sure if this was intentional and I think it may have been an oversight while refactoring?
Should we update the production code to use `TreeSet` instead? Please advise.
--
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-19683: Final test replacements [8/N] [kafka]
Posted by "shashankhs11 (via GitHub)" <gi...@apache.org>.
shashankhs11 commented on code in PR #20944:
URL: https://github.com/apache/kafka/pull/20944#discussion_r2560575236
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java:
##########
@@ -3374,33 +3352,41 @@ public void shouldCloseActiveTasksAndPropagateStreamsProducerExceptionsOnCleanSh
@Test
public void shouldOnlyCommitRevokedStandbyTaskAndPropagatePrepareCommitException() {
Review Comment:
Renamed to `shouldPropagateSuspendExceptionWhenRevokingStandbyTask` in 3d7010f
--
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