@Test
    public void testSubscriptionChangeWithAuthorizationFailure() {
        // Subscribe to two topics of which only one is authorized and verify that metadata failure is propagated.
        subscriptions.subscribe(Utils.mkSet(topic1, topic2), Optional.of(rebalanceListener));
        client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                Collections.singletonMap(topic2, Errors.TOPIC_AUTHORIZATION_FAILED), singletonMap(topic1, 1)));
        assertThrows(TopicAuthorizationException.class, () -> coordinator.poll(time.timer(Long.MAX_VALUE)));

        client.respond(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        // Fail the first JoinGroup request
        client.prepareResponse(joinGroupLeaderResponse(0, consumerId, Collections.emptyMap(),
                Errors.GROUP_AUTHORIZATION_FAILED));
        assertThrows(GroupAuthorizationException.class, () -> coordinator.poll(time.timer(Long.MAX_VALUE)));

        // Change subscription to include only the authorized topic. Complete rebalance and check that
        // references to topic2 have been removed from SubscriptionState.
        subscriptions.subscribe(Utils.mkSet(topic1), Optional.of(rebalanceListener));
        assertEquals(Collections.singleton(topic1), subscriptions.metadataTopics());
        client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                Collections.emptyMap(), singletonMap(topic1, 1)));

        Map<String, List<String>> memberSubscriptions = singletonMap(consumerId, singletonList(topic1));
        partitionAssignor.prepare(singletonMap(consumerId, singletonList(t1p)));
        client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE));
        client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));

        assertEquals(singleton(topic1), subscriptions.subscription());
        assertEquals(singleton(topic1), subscriptions.metadataTopics());
    }
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