[INLONG-10601][Manager] Optimize the agent task configuration process (…

…apache#10602)

inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/AgentConfigInfo.java

@@ -34,6 +34,7 @@ public class AgentConfigInfo {
     AgentResponseCode code;
     private String zkUrl;
     private AgentClusterInfo cluster;
+    private Integer version;
     private String md5;
 
     @Data

inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/TaskResult.java

@@ -36,6 +36,7 @@ public class TaskResult {
     private List<CmdConfig> cmdConfigs;
     private List<DataConfig> dataConfigs;
     private String md5;
+    private Integer version;
     AgentResponseCode code;
 
 }

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

@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.manager.dao.entity;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * Agent task config entity, including agent ip, cluster name, etc.
+ */
+@Data
+public class AgentTaskConfigEntity implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+    private Integer id;
+    private String clusterName;
+    private String agentIp;
+
+    private String configParams;
+
+    private String taskParams;
+
+    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/AgentTaskConfigEntityMapper.java

@@ -0,0 +1,45 @@
+/*
+ * 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.
+ */
+
+package org.apache.inlong.manager.dao.mapper;
+
+import org.apache.inlong.manager.common.tenant.MultiTenantQuery;
+import org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity;
+
+import org.apache.ibatis.annotations.Options;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.cursor.Cursor;
+import org.apache.ibatis.mapping.ResultSetType;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface AgentTaskConfigEntityMapper {
+
+    int insert(AgentTaskConfigEntity record);
+
+    AgentTaskConfigEntity selectByPrimaryKey(Integer id);
+
+    AgentTaskConfigEntity selectByIdentifier(@Param("agentIp") String agentIp,
+            @Param("clusterName") String clusterName);
+
+    int updateByIdSelective(AgentTaskConfigEntity record);
+
+    @MultiTenantQuery(with = false)
+    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = Integer.MIN_VALUE)
+    Cursor<AgentTaskConfigEntity> selectAllAgentTaskConfigs();
+
+}

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

@@ -0,0 +1,100 @@
+<?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.AgentTaskConfigEntityMapper">
+    <resultMap id="BaseResultMap" type="org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity">
+        <id column="id" jdbcType="INTEGER" property="id"/>
+        <result column="agent_ip" jdbcType="VARCHAR" property="agentIp"/>
+        <result column="cluster_name" jdbcType="VARCHAR" property="clusterName"/>
+        <result column="config_params" jdbcType="VARCHAR" property="configParams"/>
+        <result column="task_params" jdbcType="VARCHAR" property="taskParams"/>
+        <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, agent_ip, cluster_name, config_params, task_params, is_deleted, creator, modifier, create_time, modify_time, version
+    </sql>
+    <insert id="insert" useGeneratedKeys="true" keyProperty="id"
+            parameterType="org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity">
+        insert into agent_task_config (id, agent_ip, cluster_name,
+                                 config_params, task_params,
+                                 creator, modifier)
+        values (#{id, jdbcType=INTEGER}, #{agentIp, jdbcType=VARCHAR}, #{clusterName, jdbcType=VARCHAR},
+                #{configParams, jdbcType=VARCHAR}, #{taskParams, jdbcType=VARCHAR},
+                #{creator, jdbcType=VARCHAR}, #{modifier, jdbcType=VARCHAR})
+    </insert>
+
+    <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
+        select
+        <include refid="Base_Column_List"/>
+        from agent_task_config
+        where id = #{id,jdbcType=INTEGER}
+        and is_deleted = 0
+    </select>
+    <select id="selectByIdentifier" resultType="org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity">
+        select
+        <include refid="Base_Column_List"/>
+        from agent_task_config
+        where agent_ip = #{agentIp,jdbcType=VARCHAR}
+        and cluster_name = #{clusterName, jdbcType=VARCHAR}
+        and is_deleted = 0
+    </select>
+    <select id="selectAllAgentTaskConfigs" resultType="org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity">
+        select
+        <include refid="Base_Column_List"/>
+        from agent_task_config
+        <where>
+            and is_deleted = 0
+            and agent_ip is not null
+        </where>
+    </select>
+    <update id="updateByIdSelective" parameterType="org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity">
+        update agent_task_config
+        <set>
+            <if test="agentIp != null">
+                agent_ip = #{agentIp,jdbcType=VARCHAR},
+            </if>
+            <if test="clusterName != null">
+                cluster_name = #{clusterName,jdbcType=VARCHAR},
+            </if>
+            <if test="configParams != null">
+                config_params = #{configParams,jdbcType=VARCHAR},
+            </if>
+            <if test="taskParams != null">
+                task_params = #{taskParams,jdbcType=VARCHAR},
+            </if>
+            <if test="isDeleted != null">
+                is_deleted = #{isDeleted,jdbcType=INTEGER},
+            </if>
+            <if test="modifier != null">
+                modifier = #{modifier,jdbcType=VARCHAR},
+            </if>
+            version = #{version,jdbcType=INTEGER} + 1
+        </set>
+        where id = #{id,jdbcType=INTEGER}
+        and version = #{version,jdbcType=INTEGER}
+    </update>
+</mapper>

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/SortConfigLoader.java → inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/ConfigLoader.java

@@ -17,6 +17,7 @@
 
 package org.apache.inlong.manager.service.core;
 
+import org.apache.inlong.manager.dao.entity.AgentTaskConfigEntity;
 import org.apache.inlong.manager.dao.entity.ClusterConfigEntity;
 import org.apache.inlong.manager.dao.entity.DataNodeEntity;
 import org.apache.inlong.manager.dao.entity.InlongGroupExtEntity;
@@ -35,7 +36,7 @@
 /**
  * Loader for sort service to load configs thought Cursor
  */
-public interface SortConfigLoader {
+public interface ConfigLoader {
 
     /**
      * Load all clusters by cursor
@@ -124,4 +125,11 @@ public interface SortConfigLoader {
      */
     List<ClusterConfigEntity> loadAllClusterConfigEntity();
 
+    /**
+     * Load all agent task config info
+     *
+     * @return List of agent task config info
+     */
+    List<AgentTaskConfigEntity> loadAllAgentTaskConfigEntity();
+
 }