private ConsumerGroupMember getOrMaybeSubscribeDynamicConsumerGroupMember(
        ConsumerGroup group,
        String memberId,
        int memberEpoch,
        List<ConsumerGroupHeartbeatRequestData.TopicPartitions> ownedTopicPartitions,
        boolean createIfNotExists,
        boolean useClassicProtocol
    ) {
        ConsumerGroupMember member = group.getOrMaybeCreateMember(memberId, createIfNotExists);
        if (!useClassicProtocol) {
            throwIfConsumerGroupMemberEpochIsInvalid(member, memberEpoch, ownedTopicPartitions);
        }
        if (createIfNotExists) {
            log.info("[GroupId {}] Member {} joins the consumer group using the {} protocol.",
                group.groupId(), memberId, useClassicProtocol ? "classic" : "consumer");
        }
        return member;
    }
Gets or subscribes a new dynamic consumer group member.

@param group                 The consumer group.
@param memberId              The member id.
@param memberEpoch           The member epoch.
@param ownedTopicPartitions  The owned partitions reported by the member.
@param createIfNotExists     Whether the member should be created or not.
@param useClassicProtocol    Whether the member uses the classic protocol.

@return The existing consumer group member or a new one.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java