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/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new c3f534d [INLONG-710] add tid in jobconf
c3f534d is described below
commit c3f534d74d76984c73da8bb2349bfe8b76e13565
Author: stingpeng <[email protected]>
AuthorDate: Sat Jul 10 18:14:47 2021 +0800
[INLONG-710] add tid in jobconf
---
.../main/java/org/apache/inlong/agent/constants/CommonConstants.java | 4 ++++
.../apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java | 1 +
.../org/apache/inlong/agent/plugin/fetcher/dtos/JobProfileDto.java | 1 +
.../src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java | 4 ++++
4 files changed, 10 insertions(+)
diff --git
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
index 7e3fc6d..3554de9 100644
---
a/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
+++
b/inlong-agent/agent-common/src/main/java/org/apache/inlong/agent/constants/CommonConstants.java
@@ -27,6 +27,9 @@ public class CommonConstants {
public static final String PROXY_BID = "proxy.bid";
public static final String POSITION_SUFFIX = ".position";
+ public static final String PROXY_TID = "proxy.tid";
+
+
public static final String PROXY_LOCAL_HOST = "proxy.localHost";
public static final String DEFAULT_PROXY_LOCALHOST =
AgentUtils.getLocalIp();
@@ -76,6 +79,7 @@ public class CommonConstants {
public static final long DEFAULT_PROXY_RETRY_SLEEP = 500;
public static final String PROXY_KEY_BID = "bid";
+ public static final String PROXY_KEY_TID = "tid";
public static final String PROXY_KEY_ID = "id";
public static final String PROXY_KEY_AGENT_IP = "agentip";
public static final String PROXY_OCEANUS_F = "f";
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
index fc64c24..6af439e 100755
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/ManagerResultFormatter.java
@@ -128,6 +128,7 @@ public class ManagerResultFormatter {
manager.setHost(agentConf.get(AGENT_MANAGER_VIP_HTTP_HOST));
manager.setPort(agentConf.get(AGENT_MANAGER_VIP_HTTP_PORT));
proxy.setBid(dataConfigs.getBusinessIdentifier());
+ proxy.setTid(dataConfigs.getDataStreamIdentifier());
proxy.setManager(manager);
return proxy;
}
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/dtos/JobProfileDto.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/dtos/JobProfileDto.java
index 17fda09..0fee5e1 100644
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/dtos/JobProfileDto.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/fetcher/dtos/JobProfileDto.java
@@ -65,6 +65,7 @@ public class JobProfileDto {
@Data
public static class Proxy {
private String bid;
+ private String tid;
private Manager manager;
}
}
\ No newline at end of file
diff --git
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
index e3521b4..1f5132d 100755
---
a/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
+++
b/inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sinks/ProxySink.java
@@ -22,6 +22,7 @@ import static
org.apache.inlong.agent.constants.CommonConstants.PROXY_KEY_AGENT_
import static org.apache.inlong.agent.constants.CommonConstants.PROXY_KEY_ID;
import static
org.apache.inlong.agent.constants.CommonConstants.PROXY_OCEANUS_BL;
import static
org.apache.inlong.agent.constants.CommonConstants.PROXY_OCEANUS_F;
+import static org.apache.inlong.agent.constants.CommonConstants.PROXY_TID;
import static
org.apache.inlong.agent.constants.JobConstants.PROXY_BATCH_FLUSH_INTERVAL;
import static
org.apache.inlong.agent.constants.JobConstants.PROXY_PACKAGE_MAX_SIZE;
import static
org.apache.inlong.agent.constants.JobConstants.PROXY_PACKAGE_MAX_TIMEOUT_MS;
@@ -65,6 +66,7 @@ public class ProxySink implements Sink {
private static final Logger LOGGER =
LoggerFactory.getLogger(ProxySink.class);
private SenderManager senderManager;
private String bid;
+ private String tid;
private String sourceFile;
private String jobInstanceId;
private int maxBatchSize;
@@ -85,6 +87,7 @@ public class ProxySink implements Sink {
public void write(Message message) {
if (message != null) {
message.getHeader().put(CommonConstants.PROXY_KEY_BID, bid);
+ message.getHeader().put(CommonConstants.PROXY_KEY_TID, tid);
if (!(message instanceof EndMessage)) {
ProxyMessage proxyMessage = ProxyMessage.parse(message);
// add proxy message to cache.
@@ -153,6 +156,7 @@ public class ProxySink implements Sink {
dataTime =
AgentUtils.timeStrConvertToMillSec(jobConf.get(JOB_DATA_TIME, ""),
jobConf.get(JOB_CYCLE_UNIT, ""));
bid = jobConf.get(PROXY_BID);
+ tid = jobConf.get(PROXY_TID);
executorService.execute(flushCache());
senderManager = new SenderManager(jobConf, bid, sourceFile);
try {
