Support scheudle infomartion management

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java

@@ -125,6 +125,9 @@ public enum ErrorCodeEnum {
     MQ_TYPE_IS_NULL(1600, "MQ type is null"),
     MQ_TYPE_NOT_SUPPORT(1601, "MQ type '%s' not support"),
 
+    SCHEDULE_NOT_FOUND(1700, "Schedule info not found"),
+    SCHEDULE_DUPLICATE(1701, "Schedule info already exist"),
+
     WORKFLOW_EXE_FAILED(4000, "Workflow execution exception"),
     WORKFLOW_APPROVER_NOT_FOUND(4001, "Workflow approver does not exist/no operation authority"),
     WORKFLOW_DELETE_RECORD_FAILED(4002, "Workflow delete record failure"),

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java

@@ -44,6 +44,8 @@ public enum OperationTarget {
 
     INLONG_ROLE,
 
-    TENANT_ROLE
+    TENANT_ROLE,
+
+    SCHEDULE
 
 }

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java

@@ -0,0 +1,29 @@
+package org.apache.inlong.manager.common.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum ScheduleStatus {
+
+    NEW(100, "new"),
+    DELETED(40, "deleted");
+
+    private final Integer code;
+    private final String description;
+
+    ScheduleStatus(Integer code, String description) {
+        this.code = code;
+        this.description = description;
+    }
+
+
+    public static ScheduleStatus forCode(int code) {
+        for (ScheduleStatus status : values()) {
+            if (status.getCode() == code) {
+                return status;
+            }
+        }
+        throw new IllegalStateException(String.format("Illegal code=%s for ScheduleStatus", code));
+    }
+
+}

inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java

@@ -0,0 +1,41 @@
+package org.apache.inlong.manager.dao.entity;
+
+import java.io.Serializable;
+import java.util.Date;
+import lombok.Data;
+
+@Data
+public class ScheduleEntity implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private Integer id;
+    // inLong group id
+    private String groupId;
+    // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+    private Integer scheduleType;
+    // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+    // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+    private String scheduleUnit;
+    private Integer scheduleInterval;
+    // schedule start time, long type timestamp
+    private Long startTime;
+    // schedule end time, long type timestamp
+    private Long endTime;
+    // delay time to start task, in minutes
+    private Integer delayTime;
+    // if task depend on itself
+    private Integer selfDepend;
+    private Integer taskParallelism;
+    // expression of crontab, used when scheduleType is crontab
+    private String crontabExpression;
+
+    private Integer status;
+    private Integer previousStatus;
+    private Integer isDeleted;
+    private String creator;
+    private String modifier;
+    private Date createTime;
+    private Date modifyTime;
+    private Integer version;
+
+}

inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityMapper.java

@@ -0,0 +1,16 @@
+package org.apache.inlong.manager.dao.mapper;
+
+import java.util.List;
+import org.apache.inlong.manager.dao.entity.ScheduleEntity;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface ScheduleEntityMapper {
+    int insert(ScheduleEntity scheduleEntity);
+
+    ScheduleEntity selectByPrimaryKey(Long id);
+
+    ScheduleEntity selectByGroupId(String groupId);
+
+    int updateByIdSelective(ScheduleEntity scheduleEntity);
+}

inlong-manager/manager-dao/src/main/resources/mappers/ScheduleEntityMapper.xml

@@ -0,0 +1,120 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+      http://www.apache.org/licenses/LICENSE-2.0
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="org.apache.inlong.manager.dao.mapper.ScheduleEntityMapper">
+    <resultMap id="BaseResultMap" type="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+        <id column="id" jdbcType="INTEGER" property="id"/>
+        <result column="group_id" jdbcType="VARCHAR" property="groupId"/>
+        <result column="schedule_type" jdbcType="INTEGER" property="scheduleType"/>
+        <result column="schedule_unit" jdbcType="VARCHAR" property="scheduleUnit"/>
+        <result column="schedule_interval" jdbcType="INTEGER" property="scheduleInterval"/>
+        <result column="start_time" jdbcType="TIMESTAMP" property="startTime"/>
+        <result column="end_time" jdbcType="TIMESTAMP" property="endTime"/>
+        <result column="delay_time" jdbcType="INTEGER" property="delayTime"/>
+        <result column="self_depend" jdbcType="INTEGER" property="selfDepend"/>
+        <result column="task_parallelism" jdbcType="INTEGER" property="taskParallelism"/>
+        <result column="crontab_expression" jdbcType="VARCHAR" property="crontabExpression"/>
+
+        <result column="status" jdbcType="INTEGER" property="status"/>
+        <result column="previous_status" jdbcType="INTEGER" property="previousStatus"/>
+        <result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
+        <result column="creator" jdbcType="VARCHAR" property="creator"/>
+        <result column="modifier" jdbcType="VARCHAR" property="modifier"/>
+        <result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
+        <result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime"/>
+        <result column="version" jdbcType="INTEGER" property="version"/>
+    </resultMap>
+
+    <sql id="Base_Column_List">
+        id, group_id, schedule_type, schedule_unit, schedule_interval, start_time,
+          end_time, delay_time, self_depend, task_parallelism, crontab_expression,
+           status, previous_status, is_deleted, creator, modifier, create_time, modify_time, version
+    </sql>
+
+    <insert id="insert" useGeneratedKeys="true" keyProperty="id"
+            parameterType="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+        insert into schedule_config (id, group_id, schedule_type, schedule_unit,
+                                     schedule_interval, start_time, end_time, delay_time,
+                                     self_depend, task_parallelism, crontab_expression,
+                                     status, previous_status, is_deleted, creator, modifier,
+                                     create_time, modify_time, version)
+        values (#{id, jdbcType=INTEGER}, #{group_id, jdbcType=VARCHAR},
+                #{schedule_type, jdbcType=VARCHAR}, #{schedule_unit, jdbcType=VARCHAR},
+                #{schedule_interval, jdbcType=INTEGER}, #{start_time, jdbcType=TIMESTAMP},
+                #{end_time, jdbcType=TIMESTAMP}, #{delay_time, jdbcType=INTEGER},
+                #{self_depend, jdbcType=INTEGER}, #{task_parallelism, jdbcType=INTEGER},
+                #{crontab_expression, jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
+                #{previousStatus,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR},
+                #{modifier,jdbcType=VARCHAR})
+    </insert>
+
+    <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
+        select
+        <include refid="Base_Column_List"/>
+        from schedule_config
+        where id = #{id,jdbcType=INTEGER}
+    </select>
+
+    <select id="selectByGroupId" parameterType="java.lang.String" resultMap="BaseResultMap">
+        select
+        <include refid="Base_Column_List"/>
+        from schedule_config
+        where group_id = #{group_id,jdbcType=VARCHAR}
+    </select>
+
+    <update id="updateByIdSelective" parameterType="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+        update schedule_config
+        <set>
+            <if test="group_id != null">
+                group_id = #{group_id, jdbcType=VARCHAR},
+            </if>
+            <if test="schedule_type != null">
+                schedule_type = #{schedule_type, jdbcType=INTEGER},
+            </if>
+            <if test="schedule_unit !=null">
+                schedule_unit = #{schedule_unit, jdbcType=VARCHAR},
+            </if>
+            <if test="schedule_interval != null">
+                schedule_interval = #{schedule_interval, jdbcType=INTEGER},
+            </if>
+            <if test="start_time != null">
+                start_time = #{start_time, jdbcType=TIMESTAMP},
+            </if>
+            <if test="end_time != null">
+                end_time = #{end_time, jdbcType=TIMESTAMP},
+            </if>
+            <if test="delay_time != null">
+                delay_time = #{delay_time, jdbcType=INTEGER},
+            </if>
+            <if test="self_depend != null">
+                self_depend = #{self_depend, jdbcType=INTEGER},
+            </if>
+            <if test="task_parallelism != null">
+                task_parallelism = #{task_parallelism, jdbcType=INTEGER},
+            </if>
+            <if test="crontab_expression != null">
+                crontab_expression = #{crontab_expression, jdbcType=VARCHAR},
+            </if>
+        </set>
+        <where>
+            id = #{id, jdbcType=INTEGER}
+        </where>
+    </update>
+
+</mapper>

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java

@@ -0,0 +1,57 @@
+package org.apache.inlong.manager.pojo.schedule;
+
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import javax.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.inlong.manager.common.validation.UpdateValidation;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel("Schedule response")
+public class ScheduleInfo {
+
+    @ApiModelProperty(value = "Primary key")
+    @NotNull(groups = UpdateValidation.class)
+    private Integer id;
+
+    @ApiModelProperty("Inlong Group ID")
+    @NotNull
+    private String groupId;
+
+    // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+    @ApiModelProperty("Schedule type")
+    private Integer scheduleType;
+
+    // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+    // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+    @ApiModelProperty("TimeUnit for schedule interval")
+    private String intervalTimeunit;
+
+    @ApiModelProperty("Schedule interval")
+    private Integer interval;
+
+    @ApiModelProperty("Start time")
+    private Long startTime;
+
+    @ApiModelProperty("End time")
+    private Long endTime;
+
+    @ApiModelProperty("Delay time")
+    private Integer delayTime;
+
+    @ApiModelProperty("Self depend")
+    private Integer selfDepend;
+
+    @ApiModelProperty("Schedule task parallelism")
+    private Integer taskParallelism;
+
+    @ApiModelProperty("Schedule task parallelism")
+    private Integer crontabExpression;
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java

@@ -0,0 +1,56 @@
+package org.apache.inlong.manager.pojo.schedule;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import javax.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.inlong.manager.common.validation.UpdateValidation;
+
+@Data
+@ApiModel("Schedule request")
+public class ScheduleInfoRequest {
+
+    @ApiModelProperty(value = "Primary key")
+    @NotNull(groups = UpdateValidation.class)
+    private Integer id;
+
+    @ApiModelProperty("Inlong Group ID")
+    @NotNull
+    private String groupId;
+
+    // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+    @ApiModelProperty("Schedule type")
+    private Integer scheduleType;
+
+    // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+    // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+    @ApiModelProperty("TimeUnit for schedule interval")
+    private String intervalTimeunit;
+
+    @ApiModelProperty("Schedule interval")
+    private Integer interval;
+
+    @ApiModelProperty("Start time")
+    private Long startTime;
+
+    @ApiModelProperty("End time")
+    private Long endTime;
+
+    @ApiModelProperty("Delay time")
+    private Integer delayTime;
+
+    @ApiModelProperty("Self depend")
+    private Integer selfDepend;
+
+    @ApiModelProperty("Schedule task parallelism")
+    private Integer taskParallelism;
+
+    @ApiModelProperty("Schedule task parallelism")
+    private Integer crontabExpression;
+
+}
+

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java

@@ -0,0 +1,45 @@
+package org.apache.inlong.manager.service.schedule;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+public interface ScheduleService {
+
+    /**
+     * Save schedule info.
+     *
+     * @param scheduleInfo schedule request need to save
+     * @param operator name of operator
+     * @return schedule info id in backend storage
+     */
+    int save(@Valid @NotNull(message = "schedule request cannot be null") ScheduleInfoRequest scheduleInfo,
+            String operator);
+
+    /**
+     * Query whether schedule info exists for specified inlong group
+     *
+     * @param groupId the group id to be queried
+     * @return does it exist
+     */
+    Boolean exist(String groupId);
+
+    /**
+     * Get schedule info based on inlong group id
+     *
+     * @param groupId inlong group id
+     * @return detail of inlong group
+     */
+    ScheduleInfo get(String groupId);
+
+    /**
+     * Modify schedule information
+     *
+     * @param scheduleInfo schedule request that needs to be modified
+     * @param operator name of operator
+     * @return whether succeed
+     */
+    Boolean update(@Valid @NotNull(message = "schedule request cannot be null") ScheduleInfoRequest scheduleInfo,
+            String operator);
+}