@Test
    public void shouldUpdateConsumerGroupMetadataBeforeCallbacks() {
        final MockRebalanceListener rebalanceListener = new MockRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                assertEquals(2, coordinator.groupMetadata().generationId());
            }
        };

        subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));
        {
            ByteBuffer buffer = ConsumerProtocol.serializeAssignment(
                new ConsumerPartitionAssignor.Assignment(Collections.singletonList(t1p), ByteBuffer.wrap(new byte[0])));
            coordinator.onJoinComplete(1, "memberId", partitionAssignor.name(), buffer);
        }

        ByteBuffer buffer = ConsumerProtocol.serializeAssignment(
            new ConsumerPartitionAssignor.Assignment(Collections.emptyList(), ByteBuffer.wrap(new byte[0])));
        coordinator.onJoinComplete(2, "memberId", partitionAssignor.name(), buffer);
    }
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