You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by we...@apache.org on 2024/07/05 02:18:58 UTC
(inlong) branch master updated: [INLONG-10564][Agent] Request configuration with md5 included (#10565)
This is an automated email from the ASF dual-hosted git repository.
wenweihuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 1a01ea9dc5 [INLONG-10564][Agent] Request configuration with md5 included (#10565)
1a01ea9dc5 is described below
commit 1a01ea9dc511ca9b302d20c89f7d92e484c8dd1a
Author: justinwwhuang <hw...@163.com>
AuthorDate: Fri Jul 5 10:18:53 2024 +0800
[INLONG-10564][Agent] Request configuration with md5 included (#10565)
---
.../apache/inlong/agent/plugin/fetcher/ManagerFetcher.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
index d31126860e..b4e54071de 100644
--- a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
+++ b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerFetcher.java
@@ -30,6 +30,7 @@ import org.apache.inlong.agent.utils.ThreadUtils;
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 class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
public TaskRequest getTaskRequest() {
TaskRequest request = new TaskRequest();
+ request.setMd5(taskResultMd5);
request.setAgentIp(localIp);
request.setUuid(uuid);
request.setClusterName(clusterName);
@@ -172,6 +176,7 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
public AgentConfigRequest getAgentConfigInfoRequest() {
AgentConfigRequest request = new AgentConfigRequest();
+ request.setMd5(agentConfigMd5);
request.setClusterTag(clusterTag);
request.setClusterName(clusterName);
request.setIp(localIp);
@@ -189,17 +194,19 @@ public class ManagerFetcher extends AbstractDaemon implements ProfileFetcher {
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);