@Override
    public Boolean submitOfflineJob(String groupId, List<InlongStreamInfo> streamInfoList, Boundaries boundaries) {
        if (offlineJobOperator == null) {
            offlineJobOperator = OfflineJobOperatorFactory.getOfflineJobOperator();
        }
        try {
            offlineJobOperator.submitOfflineJob(groupId, streamInfoList, boundaries);
            LOGGER.info("Submit offline job for group {} and stream list {} success.", groupId,
                    streamInfoList.stream().map(InlongStreamInfo::getName).collect(Collectors.toList()));
        } catch (Exception e) {
            String errorMsg = String.format("Submit offline job failed for groupId=%s", groupId);
            LOGGER.error(errorMsg, e);
            throw new BusinessException(errorMsg);
        }
        return true;
    }
There are three places may trigger resister schedule info to schedule engine:
- 1. new group approved with schedule info
- 2. new schedule info for an exist approved inlong group added
- 3. group's schedule info updated
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperatorImpl.java