public ConsumerGroup consumerGroup(
        String groupId,
        long committedOffset
    ) throws GroupIdNotFoundException {
        Group group = group(groupId, committedOffset);

        if (group.type() == CONSUMER) {
            return (ConsumerGroup) group;
        } else {
            // We don't support upgrading/downgrading between protocols at the moment so
            // we throw an exception if a group exists with the wrong type.
            throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.",
                groupId));
        }
    }
@Test
    public void testJoiningConsumerGroupWithExistingStaticMemberAndNewSubscription() throws Exception {
        String groupId = "group-id";
        Uuid fooTopicId = Uuid.randomUuid();
        String fooTopicName = "foo";
        Uuid barTopicId = Uuid.randomUuid();
        String barTopicName = "bar";
        Uuid zarTopicId = Uuid.randomUuid();
        String zarTopicName = "zar";

        String memberId1 = Uuid.randomUuid().toString();
        String memberId2 = Uuid.randomUuid().toString();
        String instanceId = "instance-id";

        MockPartitionAssignor assignor = new MockPartitionAssignor("range");
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .withConsumerGroupAssignors(Collections.singletonList(assignor))
            .withMetadataImage(new MetadataImageBuilder()
                .addTopic(fooTopicId, fooTopicName, 2)
                .addTopic(barTopicId, barTopicName, 1)
                .addTopic(zarTopicId, zarTopicName, 1)
                .addRacks()
                .build())
            .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
                .withSubscriptionMetadata(new HashMap<String, TopicMetadata>() {
                    {
                        put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 2, mkMapOfPartitionRacks(2)));
                        put(barTopicName, new TopicMetadata(barTopicId, barTopicName, 1, mkMapOfPartitionRacks(1)));
                        put(zarTopicName, new TopicMetadata(zarTopicId, zarTopicName, 1, mkMapOfPartitionRacks(1)));
                    }
                })
                .withMember(new ConsumerGroupMember.Builder(memberId1)
                    .setInstanceId(instanceId)
                    .setState(MemberState.STABLE)
                    .setMemberEpoch(10)
                    .setPreviousMemberEpoch(10)
                    .setRebalanceTimeoutMs(500)
                    .setClientId(DEFAULT_CLIENT_ID)
                    .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
                    .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
                    .setAssignedPartitions(mkAssignment(
                        mkTopicAssignment(fooTopicId, 0),
                        mkTopicAssignment(barTopicId, 0)))
                    .setClassicMemberMetadata(
                        new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
                            .setSessionTimeoutMs(5000)
                            .setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(
                                GroupMetadataManagerTestContext.toConsumerProtocol(
                                    Arrays.asList(fooTopicName, barTopicName),
                                    Arrays.asList(new TopicPartition(fooTopicName, 0), new TopicPartition(fooTopicName, 1))
                                )
                            ))
                    )
                    .build())
                .withMember(new ConsumerGroupMember.Builder(memberId2)
                    .setState(MemberState.STABLE)
                    .setMemberEpoch(10)
                    .setPreviousMemberEpoch(10)
                    .setRebalanceTimeoutMs(500)
                    .setClientId(DEFAULT_CLIENT_ID)
                    .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
                    .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName))
                    .setAssignedPartitions(mkAssignment(
                        mkTopicAssignment(fooTopicId, 1)))
                    .build())
                .withAssignment(memberId1, mkAssignment(
                    mkTopicAssignment(fooTopicId, 0),
                    mkTopicAssignment(barTopicId, 0)))
                .withAssignment(memberId2, mkAssignment(
                    mkTopicAssignment(fooTopicId, 1)))
                .withAssignmentEpoch(10))
            .build();
        ConsumerGroup group = context.groupMetadataManager.consumerGroup(groupId);
        group.setMetadataRefreshDeadline(Long.MAX_VALUE, 11);

        assignor.prepareGroupAssignment(new GroupAssignment(
            new HashMap<String, MemberAssignment>() {
                {
                    put(memberId1, new MemberAssignmentImpl(mkAssignment(
                        mkTopicAssignment(fooTopicId, 0),
                        mkTopicAssignment(zarTopicId, 0)
                    )));
                    put(memberId2, new MemberAssignmentImpl(mkAssignment(
                        mkTopicAssignment(barTopicId, 0),
                        mkTopicAssignment(fooTopicId, 1)
                    )));
                }
            }
        ));

        // Member 1 rejoins with a new subscription list.
        JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
            .withGroupId(groupId)
            .withMemberId(memberId1)
            .withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
                Arrays.asList(fooTopicName, barTopicName, zarTopicName),
                Collections.emptyList()))
            .build();
        GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(request);

        ConsumerGroupMember expectedMember = new ConsumerGroupMember.Builder(memberId1)
            .setInstanceId(instanceId)
            .setMemberEpoch(11)
            .setPreviousMemberEpoch(10)
            .setRebalanceTimeoutMs(500)
            .setClientId(DEFAULT_CLIENT_ID)
            .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
            .setState(MemberState.STABLE)
            .setSubscribedTopicNames(Arrays.asList(fooTopicName, barTopicName, zarTopicName))
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 0),
                mkTopicAssignment(zarTopicId, 0)))
            .setClassicMemberMetadata(
                new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
                    .setSessionTimeoutMs(request.sessionTimeoutMs())
                    .setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(
                        GroupMetadataManagerTestContext.toConsumerProtocol(
                            Arrays.asList(fooTopicName, barTopicName, zarTopicName),
                            Collections.emptyList()
                        )
                    ))
            )
            .build();

        List<CoordinatorRecord> expectedRecords = Arrays.asList(
            GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember),
            GroupCoordinatorRecordHelpers.newConsumerGroupEpochRecord(groupId, 11),
            GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId1, mkAssignment(
                mkTopicAssignment(fooTopicId, 0),
                mkTopicAssignment(zarTopicId, 0))),
            GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, memberId2, mkAssignment(
                mkTopicAssignment(barTopicId, 0),
                mkTopicAssignment(fooTopicId, 1))),
            GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentEpochRecord(groupId, 11),
            GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember)
        );
        assertRecordsEquals(expectedRecords.subList(0, 2), joinResult.records.subList(0, 2));
        assertUnorderedListEquals(expectedRecords.subList(2, 4), joinResult.records.subList(2, 4));
        assertRecordsEquals(expectedRecords.subList(4, 6), joinResult.records.subList(4, 6));

        joinResult.appendFuture.complete(null);
        assertEquals(
            new JoinGroupResponseData()
                .setMemberId(memberId1)
                .setGenerationId(11)
                .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                .setProtocolName("range"),
            joinResult.joinFuture.get()
        );
        context.assertSessionTimeout(groupId, memberId1, request.sessionTimeoutMs());
        context.assertSyncTimeout(groupId, memberId1, request.rebalanceTimeoutMs());
    }