guozhangwang commented on code in PR #12386:
URL: https://github.com/apache/kafka/pull/12386#discussion_r923908100
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -143,6 +142,27 @@ public void shouldThrowIfStandbyTaskNotInStateRunning() {
shouldThrowIfTaskNotInGivenState(task, State.RUNNING);
}
+ @Test
+ public void shouldThrowIfAddingActiveTasksWithSameId() throws Exception {
+ final StreamTask task1 =
createActiveStatefulTaskInStateRestoring(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_A_0));
+ final StreamTask task2 =
createActiveStatefulTaskInStateRestoring(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_A_0));
+ shouldThrowIfAddingTasksWithSameId(task1, task2);
+ }
+
+ @Test
+ public void shouldThrowIfAddingStandbyTasksWithSameId() throws Exception {
+ final StandbyTask task1 = createStandbyTaskInStateRunning(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_B_0));
+ final StandbyTask task2 = createStandbyTaskInStateRunning(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_B_0));
+ shouldThrowIfAddingTasksWithSameId(task1, task2);
+ }
+
+ @Test
+ public void shouldThrowIfAddingActiveAndStandbyTaskWithSameId() throws
Exception {
+ final StreamTask task1 =
createActiveStatefulTaskInStateRestoring(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_A_0));
+ final StandbyTask task2 = createStandbyTaskInStateRunning(TASK_0_0,
Collections.singletonList(TOPIC_PARTITION_B_0));
+ shouldThrowIfAddingTasksWithSameId(task1, task2);
+ }
+
private void shouldThrowIfTaskNotInGivenState(final Task task, final State
correctState) {
Review Comment:
Ack.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
