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

Code before:
                || !areOwnedTasksContainedInAssignedTasks(ownedWarmupTasks, member.assignedTasks().warmupTasks())) {
                throw new FencedMemberEpochException("The streams group member has a smaller member "
                    + "epoch (" + receivedMemberEpoch + ") than the one known by the group coordinator ("
                    + member.memberEpoch() + "). The member must abandon all its partitions and rejoin.");
            }
        }
    }

    /**
     * Validates that the requested tasks exist in the configured topology and partitions are valid.
     * If tasks is null, does nothing. If an invalid task is found, throws InvalidRequestException.
     *
     * @param topology The configured topology.
     * @param tasks    The list of requested tasks.
     */
    private static void throwIfRequestContainsInvalidTasks(
        SortedMap<String, ConfiguredSubtopology> subtopologySortedMap,
        List<StreamsGroupHeartbeatRequestData.TaskIds> tasks
    ) {
        if (tasks == null) return;
        for (StreamsGroupHeartbeatRequestData.TaskIds task : tasks) {
            String subtopologyId = task.subtopologyId();
            if (!subtopologySortedMap.containsKey(subtopologyId)) {
                throw new InvalidRequestException("Subtopology " + subtopologyId + " does not exist in the topology.");
            }
            ConfiguredSubtopology subtopology = subtopologySortedMap.get(subtopologyId);
            int numTasks = subtopology.numberOfTasks();
            for (Integer partition : task.partitions()) {
                if (partition < 0 || partition >= numTasks) {
                    throw new InvalidRequestException("Task " + partition + " for subtopology " + subtopologyId +
                        " is invalid. Number of tasks for this subtopology: " + numTasks);
                }
            }
        }
    }

    /**
     * Validates if the received classic member protocols are supported by the group.
     *
     * @param group         The ConsumerGroup.
     * @param memberId      The joining member id.
     * @param protocolType  The joining member protocol type.
     * @param protocols     The joining member protocol collection.
     */
    private void throwIfClassicProtocolIsNotSupported(
        ConsumerGroup group,
        String memberId,
        String protocolType,
        JoinGroupRequestProtocolCollection protocols
    ) {
        if (!group.supportsClassicProtocols(protocolType, ClassicGroupMember.plainProtocolSet(protocols))) {

Review comment (refactor):
nit: move this up and change the check to `if (subtopology == null) ..` minor point, so I'll leave it up to you.

Code after:
    /**
     * Validates that the requested tasks exist in the configured topology and partitions are valid.
     * If tasks is null, does nothing. If an invalid task is found, throws InvalidRequestException.
     *
     * @param subtopologySortedMap The configured topology.
     * @param tasks                The list of requested tasks.
     */
    private static void throwIfRequestContainsInvalidTasks(
        SortedMap<String, ConfiguredSubtopology> subtopologySortedMap,
        List<StreamsGroupHeartbeatRequestData.TaskIds> tasks
    ) {
        if (tasks == null || tasks.isEmpty()) return;
        for (StreamsGroupHeartbeatRequestData.TaskIds task : tasks) {
            String subtopologyId = task.subtopologyId();
            ConfiguredSubtopology subtopology = subtopologySortedMap.get(subtopologyId);
            if (subtopology == null) {
                throw new InvalidRequestException("Subtopology " + subtopologyId + " does not exist in the topology.");
            }
            int numTasks = subtopology.numberOfTasks();
            for (Integer partition : task.partitions()) {
                if (partition < 0 || partition >= numTasks) {
                    throw new InvalidRequestException("Task " + partition + " for subtopology " + subtopologyId +
                        " is invalid. Number of tasks for this subtopology: " + numTasks);
                }
            }
        }
    }

    /**
     * Validates if the received classic member protocols are supported by the group.
     *
     * @param group         The ConsumerGroup.
     * @param memberId      The joining member id.
     * @param protocolType  The joining member protocol type.
     * @param protocols     The joining member protocol collection.
     */
    private void throwIfClassicProtocolIsNotSupported(
        ConsumerGroup group,
        String memberId,
        String protocolType,
        JoinGroupRequestProtocolCollection protocols
    ) {
        if (!group.supportsClassicProtocols(protocolType, ClassicGroupMember.plainProtocolSet(protocols))) {
            throw Errors.INCONSISTENT_GROUP_PROTOCOL.exception("Member " + memberId + "'s protocols are not supported.");
        }
    }

    /**
     * Validates if the consumer group member uses the classic protocol.
     *
     * @param member The ConsumerGroupMember.