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

Code before:
                    Map.of(),
                    Map.of()
                )
            )
            .build();

        // m2 should not be able to acquire foo-1 because the partition is
        // still owned by another member.
        assertThrows(IllegalStateException.class, () -> streamsGroup.updateMember(m2));
    }


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

        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 fail. 
        // A debug message is logged but no exception is thrown.
        streamsGroup.removeTaskProcessIds(
            TaskAssignmentTestUtil.mkTasksTupleWithCommonEpoch(taskRole, 10, mkTasks(fooSubtopologyId, 1)),
            "process1"
        );
        assertEquals(m1.assignedTasks(), streamsGroup.getMemberOrThrow("m1").assignedTasks());
    }

    @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"
        );

        // Changing the epoch should fail because the owner of the partition
        // should remove it first.
        assertThrows(IllegalStateException.class, () -> streamsGroup.addTaskProcessId(

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

Code after:
                    Map.of(),
                    Map.of()
                )
            )
            .build();

        // 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)),