public boolean isSubscribedToTopic(String topic) {
        return subscribedTopics.map(topics -> topics.contains(topic))
            .orElse(usesConsumerGroupProtocol());
    }
@Test
    public void testIsSubscribedToTopic() {
        ClassicGroup group = new ClassicGroup(new LogContext(), "groupId", EMPTY, Time.SYSTEM, mock(GroupCoordinatorMetricsShard.class));

        // 1. group has no protocol type => not subscribed
        assertFalse(group.isSubscribedToTopic("topic"));

        // 2. group does not use consumer group protocol type => not subscribed
        JoinGroupRequestProtocolCollection protocols = new JoinGroupRequestProtocolCollection();
        protocols.add(new JoinGroupRequestProtocol()
            .setName("range")
            .setMetadata(ConsumerProtocol.serializeSubscription(
                new ConsumerPartitionAssignor.Subscription(Collections.singletonList("topic"))).array()));

        ClassicGroupMember memberWithNonConsumerProtocol = new ClassicGroupMember(
            "memberWithNonConsumerProtocol",
            Optional.empty(),
            clientId,
            clientHost,
            rebalanceTimeoutMs,
            sessionTimeoutMs,
            "My Protocol",
            protocols
        );

        group.add(memberWithNonConsumerProtocol);
        group.transitionTo(PREPARING_REBALANCE);
        group.initNextGeneration();
        assertTrue(group.isInState(COMPLETING_REBALANCE));
        assertEquals(Optional.empty(), group.computeSubscribedTopics());
        assertFalse(group.isSubscribedToTopic("topic"));

        // 3. group uses consumer group protocol type but empty members => not subscribed
        group.remove("memberWithNonConsumerProtocol");
        ClassicGroupMember memberWithConsumerProtocol = new ClassicGroupMember(
            "memberWithConsumerProtocol",
            Optional.empty(),
            clientId,
            clientHost,
            rebalanceTimeoutMs,
            sessionTimeoutMs,
            "consumer",
            protocols
        );

        group.add(memberWithConsumerProtocol);
        group.remove("memberWithConsumerProtocol");
        group.transitionTo(PREPARING_REBALANCE);
        group.initNextGeneration();
        assertTrue(group.isInState(EMPTY));
        assertEquals(Optional.of(Collections.emptySet()), group.computeSubscribedTopics());
        assertTrue(group.usesConsumerGroupProtocol());
        assertFalse(group.isSubscribedToTopic("topic"));

        // 4. group uses consumer group protocol type with member subscription => subscribed
        group.add(memberWithConsumerProtocol);
        group.transitionTo(PREPARING_REBALANCE);
        group.initNextGeneration();
        assertTrue(group.isInState(COMPLETING_REBALANCE));
        assertEquals(Optional.of(Collections.singleton("topic")), group.computeSubscribedTopics());
        assertTrue(group.usesConsumerGroupProtocol());
        assertTrue(group.isSubscribedToTopic("topic"));
    }