[INLONG-2894][Agent] Adapt the interface and field modification of the manager module (#2897)

Author: healchow

inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/FetcherConstants.java

@@ -17,6 +17,9 @@
 
 package org.apache.inlong.agent.constant;
 
+/**
+ * Constants of job fetcher.
+ */
 public class FetcherConstants {
 
     public static final String AGENT_FETCHER_INTERVAL = "agent.fetcher.interval";
@@ -37,7 +40,7 @@ public class FetcherConstants {
     public static final String DEFAULT_AGENT_MANAGER_VIP_HTTP_PREFIX_PATH = "/api/inlong/manager/openapi";
 
     public static final String AGENT_MANAGER_TASK_HTTP_PATH = "agent.manager.task.http.path";
-    public static final String DEFAULT_AGENT_MANAGER_TASK_HTTP_PATH = "/agent/getTask";
+    public static final String DEFAULT_AGENT_MANAGER_TASK_HTTP_PATH = "/agent/reportAndGetTask";
 
     public static final String AGENT_MANAGER_IP_CHECK_HTTP_PATH = "agent.manager.vip.http.checkIP.path";
     public static final String DEFAULT_AGENT_TDM_IP_CHECK_HTTP_PATH = "/fileAgent/confirmAgentIp";

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

@@ -169,8 +169,8 @@ public static TriggerProfile convertToTriggerProfile(DataConfig dataConfigs) {
         profileDto.setProxy(proxy);
         Job job = new Job();
 
-        //common Attribu
-        job.setId(String.valueOf(dataConfigs.getJobId()));
+        // common attribute
+        job.setId(String.valueOf(dataConfigs.getTaskId()));
         job.setChannel(DEFAULT_CHANNEL);
         job.setIp(dataConfigs.getIp());
         job.setOp(dataConfigs.getOp());

inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java

@@ -46,11 +46,10 @@ public class ManagerResultFormatter {
      */
     public static JsonObject getResultData(String jsonStr) {
         JsonObject object = GSON.fromJson(jsonStr, JsonObject.class);
-        if (object == null || !object.has(RESULT_CODE)
-                || !object.has(RESULT_DATA)
+        if (object == null || !object.has(RESULT_CODE) || !object.has(RESULT_DATA)
                 || !SUCCESS_CODE.equals(object.get(RESULT_CODE).getAsString())) {
             throw new IllegalArgumentException("cannot get result data,"
-                    + " please check manager status, return str is" + jsonStr);
+                    + " please check manager status, return str is " + jsonStr);
 
         }
         return object;

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

@@ -33,7 +33,7 @@ public class DataConfig {
     private String inlongGroupId;
     private String inlongStreamId;
     private String op;
-    private Integer jobId;
+    private Integer taskId;
     private Integer taskType;
     private String taskName;
     private String snapshot;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java

@@ -83,7 +83,7 @@ public class AgentServiceImpl implements AgentService {
     @Autowired
     private InlongStreamFieldEntityMapper streamFieldMapper;
     @Autowired
-    private InlongStreamEntityMapper inlongStreamMapper;
+    private InlongStreamEntityMapper streamMapper;
 
     /**
      * If the reported task time and the modification time in the database exceed this value,
@@ -121,20 +121,21 @@ private TaskResult getTaskResult(TaskRequest request) {
         List<StreamSourceEntity> entityList = sourceMapper.selectByIpAndUuid(agentIp, uuid);
         for (StreamSourceEntity entity : entityList) {
             DataConfig dataConfig = new DataConfig();
-            dataConfig.setJobId(entity.getId());
+            dataConfig.setTaskId(entity.getId());
             SourceType sourceType = SourceType.forType(entity.getSourceType());
             dataConfig.setTaskType(sourceType.getTaskType().getType());
             dataConfig.setTaskName(entity.getSourceName());
             dataConfig.setOp(String.valueOf(entity.getStatus() % 100));
-            String inlongGroupId = entity.getInlongGroupId();
-            String inlongStreamId = entity.getInlongStreamId();
-            dataConfig.setInlongGroupId(inlongGroupId);
-            dataConfig.setInlongStreamId(inlongStreamId);
             dataConfig.setIp(entity.getAgentIp());
             dataConfig.setUuid(entity.getUuid());
             dataConfig.setExtParams(entity.getExtParams());
             dataConfig.setSnapshot(entity.getSnapshot());
-            InlongStreamEntity inlongStreamEntity = inlongStreamMapper.selectByIdentifier(inlongGroupId,inlongStreamId);
+
+            String groupId = entity.getInlongGroupId();
+            String streamId = entity.getInlongStreamId();
+            dataConfig.setInlongGroupId(groupId);
+            dataConfig.setInlongStreamId(streamId);
+            InlongStreamEntity inlongStreamEntity = streamMapper.selectByIdentifier(groupId, streamId);
             dataConfig.setSyncSend(inlongStreamEntity.getSyncSend());
             dataConfigs.add(dataConfig);
         }