@Test
    public void testRejoinGroup() {
        String otherTopic = "otherTopic";
        final List<TopicPartition> owned = Collections.emptyList();
        final List<TopicPartition> assigned = singletonList(t1p);

        subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));

        // join the group once
        joinAsFollowerAndReceiveAssignment(coordinator, assigned);

        assertEquals(0, rebalanceListener.revokedCount);
        assertNull(rebalanceListener.revoked);
        assertEquals(1, rebalanceListener.assignedCount);
        assertEquals(getAdded(owned, assigned), rebalanceListener.assigned);

        // and join the group again
        rebalanceListener.revoked = null;
        rebalanceListener.assigned = null;
        subscriptions.subscribe(new HashSet<>(Arrays.asList(topic1, otherTopic)), Optional.of(rebalanceListener));
        client.prepareResponse(joinGroupFollowerResponse(2, consumerId, "leader", Errors.NONE));
        client.prepareResponse(syncGroupResponse(assigned, Errors.NONE));
        coordinator.joinGroupIfNeeded(time.timer(Long.MAX_VALUE));

        Collection<TopicPartition> revoked = getRevoked(assigned, assigned);
        Collection<TopicPartition> added = getAdded(assigned, assigned);
        assertEquals(revoked.isEmpty() ? 0 : 1, rebalanceListener.revokedCount);
        assertEquals(revoked.isEmpty() ? null : revoked, rebalanceListener.revoked);
        assertEquals(2, rebalanceListener.assignedCount);
        assertEquals(added, rebalanceListener.assigned);
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java