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

Code before:

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

        // We allow changing the epoch with the same process id. 
        streamsGroup.addTaskProcessId(
            new TasksTupleWithEpochs(
                mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopologyId, 1)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 2)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 3))
            ),
            "process"
        );

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

    @Test
    public void testDeletingMemberRemovesProcessId() {
        String fooSubtopology = "foo-sub";
        String barSubtopology = "bar-sub";
        String zarSubtopology = "zar-sub";

        StreamsGroup streamsGroup = createStreamsGroup("foo");
        StreamsGroupMember member;

        member = new StreamsGroupMember.Builder("member")
            .setProcessId("process")
            .setAssignedTasks(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopology, 1)),
                    mkTasksPerSubtopology(mkTasks(fooSubtopology, 2)),
                    mkTasksPerSubtopology(mkTasks(fooSubtopology, 3))
                )
            )
            .setTasksPendingRevocation(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(barSubtopology, 4)),
                    mkTasksPerSubtopology(mkTasks(barSubtopology, 5)),
                    mkTasksPerSubtopology(mkTasks(barSubtopology, 6))

Review comment (suggestion):
Let's test the process ID replacement branch.
```suggestion
            "process1"
        );

        // We allow replacing with a different process id. 
        streamsGroup.addTaskProcessId(
            new TasksTupleWithEpochs(
                mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopologyId, 1)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 2)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 3))
            ),
            "process2"
        );

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

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

        // We allow replacing with a different process id.
        streamsGroup.addTaskProcessId(
            new TasksTupleWithEpochs(
                mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopologyId, 1)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 2)),
                mkTasksPerSubtopology(mkTasks(fooSubtopologyId, 3))
            ),
            "process2"
        );

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

    @Test
    public void testDeletingMemberRemovesProcessId() {
        String fooSubtopology = "foo-sub";
        String barSubtopology = "bar-sub";
        String zarSubtopology = "zar-sub";

        StreamsGroup streamsGroup = createStreamsGroup("foo");
        StreamsGroupMember member;

        member = new StreamsGroupMember.Builder("member")
            .setProcessId("process")
            .setAssignedTasks(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(fooSubtopology, 1)),
                    mkTasksPerSubtopology(mkTasks(fooSubtopology, 2)),
                    mkTasksPerSubtopology(mkTasks(fooSubtopology, 3))
                )
            )
            .setTasksPendingRevocation(
                new TasksTupleWithEpochs(
                    mkTasksPerSubtopologyWithCommonEpoch(10, mkTasks(barSubtopology, 4)),
                    mkTasksPerSubtopology(mkTasks(barSubtopology, 5)),
                    mkTasksPerSubtopology(mkTasks(barSubtopology, 6))
                )