Repository: apache/kafka
File: group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java (Java)

Code before:
            throw new GroupIdNotFoundException(String.format("Consumer group %s not found.", groupId));
        }

        if (group == null) {
            return new ConsumerGroup(snapshotRegistry, groupId, metrics);
        } else if (createIfNotExists && maybeDeleteEmptyClassicGroup(group, records)) {
            log.info("[GroupId {}] Converted the empty classic group to a consumer group.", groupId);
            return new ConsumerGroup(snapshotRegistry, groupId, metrics);
        } else {
            if (group.type() == CONSUMER) {
                return (ConsumerGroup) group;
            } else if (createIfNotExists && group.type() == CLASSIC) {
                validateOnlineUpgrade((ClassicGroup) group);
                return convertToConsumerGroup((ClassicGroup) group, records);
            } else {
                throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.", groupId));
            }
        }
    }

    /**
     * Gets or creates a streams group without updating the groups map.
     * The group will be materialized during the replay.
     *
     * If there is an empty classic consumer group of the same name, it will be deleted and a new streams
     * group.
     *
     * @param groupId           The group ID.
     * @param records           The record list to which the group tombstones are written
     *                          if the group is empty and is a classic group.
     *
     * @return A StreamsGroup.
     *
     * Package private for testing.
     */
    StreamsGroup getOrCreateStreamsGroup(
        String groupId,
        List<CoordinatorRecord> records
    ) {
        Group group = groups.get(groupId);

        if (group == null) {
            return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
        } else if (maybeDeleteEmptyClassicGroup(group, records)) {
            log.info("[GroupId {}] Converted the empty classic group to a streams group.", groupId);
            return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
        } else {
            return castToStreamsGroup(group);
        }
    }


Review comment (suggestion):
The comment is incomplete - it ends abruptly with 'a new streams' without completing the sentence. Should be 'a new streams group will be created.'
```suggestion
     * group will be created.
```

Code after:
            throw new GroupIdNotFoundException(String.format("Consumer group %s not found.", groupId));
        }

        if (group == null) {
            return new ConsumerGroup(snapshotRegistry, groupId, metrics);
        } else if (createIfNotExists && maybeDeleteEmptyClassicGroup(group, records)) {
            log.info("[GroupId {}] Converted the empty classic group to a consumer group.", groupId);
            return new ConsumerGroup(snapshotRegistry, groupId, metrics);
        } else {
            if (group.type() == CONSUMER) {
                return (ConsumerGroup) group;
            } else if (createIfNotExists && group.type() == CLASSIC) {
                validateOnlineUpgrade((ClassicGroup) group);
                return convertToConsumerGroup((ClassicGroup) group, records);
            } else {
                throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.", groupId));
            }
        }
    }

    /**
     * Gets or creates a streams group without updating the groups map.
     * The group will be materialized during the replay.
     *
     * If there is an empty classic consumer group of the same name, it will be deleted and a new streams
     * group will be created.
     *
     * @param groupId           The group ID.
     * @param records           The record list to which the group tombstones are written
     *                          if the group is empty and is a classic group.
     *
     * @return A StreamsGroup.
     *
     * Package private for testing.
     */
    StreamsGroup getOrCreateStreamsGroup(
        String groupId,
        List<CoordinatorRecord> records
    ) {
        Group group = groups.get(groupId);

        if (group == null) {
            return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
        } else if (maybeDeleteEmptyClassicGroup(group, records)) {
            log.info("[GroupId {}] Converted the empty classic group to a streams group.", groupId);
            return new StreamsGroup(logContext, snapshotRegistry, groupId, metrics);
        } else {
            return castToStreamsGroup(group);
        }
    }