You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by he...@apache.org on 2022/03/05 11:30:53 UTC
[incubator-inlong] branch master updated: [INLONG-2933][Agent][Manager] Change the type of the deliveryTime field from Date to String (#2935)
This is an automated email from the ASF dual-hosted git repository.
healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 098d0ab [INLONG-2933][Agent][Manager] Change the type of the deliveryTime field from Date to String (#2935)
098d0ab is described below
commit 098d0ab8a1d9da3de1fd42d3d9a1219c9c053282
Author: healchow <he...@gmail.com>
AuthorDate: Sat Mar 5 19:30:49 2022 +0800
[INLONG-2933][Agent][Manager] Change the type of the deliveryTime field from Date to String (#2935)
---
.../apache/inlong/agent/conf/TriggerProfile.java | 15 +---
.../apache/inlong/agent/pojo/JobProfileDto.java | 13 ++-
.../org/apache/inlong/agent/utils/HttpManager.java | 52 ++++++------
.../apache/inlong/agent/db/TestBerkeleyDBImp.java | 3 +-
.../org/apache/inlong/common/db/CommandEntity.java | 12 ++-
.../inlong/common/pojo/agent/DataConfig.java | 9 +-
.../service/core/impl/AgentServiceImpl.java | 97 ++++++++++++----------
7 files changed, 95 insertions(+), 106 deletions(-)
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TriggerProfile.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TriggerProfile.java
index 82bc73b..6d1aa79 100755
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TriggerProfile.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/conf/TriggerProfile.java
@@ -21,19 +21,11 @@ import org.apache.inlong.agent.constant.JobConstants;
import org.apache.inlong.agent.pojo.JobProfileDto;
import org.apache.inlong.common.pojo.agent.DataConfig;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
-import java.util.Date;
-
/**
* profile used in trigger. Trigger profile is a special job profile
*/
public class TriggerProfile extends JobProfile {
- private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
-
/**
* Parse a given json string and get a TriggerProfile
*/
@@ -75,11 +67,8 @@ public class TriggerProfile extends JobProfile {
return getInt(JobConstants.JOB_OP);
}
- public Date getDeliveryTime() {
- String dateStr = get(JobConstants.JOB_DELIVERY_TIME);
- LocalDateTime localDateTime = LocalDateTime.parse(dateStr, TIME_FORMATTER);
- Instant instant = localDateTime.atZone(ZoneId.systemDefault()).toInstant();
- return Date.from(instant);
+ public String getDeliveryTime() {
+ return get(JobConstants.JOB_DELIVERY_TIME);
}
}
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
index 1441c1a..ffca852 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/pojo/JobProfileDto.java
@@ -17,19 +17,17 @@
package org.apache.inlong.agent.pojo;
-import static java.util.Objects.requireNonNull;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_HOST;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_PORT;
-
-import com.fasterxml.jackson.annotation.JsonFormat;
import com.google.gson.Gson;
-import java.util.Date;
import lombok.Data;
import org.apache.inlong.agent.conf.AgentConfiguration;
import org.apache.inlong.agent.conf.TriggerProfile;
import org.apache.inlong.common.enums.TaskTypeEnum;
import org.apache.inlong.common.pojo.agent.DataConfig;
+import static java.util.Objects.requireNonNull;
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_HOST;
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_VIP_HTTP_PORT;
+
@Data
public class JobProfileDto {
@@ -214,8 +212,7 @@ public class JobProfileDto {
private String name;
private String op;
private String retryTime;
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
- private Date deliveryTime;
+ private String deliveryTime;
private String uuid;
private FileJob fileJob;
diff --git a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
index 4c747d1..86e95a9 100644
--- a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
+++ b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/utils/HttpManager.java
@@ -17,13 +17,6 @@
package org.apache.inlong.agent.utils;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_HTTP_APPLICATION_JSON;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_HTTP_SUCCESS_CODE;
-import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_REQUEST_TIMEOUT;
-import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_REQUEST_TIMEOUT;
-
-import java.util.concurrent.TimeUnit;
-
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.http.client.config.RequestConfig;
@@ -38,19 +31,33 @@ import org.apache.inlong.agent.conf.AgentConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_HTTP_APPLICATION_JSON;
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_HTTP_SUCCESS_CODE;
+import static org.apache.inlong.agent.constant.FetcherConstants.AGENT_MANAGER_REQUEST_TIMEOUT;
+import static org.apache.inlong.agent.constant.FetcherConstants.DEFAULT_AGENT_MANAGER_REQUEST_TIMEOUT;
+
public class HttpManager {
private static final Logger LOGGER = LoggerFactory.getLogger(HttpManager.class);
- private final CloseableHttpClient httpClient;
- private static Gson gson;
+ private static final Gson gson;
static {
- final GsonBuilder gsonBuilder = new GsonBuilder();
+ final GsonBuilder gsonBuilder = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss");
gson = gsonBuilder.create();
}
+ private final CloseableHttpClient httpClient;
+
+ public HttpManager(AgentConfiguration conf) {
+ httpClient = constructHttpClient(conf.getInt(AGENT_MANAGER_REQUEST_TIMEOUT,
+ DEFAULT_AGENT_MANAGER_REQUEST_TIMEOUT));
+ }
+
/**
* construct http client
+ *
* @param timeout - timeout setting
* @return closeable timeout
*/
@@ -60,18 +67,13 @@ public class HttpManager {
}
long timeoutInMs = TimeUnit.SECONDS.toMillis(timeout);
RequestConfig requestConfig = RequestConfig.custom()
- .setConnectTimeout((int) timeoutInMs)
- .setSocketTimeout((int) timeoutInMs).build();
+ .setConnectTimeout((int) timeoutInMs)
+ .setSocketTimeout((int) timeoutInMs).build();
HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
httpClientBuilder.setDefaultRequestConfig(requestConfig);
return httpClientBuilder.build();
}
- public HttpManager(AgentConfiguration conf) {
- httpClient = constructHttpClient(conf.getInt(AGENT_MANAGER_REQUEST_TIMEOUT,
- DEFAULT_AGENT_MANAGER_REQUEST_TIMEOUT));
- }
-
public String doSentPost(String url, Object dto) {
try {
HttpPost post = getHttpPost(url);
@@ -81,7 +83,7 @@ public class HttpManager {
CloseableHttpResponse response = httpClient.execute(post);
String returnStr = EntityUtils.toString(response.getEntity());
if (returnStr != null && !returnStr.isEmpty()
- && response.getStatusLine().getStatusCode() == 200) {
+ && response.getStatusLine().getStatusCode() == 200) {
LOGGER.info("request url {}, dto: {}, return str {}", url, dto, returnStr);
return returnStr;
}
@@ -92,7 +94,7 @@ public class HttpManager {
return null;
}
- public String toJsonStr(Object obj) {
+ public String toJsonStr(Object obj) {
return gson.toJson(obj);
}
@@ -102,7 +104,7 @@ public class HttpManager {
CloseableHttpResponse response = httpClient.execute(get);
String returnStr = EntityUtils.toString(response.getEntity());
if (returnStr != null && !returnStr.isEmpty()
- && response.getStatusLine().getStatusCode() == AGENT_HTTP_SUCCESS_CODE) {
+ && response.getStatusLine().getStatusCode() == AGENT_HTTP_SUCCESS_CODE) {
return returnStr;
}
} catch (Exception e) {
@@ -114,22 +116,16 @@ public class HttpManager {
/**
* get http post, the tauth params should be passed
- * @param url
- * @return
*/
private HttpPost getHttpPost(String url) {
- HttpPost httpPost = new HttpPost(url);
- return httpPost;
+ return new HttpPost(url);
}
/**
* get http get, the tauth params should be passed
- * @param url
- * @return
*/
private HttpGet getHttpGet(String url) {
- HttpGet httpGet = new HttpGet(url);
- return httpGet;
+ return new HttpGet(url);
}
}
diff --git a/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/db/TestBerkeleyDBImp.java b/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/db/TestBerkeleyDBImp.java
index ebecf00..2fda763 100755
--- a/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/db/TestBerkeleyDBImp.java
+++ b/inlong-agent/agent-common/src/test/java/org/apache/inlong/agent/db/TestBerkeleyDBImp.java
@@ -24,7 +24,6 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
-import java.util.Date;
import java.util.List;
public class TestBerkeleyDBImp {
@@ -81,7 +80,7 @@ public class TestBerkeleyDBImp {
@Test
public void testCommandDb() {
- CommandEntity commandEntity = new CommandEntity("1", 0, false, 1, new Date());
+ CommandEntity commandEntity = new CommandEntity("1", 0, false, 1, "2022-03-05 12:10:10");
db.putCommand(commandEntity);
CommandEntity command = db.getCommand("1");
Assert.assertEquals("1", command.getId());
diff --git a/inlong-common/src/main/java/org/apache/inlong/common/db/CommandEntity.java b/inlong-common/src/main/java/org/apache/inlong/common/db/CommandEntity.java
index f9cf516..d3c6884 100644
--- a/inlong-common/src/main/java/org/apache/inlong/common/db/CommandEntity.java
+++ b/inlong-common/src/main/java/org/apache/inlong/common/db/CommandEntity.java
@@ -17,7 +17,6 @@
package org.apache.inlong.common.db;
-import com.fasterxml.jackson.annotation.JsonFormat;
import com.sleepycat.persist.model.Entity;
import com.sleepycat.persist.model.PrimaryKey;
import com.sleepycat.persist.model.Relationship;
@@ -26,8 +25,9 @@ import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
-import java.util.Date;
-
+/**
+ * The entity of task command, used for Agent to interact with Manager and BDB.
+ */
@Entity(version = 1)
@Data
@AllArgsConstructor
@@ -40,12 +40,10 @@ public class CommandEntity {
@SecondaryKey(relate = Relationship.MANY_TO_ONE)
private boolean isAcked;
private Integer taskId;
-
/**
- * The task delivery time
+ * The task delivery time, format is 'yyyy-MM-dd HH:mm:ss'.
*/
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
- private Date deliveryTime;
+ private String deliveryTime;
public static String generateCommandId(String taskId, int opType) {
return taskId + opType;
diff --git a/inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/DataConfig.java b/inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/DataConfig.java
index 4275f41..a400e11 100644
--- a/inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/DataConfig.java
+++ b/inlong-common/src/main/java/org/apache/inlong/common/pojo/agent/DataConfig.java
@@ -17,11 +17,8 @@
package org.apache.inlong.common.pojo.agent;
-import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
-import java.util.Date;
-
/**
* The task config for agent.
*/
@@ -39,8 +36,10 @@ public class DataConfig {
private String snapshot;
private Integer syncSend;
private String extParams;
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
- private Date deliveryTime;
+ /**
+ * The task delivery time, format is 'yyyy-MM-dd HH:mm:ss'.
+ */
+ private String deliveryTime;
public boolean isValid() {
return true;
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
index e135a15..87790fb 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
@@ -61,6 +61,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
@@ -72,6 +76,7 @@ import java.util.stream.Collectors;
public class AgentServiceImpl implements AgentService {
private static final Logger LOGGER = LoggerFactory.getLogger(AgentServiceImpl.class);
+ private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private static final int UNISSUED_STATUS = 2;
private static final int ISSUED_STATUS = 3;
@@ -116,6 +121,52 @@ public class AgentServiceImpl implements AgentService {
}
/**
+ * Update the task status by the request
+ */
+ private void updateTaskStatus(TaskRequest request) {
+ if (CollectionUtils.isEmpty(request.getCommandInfo())) {
+ LOGGER.warn("task result was empty, just return");
+ return;
+ }
+
+ for (CommandEntity command : request.getCommandInfo()) {
+ Integer taskId = command.getTaskId();
+ StreamSourceEntity current = sourceMapper.selectByPrimaryKey(taskId);
+ if (current == null) {
+ continue;
+ }
+
+ LocalDateTime localDateTime = LocalDateTime.parse(command.getDeliveryTime(), TIME_FORMATTER);
+ Instant instant = localDateTime.atZone(ZoneId.systemDefault()).toInstant();
+ if (current.getModifyTime().getTime() - instant.toEpochMilli() > maxModifyTime) {
+ LOGGER.warn("task {} receive result delay more than {} ms, skip it", taskId, maxModifyTime);
+ continue;
+ }
+
+ int result = command.getCommandResult();
+ int previousStatus = current.getStatus();
+ int nextStatus = SourceState.SOURCE_NORMAL.getCode();
+ // Change the status from 30x to normal / disable / frozen
+ if (previousStatus / 100 == ISSUED_STATUS) {
+ if (Constants.RESULT_SUCCESS == result) {
+ if (SourceState.TEMP_TO_NORMAL.contains(previousStatus)) {
+ nextStatus = SourceState.SOURCE_NORMAL.getCode();
+ } else if (SourceState.BEEN_ISSUED_DELETE.getCode() == previousStatus) {
+ nextStatus = SourceState.SOURCE_DISABLE.getCode();
+ } else if (SourceState.BEEN_ISSUED_FROZEN.getCode() == previousStatus) {
+ nextStatus = SourceState.SOURCE_FROZEN.getCode();
+ }
+ } else if (Constants.RESULT_FAIL == result) {
+ nextStatus = SourceState.SOURCE_FAILED.getCode();
+ }
+
+ sourceMapper.updateStatus(taskId, nextStatus);
+ }
+ // Other tasks with status 20x will change to 30x in next getTaskResult method
+ }
+ }
+
+ /**
* Get task result by the request
*/
private TaskResult getTaskResult(TaskRequest request) {
@@ -133,7 +184,9 @@ public class AgentServiceImpl implements AgentService {
dataConfig.setIp(entity.getAgentIp());
dataConfig.setUuid(entity.getUuid());
dataConfig.setExtParams(entity.getExtParams());
- dataConfig.setSnapshot(entity.getSnapshot());
+ LocalDateTime dateTime = LocalDateTime.ofInstant(entity.getModifyTime().toInstant(),
+ ZoneId.systemDefault());
+ dataConfig.setDeliveryTime(dateTime.format(TIME_FORMATTER));
String groupId = entity.getInlongGroupId();
String streamId = entity.getInlongStreamId();
@@ -164,48 +217,6 @@ public class AgentServiceImpl implements AgentService {
}
}
- /**
- * Update the task status by the request
- */
- private void updateTaskStatus(TaskRequest request) {
- if (CollectionUtils.isEmpty(request.getCommandInfo())) {
- LOGGER.warn("task result was empty, just return");
- return;
- }
-
- for (CommandEntity command : request.getCommandInfo()) {
- Integer taskId = command.getTaskId();
- StreamSourceEntity current = sourceMapper.selectByPrimaryKey(taskId);
- if (current == null) {
- continue;
- }
-
- if (current.getModifyTime().getTime() - command.getDeliveryTime().getTime() > maxModifyTime) {
- LOGGER.warn("task {} receive result delay more than {} ms, skip it", taskId, maxModifyTime);
- continue;
- }
-
- int result = command.getCommandResult();
- int previousStatus = current.getStatus();
- int nextStatus = SourceState.SOURCE_NORMAL.getCode();
- if (previousStatus / 100 == UNISSUED_STATUS) {
- if (Constants.RESULT_SUCCESS == result) {
- if (SourceState.TEMP_TO_NORMAL.contains(previousStatus)) {
- nextStatus = SourceState.SOURCE_NORMAL.getCode();
- } else if (SourceState.BEEN_ISSUED_DELETE.getCode() == previousStatus) {
- nextStatus = SourceState.SOURCE_DISABLE.getCode();
- } else if (SourceState.BEEN_ISSUED_FROZEN.getCode() == previousStatus) {
- nextStatus = SourceState.SOURCE_FROZEN.getCode();
- }
- } else if (Constants.RESULT_FAIL == result) {
- nextStatus = SourceState.SOURCE_FAILED.getCode();
- }
-
- sourceMapper.updateStatus(taskId, nextStatus);
- }
- }
- }
-
private List<CmdConfig> getAgentCmdConfigs(TaskRequest taskRequest) {
return sourceCmdConfigMapper.queryCmdByAgentIp(taskRequest.getAgentIp()).stream().map(cmd -> {
CmdConfig cmdConfig = new CmdConfig();