@Override
    @VisibleForTesting
    public boolean handleRegister(ScheduleInfo scheduleInfo) {
        start();
        String processDefUrl = projectBaseUrl + "/" + projectCode + DS_PROCESS_URL;
        String scheduleUrl = projectBaseUrl + "/" + projectCode + DS_SCHEDULE_URL;
        String processName = scheduleInfo.getInlongGroupId() + DS_DEFAULT_PROCESS_NAME;
        String processDesc = DS_DEFAULT_PROCESS_DESC + scheduleInfo.getInlongGroupId();

        LOGGER.info("Dolphin Scheduler handle register begin for {}, Checking process definition id uniqueness...",
                scheduleInfo.getInlongGroupId());
        try {
            long processDefCode = dolphinScheduleOperator.checkAndGetUniqueId(processDefUrl, token, processName);

            boolean online = false;
            if (processDefCode != 0 || scheduledProcessMap.containsKey(processDefCode)) {

                // process definition already exists, delete and rebuild
                LOGGER.info("Process definition exists, process definition id: {}, deleting...", processDefCode);
                if (dolphinScheduleOperator.releaseProcessDef(processDefUrl, processDefCode, token, DS_OFFLINE_STATE)) {
                    dolphinScheduleOperator.deleteProcessDef(processDefUrl, token, processDefCode);
                    scheduledProcessMap.remove(processDefCode);
                }
            }
            String taskCodeUrl = projectBaseUrl + "/" + projectCode + DS_TASK_CODE_URL;

            long taskCode = dolphinScheduleOperator.genTaskCode(taskCodeUrl, token);
            LOGGER.info("Generate task code for process definition success, task code: {}", taskCode);

            long offset = DolphinScheduleUtils.calculateOffset(scheduleInfo);
            processDefCode =
                    dolphinScheduleOperator.createProcessDef(processDefUrl, token, processName, processDesc, taskCode,
                            inlongManagerUrl, username, password, offset, scheduleInfo.getInlongGroupId());
            LOGGER.info("Create process definition success, process definition code: {}", processDefCode);

            if (dolphinScheduleOperator.releaseProcessDef(processDefUrl, processDefCode, token, DS_ONLINE_STATE)) {
                LOGGER.info("Release process definition success, release status: {}", DS_ONLINE_STATE);

                int scheduleId = dolphinScheduleOperator.createScheduleForProcessDef(scheduleUrl, processDefCode, token,
                        scheduleInfo);
                LOGGER.info("Create schedule for process definition success, schedule info: {}", scheduleInfo);

                online = dolphinScheduleOperator.onlineScheduleForProcessDef(scheduleUrl, scheduleId, token);
                LOGGER.info("Online schedule for process definition, status: {}", online);
            }

            scheduledProcessMap.putIfAbsent(processDefCode, processName);
            return online;
        } catch (Exception e) {
            LOGGER.error("Failed to handle unregister dolphin scheduler: ", e);
            throw new DolphinScheduleException(
                    String.format("Failed to handle unregister dolphin scheduler: %s", e.getMessage()));
        }
    }
Handle schedule register.
@param scheduleInfo schedule info to register
inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/dolphinscheduler/DolphinScheduleEngine.java