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 testJoiningConsumerGroupReplacingExistingStaticMember() throws Exception {
        String groupId = "group-id";
        Uuid fooTopicId = Uuid.randomUuid();
        String fooTopicName = "foo";

        String memberId = Uuid.randomUuid().toString();
        String instanceId = "instance-id";
        GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
            .withConsumerGroupAssignors(Collections.singletonList(new NoOpPartitionAssignor()))
            .withMetadataImage(new MetadataImageBuilder()
                .addTopic(fooTopicId, fooTopicName, 2)
                .addRacks()
                .build())
            .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
                .withSubscriptionMetadata(new HashMap<String, TopicMetadata>() {
                    {
                        put(fooTopicName, new TopicMetadata(fooTopicId, fooTopicName, 2, mkMapOfPartitionRacks(2)));
                    }
                })
                .withMember(new ConsumerGroupMember.Builder(memberId)
                    .setInstanceId(instanceId)
                    .setState(MemberState.STABLE)
                    .setMemberEpoch(10)
                    .setPreviousMemberEpoch(10)
                    .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
                    .setRebalanceTimeoutMs(500)
                    .setAssignedPartitions(mkAssignment(
                        mkTopicAssignment(fooTopicId, 0, 1)))
                    .build())
                .withAssignment(memberId, mkAssignment(
                    mkTopicAssignment(fooTopicId, 0, 1)))
                .withAssignmentEpoch(10))
            .build();
        context.groupMetadataManager.consumerGroup(groupId).setMetadataRefreshDeadline(Long.MAX_VALUE, 10);

        JoinGroupRequestData request = new GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
            .withGroupId(groupId)
            .withMemberId(UNKNOWN_MEMBER_ID)
            .withGroupInstanceId(instanceId)
            .withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(
                Collections.singletonList(fooTopicName),
                Collections.emptyList()))
            .build();

        // The static member joins with UNKNOWN_MEMBER_ID.
        GroupMetadataManagerTestContext.JoinResult joinResult = context.sendClassicGroupJoin(
            request,
            true
        );

        // Simulate a successful write to log.
        joinResult.appendFuture.complete(null);
        String newMemberId = joinResult.joinFuture.get().memberId();
        assertNotEquals("", newMemberId);

        ConsumerGroupMember expectedCopiedMember = new ConsumerGroupMember.Builder(newMemberId)
            .setMemberEpoch(0)
            .setPreviousMemberEpoch(0)
            .setInstanceId(instanceId)
            .setState(MemberState.STABLE)
            .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 0, 1)))
            .setRebalanceTimeoutMs(500)
            .build();

        ConsumerGroupMember expectedMember = new ConsumerGroupMember.Builder(newMemberId)
            .setMemberEpoch(10)
            .setPreviousMemberEpoch(0)
            .setInstanceId(instanceId)
            .setState(MemberState.STABLE)
            .setClientId(DEFAULT_CLIENT_ID)
            .setClientHost(DEFAULT_CLIENT_ADDRESS.toString())
            .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
            .setAssignedPartitions(mkAssignment(
                mkTopicAssignment(fooTopicId, 0, 1)))
            .setRebalanceTimeoutMs(500)
            .setClassicMemberMetadata(
                new ConsumerGroupMemberMetadataValue.ClassicMemberMetadata()
                    .setSessionTimeoutMs(request.sessionTimeoutMs())
                    .setSupportedProtocols(ConsumerGroupMember.classicProtocolListFromJoinRequestProtocolCollection(request.protocols())))
            .build();

        List<CoordinatorRecord> expectedRecords = Arrays.asList(
            // Remove the old static member.
            GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentTombstoneRecord(groupId, memberId),
            GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentTombstoneRecord(groupId, memberId),
            GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionTombstoneRecord(groupId, memberId),

            // Replace the old static member by the new static member.
            GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedCopiedMember),
            GroupCoordinatorRecordHelpers.newConsumerGroupTargetAssignmentRecord(groupId, newMemberId, mkAssignment(mkTopicAssignment(fooTopicId, 0, 1))),
            GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedCopiedMember),

            // Updated the new static member.
            GroupCoordinatorRecordHelpers.newConsumerGroupMemberSubscriptionRecord(groupId, expectedMember),
            GroupCoordinatorRecordHelpers.newConsumerGroupCurrentAssignmentRecord(groupId, expectedMember)
        );
        assertRecordsEquals(expectedRecords, joinResult.records);
        assertEquals(
            new JoinGroupResponseData()
                .setMemberId(newMemberId)
                .setGenerationId(10)
                .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                .setProtocolName("range"),
            joinResult.joinFuture.get()
        );

        context.assertSessionTimeout(groupId, newMemberId, request.sessionTimeoutMs());
        context.assertSyncTimeout(groupId, newMemberId, request.rebalanceTimeoutMs());
    }