[INLONG-10564][Agent] Request configuration with md5 included

justinwwhuang · Jul 4, 2024 · 27dbc83 · 27dbc83
1 parent f112ec0
commit 27dbc83
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java b/...nt/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
@@ -30,6 +30,7 @@
 import org.apache.inlong.common.enums.PullJobTypeEnum;
 import org.apache.inlong.common.pojo.agent.AgentConfigInfo;
 import org.apache.inlong.common.pojo.agent.AgentConfigRequest;
+import org.apache.inlong.common.pojo.agent.AgentResponseCode;
 import org.apache.inlong.common.pojo.agent.DataConfig;
 import org.apache.inlong.common.pojo.agent.TaskRequest;
 import org.apache.inlong.common.pojo.agent.TaskResult;
@@ -83,6 +84,8 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
     private String uuid;
     private String clusterTag;
     private String clusterName;
+    private String taskResultMd5;
+    private String agentConfigMd5;
 
     public ManagerFetcher(AgentManager agentManager) {
         this.agentManager = agentManager;
@@ -162,6 +165,7 @@ public AgentConfigInfo getAgentConfigInfo() {
 
     public TaskRequest getTaskRequest() {
         TaskRequest request = new TaskRequest();
+        request.setMd5(taskResultMd5);
         request.setAgentIp(localIp);
         request.setUuid(uuid);
         request.setClusterName(clusterName);
@@ -172,6 +176,7 @@ public TaskRequest getTaskRequest() {
 
     public AgentConfigRequest getAgentConfigInfoRequest() {
         AgentConfigRequest request = new AgentConfigRequest();
+        request.setMd5(agentConfigMd5);
         request.setClusterTag(clusterTag);
         request.setClusterName(clusterName);
         request.setIp(localIp);
@@ -189,17 +194,19 @@ private Runnable configFetchThread() {
             while (isRunnable()) {
                 try {
                     TaskResult taskResult = getStaticConfig();
-                    if (taskResult != null) {
+                    if (taskResult != null && taskResult.getCode().equals(AgentResponseCode.SUCCESS)) {
                         List<TaskProfile> taskProfiles = new ArrayList<>();
                         taskResult.getDataConfigs().forEach((config) -> {
                             TaskProfile profile = TaskProfile.convertToTaskProfile(config);
                             taskProfiles.add(profile);
                         });
                         agentManager.getTaskManager().submitTaskProfiles(taskProfiles);
+                        taskResultMd5 = taskResult.getMd5();
                     }
                     AgentConfigInfo config = getAgentConfigInfo();
-                    if (config != null) {
+                    if (config != null && config.getCode().equals(AgentResponseCode.SUCCESS)) {
                         agentManager.subNewAgentConfigInfo(config);
+                        agentConfigMd5 = config.getMd5();
                     }
                 } catch (Throwable ex) {
                     LOGGER.warn("exception caught", ex);