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

Code before:
        throwIfEmptyString(request.memberId(), "MemberId can't be empty.");
        throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
        throwIfEmptyString(request.rackId(), "RackId can't be empty.");

        if (request.memberEpoch() == 0) {
            if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
                throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
            }
        } else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
            throw new InvalidRequestException("MemberEpoch is invalid.");
        }
    }

    /**
     * Validates the request.
     *
     * @param request The request to validate.
     * @throws InvalidRequestException      if the request is not valid.
     * @throws UnsupportedAssignorException if the assignor is not supported.
     */
    private void throwIfStreamsGroupHeartbeatRequestIsInvalid(
        StreamsGroupHeartbeatRequestData request
    ) throws InvalidRequestException, UnsupportedAssignorException {
        throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
        throwIfEmptyString(request.instanceId(), "InstanceId can't be empty.");
        throwIfEmptyString(request.rackId(), "RackId can't be empty.");
        throwIfNullOrEmpty(request.memberId(), "MemberId can't be null or empty.");


        if (request.memberEpoch() == 0) {
            if (request.rebalanceTimeoutMs() == -1) {
                throw new InvalidRequestException("RebalanceTimeoutMs must be provided in first request.");
            }
            if (request.activeTasks() == null || !request.activeTasks().isEmpty()) {
                throw new InvalidRequestException("ActiveTasks must be empty when (re-)joining.");
            }
            if (request.standbyTasks() == null || !request.standbyTasks().isEmpty()) {
                throw new InvalidRequestException("StandbyTasks must be empty when (re-)joining.");
            }
            if (request.warmupTasks() == null || !request.warmupTasks().isEmpty()) {
                throw new InvalidRequestException("WarmupTasks must be empty when (re-)joining.");
            }
            // TODO: check that active, standby and warmup do not intersect
        } else if (request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
            throwIfNull(request.instanceId(), "InstanceId can't be null.");
        } else if ((request.memberEpoch() <  0) && (request.memberEpoch() != LEAVE_GROUP_MEMBER_EPOCH)) {
            throw new InvalidRequestException("MemberEpoch is invalid.");
        }

        // TODO: Check supplied topology for invariants


Review comment (suggestion):
Don't need this here anymore since we first check if `memberId` is null or empty before reaching the if/else statements

Code after:

        if (request.memberEpoch() == 0) {
            if (request.subscribedTopicNames() == null || request.subscribedTopicNames().isEmpty()) {
                throw new InvalidRequestException("SubscribedTopicNames must be set in first request.");
            }
        } else if (request.memberEpoch() < ShareGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH) {
            throw new InvalidRequestException("MemberEpoch is invalid.");
        }
    }

    /**
     * Validates the request.
     *
     * @param request The request to validate.
     * @throws InvalidRequestException      if the request is not valid.
     * @throws UnsupportedAssignorException if the assignor is not supported.
     */
    private void throwIfStreamsGroupHeartbeatRequestIsInvalid(
        StreamsGroupHeartbeatRequestData request
    ) throws InvalidRequestException, UnsupportedAssignorException {
        throwIfEmptyString(request.groupId(), "GroupId can't be empty.");
        throwIfEmptyString(request.instanceId(), "InstanceId can't be empty.");
        throwIfEmptyString(request.rackId(), "RackId can't be empty.");
        throwIfNullOrEmpty(request.memberId(), "MemberId can't be null or empty.");


        if (request.memberEpoch() == 0) {
            if (request.rebalanceTimeoutMs() == -1) {
                throw new InvalidRequestException("RebalanceTimeoutMs must be provided in first request.");
            }
            if (request.activeTasks() == null || !request.activeTasks().isEmpty()) {
                throw new InvalidRequestException("ActiveTasks must be empty when (re-)joining.");
            }
            if (request.standbyTasks() == null || !request.standbyTasks().isEmpty()) {
                throw new InvalidRequestException("StandbyTasks must be empty when (re-)joining.");
            }
            if (request.warmupTasks() == null || !request.warmupTasks().isEmpty()) {
                throw new InvalidRequestException("WarmupTasks must be empty when (re-)joining.");
            }
            // TODO: check that active, standby and warmup do not intersect
        } else if (request.memberEpoch() == LEAVE_GROUP_STATIC_MEMBER_EPOCH) {
            throwIfNull(request.instanceId(), "InstanceId can't be null.");
        } else if ((request.memberEpoch() <  0) && (request.memberEpoch() != LEAVE_GROUP_MEMBER_EPOCH)) {
            throw new InvalidRequestException("MemberEpoch is invalid.");
        }

        // TODO: Check supplied topology for invariants

    }

    /**