Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/consumer/internals/SubscriptionStateTest.java (Java)

Code before:

        assertTrue(state.checkAssignmentMatchedSubscription(Set.of(tp0)));
        state.assignFromSubscribed(Set.of(tp0));

        // assigned partitions should immediately change
        assertEquals(Set.of(tp0), state.assignedPartitions());
        assertEquals(1, state.numAssignedPartitions());
        assertEquals(Set.of(topic), state.subscription());

        state.unsubscribe();
        // assigned partitions should immediately change
        assertTrue(state.assignedPartitions().isEmpty());
        assertEquals(0, state.numAssignedPartitions());
    }

    @Test
    public void verifyAssignmentId() {
        assertEquals(0, state.assignmentId());
        Set<TopicPartition> userAssignment = Set.of(tp0, tp1);
        state.assignFromUser(userAssignment);
        assertEquals(1, state.assignmentId());
        assertEquals(userAssignment, state.assignedPartitions());

        state.unsubscribe();
        assertEquals(2, state.assignmentId());
        assertEquals(Collections.emptySet(), state.assignedPartitions());

        Set<TopicPartition> autoAssignment = Set.of(t1p0);
        state.subscribe(Set.of(topic1), Optional.of(rebalanceListener));
        assertTrue(state.checkAssignmentMatchedSubscription(autoAssignment));
        state.assignFromSubscribed(autoAssignment);
        assertEquals(3, state.assignmentId());
        assertEquals(autoAssignment, state.assignedPartitions());
    }

    @Test
    public void partitionReset() {
        state.assignFromUser(Set.of(tp0));
        state.seek(tp0, 5);
        assertEquals(5L, state.position(tp0).offset);
        state.requestOffsetReset(tp0);
        assertFalse(state.isFetchable(tp0));
        assertTrue(state.isOffsetResetNeeded(tp0));
        assertNull(state.position(tp0));

        // seek should clear the reset and make the partition fetchable
        state.seek(tp0, 0);
        assertTrue(state.isFetchable(tp0));
        assertFalse(state.isOffsetResetNeeded(tp0));
    }


Review comment (suggestion):
```suggestion
        assertEquals(Set.of(), state.assignedPartitions());
```

Code after:

        // `groupSubscribe` does not accumulate
        assertFalse(state.groupSubscribe(Set.of(topic1)));
        assertEquals(Set.of(topic1), state.metadataTopics());

        state.subscribe(Set.of("anotherTopic"), Optional.of(rebalanceListener));
        assertEquals(Set.of(topic1, "anotherTopic"), state.metadataTopics());

        assertFalse(state.groupSubscribe(Set.of("anotherTopic")));
        assertEquals(Set.of("anotherTopic"), state.metadataTopics());
    }

    @Test
    public void partitionAssignmentChangeOnPatternSubscription() {
        state.subscribe(Pattern.compile(".*"), Optional.of(rebalanceListener));
        // assigned partitions should remain unchanged
        assertTrue(state.assignedPartitions().isEmpty());
        assertEquals(0, state.numAssignedPartitions());

        state.subscribeFromPattern(Set.of(topic));
        // assigned partitions should remain unchanged
        assertTrue(state.assignedPartitions().isEmpty());
        assertEquals(0, state.numAssignedPartitions());

        assertTrue(state.checkAssignmentMatchedSubscription(Set.of(tp1)));
        state.assignFromSubscribed(Set.of(tp1));

        // assigned partitions should immediately change
        assertEquals(Set.of(tp1), state.assignedPartitions());
        assertEquals(1, state.numAssignedPartitions());
        assertEquals(Set.of(topic), state.subscription());

        assertTrue(state.checkAssignmentMatchedSubscription(Set.of(t1p0)));
        state.assignFromSubscribed(Set.of(t1p0));

        // assigned partitions should immediately change
        assertEquals(Set.of(t1p0), state.assignedPartitions());
        assertEquals(1, state.numAssignedPartitions());
        assertEquals(Set.of(topic), state.subscription());

        state.subscribe(Pattern.compile(".*t"), Optional.of(rebalanceListener));
        // assigned partitions should remain unchanged
        assertEquals(Set.of(t1p0), state.assignedPartitions());
        assertEquals(1, state.numAssignedPartitions());

        state.subscribeFromPattern(Set.of(topic));
        // assigned partitions should remain unchanged
        assertEquals(Set.of(t1p0), state.assignedPartitions());
        assertEquals(1, state.numAssignedPartitions());

        assertTrue(state.checkAssignmentMatchedSubscription(Set.of(tp0)));