ConsumerGroup getOrMaybeCreatePersistedConsumerGroup(
        String groupId,
        boolean createIfNotExists
    ) throws GroupIdNotFoundException {
        Group group = groups.get(groupId);

        if (group == null && !createIfNotExists) {
            throw new IllegalStateException(String.format("Consumer group %s not found.", groupId));
        }

        if (group == null) {
            ConsumerGroup consumerGroup = new ConsumerGroup(snapshotRegistry, groupId, metrics);
            groups.put(groupId, consumerGroup);
            metrics.onConsumerGroupStateTransition(null, consumerGroup.state());
            return consumerGroup;
        } else {
            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 IllegalStateException(String.format("Group %s is not a consumer group.", groupId));
            }
        }
    }
The method should be called on the replay path.
Gets or maybe creates a consumer group and updates the groups map if a new group is created.

@param groupId           The group id.
@param createIfNotExists A boolean indicating whether the group should be
                         created if it does not exist.

@return A ConsumerGroup.
@throws IllegalStateException if the group does not exist and createIfNotExists is false or
                              if the group is not a consumer group.
Package private for testing.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java