Repository: apache/kafka
File: group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroupTest.java (Java)

Code before:
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 1)))
            .build();

        // m2 should not be able to acquire foo-1 because the partition is
        // still owned by another member.
        consumerGroup.updateMember(m2);
        assertEquals(mkAssignment(mkTopicAssignment(fooTopicId, 1)), 
            consumerGroup.getOrMaybeCreateMember("m1", false).assignedPartitions()
        );
    }

    @Test
    public void testRemovePartitionEpochs() {
        Uuid fooTopicId = Uuid.randomUuid();
        ConsumerGroup consumerGroup = createConsumerGroup("foo");

        ConsumerGroupMember m1 = new ConsumerGroupMember.Builder("m1")
            .setMemberEpoch(10)
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 1)))
            .build();

        consumerGroup.updateMember(m1);

        // Removing with incorrect epoch should fail. 
        // A debug message is logged but no exception is thrown.
        consumerGroup.removePartitionEpochs(
            mkAssignment(
                mkTopicAssignment(fooTopicId, 1)
            ),
            11
        );
        assertEquals(
            mkAssignment(mkTopicAssignment(fooTopicId, 1)),
            consumerGroup.getOrMaybeCreateMember("m1", false).assignedPartitions()
        );
        assertEquals(10, consumerGroup.currentPartitionEpoch(fooTopicId, 1));
    }

    @Test
    public void testAddPartitionEpochs() {
        Uuid fooTopicId = Uuid.randomUuid();
        ConsumerGroup consumerGroup = createConsumerGroup("foo");

        consumerGroup.addPartitionEpochs(
            mkAssignment(
                mkTopicAssignment(fooTopicId, 1)
            ),
            10
        );

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

Code after:
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 1)))
            .build();

        // m2 can acquire foo-1 because the epoch is larger than m1's epoch.
        // This should not throw IllegalStateException.
        consumerGroup.updateMember(m2);

        ConsumerGroupMember m3 = new ConsumerGroupMember.Builder("m3")
            .setMemberEpoch(10)
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 1)))
            .build();

        // m3 should not be able to acquire foo-1 because the epoch is smaller 
        // than the current partition epoch (11).
        assertThrows(IllegalStateException.class, () -> {
            consumerGroup.updateMember(m3);
        });
    }

    @Test
    public void testRemovePartitionEpochs() {
        Uuid fooTopicId = Uuid.randomUuid();
        ConsumerGroup consumerGroup = createConsumerGroup("foo");

        // Removing should be a no-op when there is no epoch set.
        consumerGroup.removePartitionEpochs(
            mkAssignment(
                mkTopicAssignment(fooTopicId, 1)
            ),
            10
        );

        ConsumerGroupMember m1 = new ConsumerGroupMember.Builder("m1")
            .setMemberEpoch(10)
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 1)))
            .build();

        consumerGroup.updateMember(m1);

        // Removing with incorrect epoch should do nothing. 
        // A debug message is logged, no exception is thrown.
        consumerGroup.removePartitionEpochs(
            mkAssignment(
                mkTopicAssignment(fooTopicId, 1)
            ),
            11
        );
        assertEquals(10, consumerGroup.currentPartitionEpoch(fooTopicId, 1));