Repository: apache/kafka
File: group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTest.java (Java)

Code before:

        // m1 should retain ownership of the task since m2 has a smaller epoch.
        assertThrows(IllegalStateException.class, () -> streamsGroup.updateMember(m2));
    }


    @ParameterizedTest
    @EnumSource(TaskRole.class)
    public void testRemoveTaskProcessIds(TaskRole taskRole) {
        String fooSubtopologyId = "foo-sub";
        StreamsGroup streamsGroup = createStreamsGroup("foo");

        // Removing should be a no-op when there is no epoch set.
        streamsGroup.removeTaskProcessIds(
            TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 10, mkTasks(fooSubtopologyId, 1)),
            "process"
        );

        StreamsGroupMember m1 = new StreamsGroupMember.Builder("m1")
            .setProcessId("process")
            .setAssignedTasks(TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 10, mkTasks(fooSubtopologyId, 1)))
            .build();

        streamsGroup.updateMember(m1);

        // Removing with incorrect epoch should do nothing. 
        // A debug message is logged, no exception is thrown.
        streamsGroup.removeTaskProcessIds(
            TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 9, mkTasks(fooSubtopologyId, 1)),
            "process1"
        );
        assertEquals(m1.assignedTasks(), streamsGroup.getMemberOrThrow("m1").assignedTasks());
        if (taskRole == TaskRole.ACTIVE) {
            assertEquals(10, streamsGroup.getMemberOrThrow("m1").assignedTasks().activeTasksWithEpochs()
                .get(fooSubtopologyId).get(1));
        }
    }

    @Test
    public void testAddTaskProcessIds() {
        String fooSubtopologyId = "foo-sub";
        StreamsGroup streamsGroup = createStreamsGroup("foo");

        streamsGroup.addTaskProcessId(
            new TasksTupleWithEpochs(
                mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopologyId, 1)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 2)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 3))
            ),
            "process"
        );

Review comment (bug):
```suggestion
        // Removing with incorrect process id should do nothing. 
```

Code after:

        // We allow m2 to acquire foo-1 despite the fact that m1 has ownership because the processId is different.
        streamsGroup.updateMember(m2);

        assertEquals("process2", streamsGroup.currentActiveTaskProcessId(fooSubtopologyId, 1));
    }


    @ParameterizedTest
    @EnumSource(TaskRole.class)
    public void testRemoveTaskProcessIds(TaskRole taskRole) {
        String fooSubtopologyId = "foo-sub";
        StreamsGroup streamsGroup = createStreamsGroup("foo");

        // Removing should be a no-op when there is no process id set.
        streamsGroup.removeTaskProcessIds(
            TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 10, mkTasks(fooSubtopologyId, 1)),
            "process"
        );

        StreamsGroupMember m1 = new StreamsGroupMember.Builder("m1")
            .setProcessId("process")
            .setAssignedTasks(TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 10, mkTasks(fooSubtopologyId, 1)))
            .build();

        streamsGroup.updateMember(m1);

        // Removing with incorrect process id should do nothing. 
        // A debug message is logged, no exception is thrown.
        streamsGroup.removeTaskProcessIds(
            TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 9, mkTasks(fooSubtopologyId, 1)),
            "process1"
        );
        if (taskRole == TaskRole.ACTIVE) {
            assertEquals("process", streamsGroup.currentActiveTaskProcessId(fooSubtopologyId, 1));
        }
    }

    @Test
    public void testAddTaskProcessIds() {
        String fooSubtopologyId = "foo-sub";
        StreamsGroup streamsGroup = createStreamsGroup("foo");

        streamsGroup.addTaskProcessId(
            new TasksTupleWithEpochs(
                mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopologyId, 1)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 2)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 3))
            ),
            "process"
        );