This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 45ca134f94 [INLONG-9736][Agent] Make time zone as a common parameter
(#9737)
45ca134f94 is described below
commit 45ca134f940f5f30f946e612232ca1f0d4742e40
Author: justinwwhuang <[email protected]>
AuthorDate: Wed Feb 28 11:03:39 2024 +0800
[INLONG-9736][Agent] Make time zone as a common parameter (#9737)
---
.../org/apache/inlong/agent/conf/TaskProfile.java | 2 +-
.../inlong/agent/constant/CommonConstants.java | 2 +-
.../inlong/agent/constant/FetcherConstants.java | 2 +-
.../inlong/agent/constant/TaskConstants.java | 2 +-
.../org/apache/inlong/agent/pojo/FileTask.java | 2 -
.../apache/inlong/agent/pojo/TaskProfileDto.java | 5 +--
.../inlong/agent/core/AgentBaseTestsHelper.java | 2 +-
.../agent/plugin/fetcher/ManagerFetcher.java | 49 ++++++----------------
.../inlong/agent/plugin/task/file/LogFileTask.java | 2 +-
.../inlong/agent/plugin/AgentBaseTestsHelper.java | 2 +-
10 files changed, 22 insertions(+), 48 deletions(-)
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TaskProfile.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TaskProfile.java
index ed96536a32..1f77433c9f 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TaskProfile.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TaskProfile.java
@@ -70,7 +70,7 @@ public class TaskProfile extends AbstractConfiguration {
}
public String getTimeZone() {
- return get(TaskConstants.TASK_FILE_TIME_ZONE);
+ return get(TaskConstants.TASK_TIME_ZONE);
}
public TaskStateEnum getState() {
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
index 2e16752904..eea64d176d 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/CommonConstants.java
@@ -75,7 +75,7 @@ public class CommonConstants {
public static final String PROXY_SENDER_MAX_TIMEOUT =
"proxy.sender.maxTimeout";
// max timeout in seconds.
- public static final int DEFAULT_PROXY_SENDER_MAX_TIMEOUT = 20;
+ public static final int DEFAULT_PROXY_SENDER_MAX_TIMEOUT = 60;
public static final String PROXY_SENDER_MAX_RETRY =
"proxy.sender.maxRetry";
public static final int DEFAULT_PROXY_SENDER_MAX_RETRY = 5;
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/FetcherConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/FetcherConstants.java
index b925cf5732..2214db1fbb 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/FetcherConstants.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/FetcherConstants.java
@@ -23,7 +23,7 @@ package org.apache.inlong.agent.constant;
public class FetcherConstants {
public static final String AGENT_FETCHER_INTERVAL =
"agent.fetcher.interval";
- public static final int DEFAULT_AGENT_FETCHER_INTERVAL = 10;
+ public static final int DEFAULT_AGENT_FETCHER_INTERVAL = 60;
public static final String AGENT_HEARTBEAT_INTERVAL =
"agent.heartbeat.interval";
public static final int DEFAULT_AGENT_HEARTBEAT_INTERVAL = 10;
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/TaskConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/TaskConstants.java
index eaa71f4751..0498358d2d 100755
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/TaskConstants.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constant/TaskConstants.java
@@ -64,7 +64,7 @@ public class TaskConstants extends CommonConstants {
public static final String TASK_DIR_FILTER_PATTERN =
"task.fileTask.dir.pattern"; // deprecated
public static final String FILE_DIR_FILTER_PATTERNS =
"task.fileTask.dir.patterns";
public static final String TASK_FILE_TIME_OFFSET =
"task.fileTask.timeOffset";
- public static final String TASK_FILE_TIME_ZONE = "task.fileTask.timeZone";
+ public static final String TASK_TIME_ZONE = "task.timeZone";
public static final String TASK_FILE_MAX_WAIT =
"task.fileTask.file.max.wait";
public static final String TASK_CYCLE_UNIT = "task.cycleUnit";
public static final String FILE_TASK_CYCLE_UNIT =
"task.fileTask.cycleUnit";
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/FileTask.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/FileTask.java
index f31ec703c2..df7ab22053 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/FileTask.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/FileTask.java
@@ -112,8 +112,6 @@ public class FileTask {
// '1d' means one day after, '-1d' means one day before
// Null means from current timestamp
private String timeOffset;
- // Asia/Shanghai
- private String timeZone;
// For example: a=b&c=b&e=f
private String additionalAttr;
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/TaskProfileDto.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/TaskProfileDto.java
index 2730ce8f60..5c02954f5d 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/TaskProfileDto.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/TaskProfileDto.java
@@ -147,9 +147,6 @@ public class TaskProfileDto {
if (taskConfig.getTimeOffset() != null) {
fileTask.setTimeOffset(taskConfig.getTimeOffset());
}
- if (taskConfig.getTimeZone() != null) {
- fileTask.setTimeZone(taskConfig.getTimeZone());
- }
if (taskConfig.getAdditionalAttr() != null) {
fileTask.setAddictiveString(taskConfig.getAdditionalAttr());
@@ -413,6 +410,7 @@ public class TaskProfileDto {
task.setState(dataConfig.getState());
task.setPredefinedFields(dataConfig.getPredefinedFields());
task.setCycleUnit(CycleUnitType.REAL_TIME);
+ task.setTimeZone(dataConfig.getTimeZone());
// set sink type
if (dataConfig.getDataReportType() ==
NORMAL_SEND_TO_DATAPROXY.ordinal()) {
@@ -523,6 +521,7 @@ public class TaskProfileDto {
private String predefinedFields;
private Integer state;
private String cycleUnit;
+ private String timeZone;
private FileTask fileTask;
private BinlogJob binlogJob;
diff --git
a/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/AgentBaseTestsHelper.java
b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/AgentBaseTestsHelper.java
index fa6d0ebd66..e0d140315a 100755
---
a/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/AgentBaseTestsHelper.java
+++
b/inlong-agent/agent-core/src/test/java/org/apache/inlong/agent/core/AgentBaseTestsHelper.java
@@ -89,11 +89,11 @@ public class AgentBaseTestsHelper {
dataConfig.setDataReportType(1);
dataConfig.setTaskType(3);
dataConfig.setTaskId(taskId);
+ dataConfig.setTimeZone(timeZone);
dataConfig.setState(state.ordinal());
FileTaskConfig fileTaskConfig = new FileTaskConfig();
fileTaskConfig.setPattern(pattern);
fileTaskConfig.setTimeOffset("0h");
- fileTaskConfig.setTimeZone(timeZone);
fileTaskConfig.setMaxFileCount(100);
fileTaskConfig.setCycleUnit("h");
fileTaskConfig.setRetry(retry);
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 fabc02a139..f37b77bc45 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
@@ -44,13 +44,16 @@ import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import static
org.apache.inlong.agent.constant.AgentConstants.AGENT_CLUSTER_NAME;
import static org.apache.inlong.agent.constant.AgentConstants.AGENT_UNIQ_ID;
import static
org.apache.inlong.agent.constant.AgentConstants.DEFAULT_AGENT_UNIQ_ID;
+import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_FETCHER_INTERVAL;
import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_ADDR;
import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_RETURN_PARAM_DATA;
import static
org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_TASK_HTTP_PATH;
+import static
org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_FETCHER_INTERVAL;
import static
org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_CONFIG_HTTP_PATH;
import static
org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_TASK_HTTP_PATH;
import static
org.apache.inlong.agent.plugin.fetcher.ManagerResultFormatter.getResultData;
@@ -172,45 +175,19 @@ public class ManagerFetcher extends AbstractDaemon
implements ProfileFetcher {
private Runnable taskConfigFetchThread() {
return () -> {
Thread.currentThread().setName("ManagerFetcher");
- int normalTaskId = 100;
- int testState = 0;
- int retryTaskId = 800;
- long count = 1;
while (isRunnable()) {
try {
- /*
- * int configSleepTime =
conf.getInt(AGENT_FETCHER_INTERVAL, DEFAULT_AGENT_FETCHER_INTERVAL);
- *
TimeUnit.SECONDS.sleep(AgentUtils.getRandomBySeed(configSleepTime));
- */
- // fetch task config from manager
- TaskResult taskresult;
- String testDir = conf.get("test.dir", "");
- LOGGER.info("test123 test.dir {}", testDir);
- if (testDir == "") {
- taskresult = getStaticConfig();
- } else {
- if (count % 10 == 0) {
- normalTaskId++;
- retryTaskId++;
- }
- if (testState == 1) {
- testState = 2;
- } else {
- testState = 1;
- }
- taskresult = getTestConfig(testDir, normalTaskId,
retryTaskId, testState);
+ int configSleepTime = conf.getInt(AGENT_FETCHER_INTERVAL,
DEFAULT_AGENT_FETCHER_INTERVAL);
+ TaskResult taskResult = getStaticConfig();
+ if (taskResult != null) {
+ List<TaskProfile> taskProfiles = new ArrayList<>();
+ taskResult.getDataConfigs().forEach((config) -> {
+ TaskProfile profile =
TaskProfile.convertToTaskProfile(config);
+ taskProfiles.add(profile);
+ });
+
agentManager.getTaskManager().submitTaskProfiles(taskProfiles);
}
- if (taskresult == null) {
- continue;
- }
- List<TaskProfile> taskProfiles = new ArrayList<>();
- taskresult.getDataConfigs().forEach((config) -> {
- TaskProfile profile =
TaskProfile.convertToTaskProfile(config);
- taskProfiles.add(profile);
- });
-
agentManager.getTaskManager().submitTaskProfiles(taskProfiles);
- count++;
- AgentUtils.silenceSleepInSeconds(60);
+
TimeUnit.SECONDS.sleep(AgentUtils.getRandomBySeed(configSleepTime));
} catch (Throwable ex) {
LOGGER.warn("exception caught", ex);
ThreadUtils.threadThrowableHandler(Thread.currentThread(),
ex);
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
index 2f7b8e500f..d6fc5acd6c 100644
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/task/file/LogFileTask.java
@@ -147,7 +147,7 @@ public class LogFileTask extends Task {
LOGGER.error("task profile cycle unit must be consistent");
return false;
}
- if (!profile.hasKey(TaskConstants.TASK_FILE_TIME_ZONE)) {
+ if (!profile.hasKey(TaskConstants.TASK_TIME_ZONE)) {
LOGGER.error("task profile needs time zone");
return false;
}
diff --git
a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/AgentBaseTestsHelper.java
b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/AgentBaseTestsHelper.java
index 2e61c6766b..32234574ee 100755
---
a/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/AgentBaseTestsHelper.java
+++
b/inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/AgentBaseTestsHelper.java
@@ -95,12 +95,12 @@ public class AgentBaseTestsHelper {
dataConfig.setDataReportType(1);
dataConfig.setTaskType(3);
dataConfig.setTaskId(taskId);
+ dataConfig.setTimeZone("GMT-8:00");
dataConfig.setState(state.ordinal());
FileTaskConfig fileTaskConfig = new FileTaskConfig();
fileTaskConfig.setPattern(pattern);
fileTaskConfig.setTimeOffset("0d");
// GMT-8:00 same with Asia/Shanghai
- fileTaskConfig.setTimeZone("GMT-8:00");
fileTaskConfig.setMaxFileCount(100);
fileTaskConfig.setCycleUnit(cycleUnit);
fileTaskConfig.setRetry(retry);
