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

Code before:

        streamsGroup.updateMember(m1);

        StreamsGroupMember m2 = new StreamsGroupMember.Builder("m2")
            .setProcessId("process2")
            .setAssignedTasks(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(9, mkTasks(fooSubtopologyId, 1)),
                    Map.of(),
                    Map.of()
                )
            )
            .build();

        // 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));
        }
    }


Review comment (suggestion):
```suggestion
        // Removing should be a no-op when there is no process id set.
```

Code after:

        streamsGroup.updateMember(m1);

        StreamsGroupMember m2 = new StreamsGroupMember.Builder("m2")
            .setProcessId("process2")
            .setAssignedTasks(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(9, mkTasks(fooSubtopologyId, 1)),
                    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));
        }
    }