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

Code before:
    }

    @Test
    public void testUpdatingMemberUpdatesPartitionEpochWhenPartitionIsNotReleased() {
        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);

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

        // m2 can acquire foo-1 because the epoch is at least as large as m1's epoch.
        consumerGroup.updateMember(m2);
        assertEquals(mkAssignment(mkTopicAssignment(fooTopicId, 1)), 
            consumerGroup.getOrMaybeCreateMember("m1", false).assignedPartitions()
        );

        ConsumerGroupMember m3 = new ConsumerGroupMember.Builder("m3")
            .setMemberEpoch(9)
            .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 (10).
        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

Review comment (suggestion):
I can't follow the purpose of this assert? (It looks like `getOrMaybeCreateMember` will always return the exact same `ConsumerGroupMember` instance passed into `updateMember`).

Code after:
    }

    @Test
    public void testUpdatingMemberUpdatesPartitionEpochWhenPartitionIsNotReleased() {
        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);

        ConsumerGroupMember m2 = new ConsumerGroupMember.Builder("m2")
            .setMemberEpoch(11)
            .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
        );