@Override
    public Boolean submitOfflineJob(OfflineJobRequest request) {
        // 1. get stream info list
        String groupId = request.getGroupId();
        InlongGroupInfo groupInfo = get(groupId);
        if (groupInfo == null) {
            String msg = String.format("InLong group not found for group=%s", groupId);
            LOGGER.error(msg);
            throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND);
        }

        List<InlongStreamInfo> streamInfoList = streamService.list(groupId);
        if (CollectionUtils.isEmpty(streamInfoList)) {
            LOGGER.warn("No stream info found for group {}, skip submit offline job", groupId);
            return false;
        }

        // check if source type is bounded source
        streamInfoList.forEach(this::checkBoundedSource);

        // get the source boundaries
        checkSourceBoundaryType(request.getBoundaryType());
        BoundaryType boundaryType = BoundaryType.getInstance(request.getBoundaryType());
        if (boundaryType == null) {
            throw new BusinessException(ErrorCodeEnum.BOUNDARY_TYPE_NOT_SUPPORTED,
                    String.format(ErrorCodeEnum.BOUNDARY_TYPE_NOT_SUPPORTED.getMessage(), request.getBoundaryType()));
        }
        Boundaries boundaries = new Boundaries(request.getLowerBoundary(), request.getUpperBoundary(), boundaryType);

        LOGGER.info("Check bounded source success, start to submitting offline job for group {}", groupId);

        return scheduleOperator.submitOfflineJob(groupId, streamInfoList, boundaries);
    }
Add constraints to the start and end time of the offline synchronization group.
1. startTime must >= current time
2. endTime must >= startTime
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java