This is an automated email from the ASF dual-hosted git repository.
gosonzhang 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 2f62a68c12 [INLONG-8161][DataProxy] Optimize BatchPackProfile related
classes implementation (#8187)
2f62a68c12 is described below
commit 2f62a68c121e1ddd89af50c8aaf8f969b8699186
Author: Goson Zhang <[email protected]>
AuthorDate: Thu Jun 8 14:45:48 2023 +0800
[INLONG-8161][DataProxy] Optimize BatchPackProfile related classes
implementation (#8187)
---
inlong-dataproxy/conf/common.properties | 2 +
.../apache/inlong/dataproxy/base/SinkRspEvent.java | 4 +
.../dataproxy/config/CommonConfigHolder.java | 22 ++-
.../inlong/dataproxy/config/ConfigManager.java | 31 +++-
.../dataproxy/config/holder/MetaConfigHolder.java | 7 +
.../dataproxy/config/holder/PropertiesHolder.java | 2 +-
.../inlong/dataproxy/consts/ConfigConstants.java | 1 -
.../inlong/dataproxy/sink/common/SinkContext.java | 4 +-
.../inlong/dataproxy/sink/mq/BatchPackManager.java | 35 ++--
.../inlong/dataproxy/sink/mq/BatchPackProfile.java | 109 +-----------
.../sink/mq/MessageQueueClusterProducer.java | 6 +-
.../dataproxy/sink/mq/MessageQueueHandler.java | 4 +-
.../sink/mq/MessageQueueZoneProducer.java | 32 ++--
.../dataproxy/sink/mq/MessageQueueZoneSink.java | 26 ++-
.../sink/mq/MessageQueueZoneSinkContext.java | 29 +--
.../dataproxy/sink/mq/MessageQueueZoneWorker.java | 12 +-
.../dataproxy/sink/mq/OrderBatchPackProfileV0.java | 141 ---------------
.../mq/{BatchPackProfile.java => PackProfile.java} | 146 +++++----------
.../sink/mq/SimpleBatchPackProfileV0.java | 91 ----------
.../dataproxy/sink/mq/SimplePackProfile.java | 198 +++++++++++++++++++++
.../dataproxy/sink/mq/kafka/KafkaHandler.java | 127 +++++--------
.../dataproxy/sink/mq/pulsar/PulsarHandler.java | 114 +++++-------
.../inlong/dataproxy/sink/mq/tube/TubeHandler.java | 118 +++++-------
.../dataproxy/source2/InLongMessageHandler.java | 11 +-
.../dataproxy/source2/v0msg/AbsV0MsgCodec.java | 1 +
.../config/holder/TestCommonConfigHolder.java | 2 +-
.../config/holder/TestMetaConfigHolder.java | 2 +
.../src/test/resources/common.properties | 4 +-
28 files changed, 536 insertions(+), 745 deletions(-)
diff --git a/inlong-dataproxy/conf/common.properties
b/inlong-dataproxy/conf/common.properties
index e9ca6e57cb..8b16900c2a 100644
--- a/inlong-dataproxy/conf/common.properties
+++ b/inlong-dataproxy/conf/common.properties
@@ -28,6 +28,8 @@ proxy.cluster.extTag=default=true
proxy.cluster.inCharges=admin
# synchronize interval of meta config (millisecond)
meta.config.sync.interval.ms=10000
+# whether to startup using the local metadata.json file without connecting to
the Manager
+startup.using.local.meta.file.enable=false
# metric config
metricDomains=DataProxy
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
index d57ec9645e..e1db45970d 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/SinkRspEvent.java
@@ -56,6 +56,10 @@ public class SinkRspEvent implements Event {
event.setBody(bytes);
}
+ public Event getEvent() {
+ return this.event;
+ }
+
/**
* Get event reported channel context
*
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
index 3fd6609dd2..02afd1a80e 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/CommonConfigHolder.java
@@ -72,7 +72,12 @@ public class CommonConfigHolder {
private static final String KEY_META_CONFIG_SYNC_INTERVAL_MS =
"meta.config.sync.interval.ms";
private static final String KEY_CONFIG_CHECK_INTERVAL_MS =
"configCheckInterval";
public static final long VAL_DEF_CONFIG_SYNC_INTERVAL_MS = 60000L;
- // Whether to accept messages without mapping between groupId/streamId and
topic
+ public static final long VAL_MIN_CONFIG_SYNC_INTERVAL_MS = 10000L;
+ // whether to startup using the local metadata.json file without
connecting to the Manager
+ private static final String KEY_ENABLE_STARTUP_USING_LOCAL_META_FILE =
+ "startup.using.local.meta.file.enable";
+ public static final boolean VAL_DEF_ENABLE_STARTUP_USING_LOCAL_META_FILE =
false;
+ // whether to accept messages without mapping between groupId/streamId and
topic
public static final String KEY_NOTFOUND_TOPIC_ACCEPT =
"source.topic.notfound.accept";
public static final boolean VAL_DEF_NOTFOUND_TOPIC_ACCEPT = false;
// whether enable whitelist, optional field.
@@ -147,6 +152,7 @@ public class CommonConfigHolder {
private IManagerIpListParser ipListParser = null;
private String managerAuthSecretId = "";
private String managerAuthSecretKey = "";
+ private boolean enableStartupUsingLocalMetaFile =
VAL_DEF_ENABLE_STARTUP_USING_LOCAL_META_FILE;
private long metaConfigSyncInvlMs = VAL_DEF_CONFIG_SYNC_INTERVAL_MS;
private boolean enableAudit = VAL_DEF_ENABLE_AUDIT;
private final HashSet<String> auditProxys = new HashSet<>();
@@ -301,6 +307,10 @@ public class CommonConfigHolder {
return maxBufferQueueSizeKb;
}
+ public boolean isEnableStartupUsingLocalMetaFile() {
+ return enableStartupUsingLocalMetaFile;
+ }
+
public String getEventHandler() {
return eventHandler;
}
@@ -360,7 +370,15 @@ public class CommonConfigHolder {
tmpValue = this.props.get(KEY_CONFIG_CHECK_INTERVAL_MS);
}
if (StringUtils.isNotEmpty(tmpValue)) {
- this.metaConfigSyncInvlMs = NumberUtils.toLong(tmpValue.trim(),
VAL_DEF_CONFIG_SYNC_INTERVAL_MS);
+ long tmpSyncInvMs = NumberUtils.toLong(tmpValue.trim(),
VAL_DEF_CONFIG_SYNC_INTERVAL_MS);
+ if (tmpSyncInvMs >= VAL_MIN_CONFIG_SYNC_INTERVAL_MS) {
+ this.metaConfigSyncInvlMs = tmpSyncInvMs;
+ }
+ }
+ // read enable startup using local meta file
+ tmpValue = this.props.get(KEY_ENABLE_STARTUP_USING_LOCAL_META_FILE);
+ if (StringUtils.isNotEmpty(tmpValue)) {
+ this.enableStartupUsingLocalMetaFile =
"TRUE".equalsIgnoreCase(tmpValue.trim());
}
// read whether accept msg without topic
tmpValue = this.props.get(KEY_NOTFOUND_TOPIC_ACCEPT);
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/ConfigManager.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/ConfigManager.java
index dea37b2c3d..2507419686 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/ConfigManager.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/ConfigManager.java
@@ -62,6 +62,8 @@ public class ConfigManager {
private static final Logger LOG =
LoggerFactory.getLogger(ConfigManager.class);
public static final List<ConfigHolder> CONFIG_HOLDER_LIST = new
ArrayList<>();
+ // whether handshake manager ok
+ public static final AtomicBoolean handshakeManagerOk = new
AtomicBoolean(false);
private static volatile boolean isInit = false;
private static ConfigManager instance = null;
// node weight configure
@@ -77,7 +79,7 @@ public class ConfigManager {
// source report configure holder
private final SourceReportConfigHolder sourceReportConfigHolder = new
SourceReportConfigHolder();
// mq clusters ready
- private final AtomicBoolean mqClusterReady = new AtomicBoolean(false);
+ private volatile boolean mqClusterReady = false;
/**
* get instance for config manager
@@ -95,8 +97,8 @@ public class ConfigManager {
ReloadConfigWorker reloadProperties =
ReloadConfigWorker.create(instance);
reloadProperties.setDaemon(true);
reloadProperties.start();
+ isInit = true;
}
- isInit = true;
}
return instance;
}
@@ -209,11 +211,11 @@ public class ConfigManager {
}
public boolean isMqClusterReady() {
- return mqClusterReady.get();
+ return mqClusterReady;
}
- public void updMqClusterStatus(boolean isStarted) {
- mqClusterReady.set(isStarted);
+ public void setMqClusterReady() {
+ mqClusterReady = true;
}
/**
@@ -243,6 +245,7 @@ public class ConfigManager {
long count = 0;
long startTime;
long wstTime;
+ boolean fisrtCheck = true;
LOG.info("Reload-Config Worker started!");
while (isRunning) {
count += 1;
@@ -254,10 +257,17 @@ public class ConfigManager {
holder.executeCallbacks();
}
}
- // wait for 3 * check-time to update remote config
- if (count % 3 == 0) {
+ // connect to manager
+ if (fisrtCheck) {
+ fisrtCheck = false;
checkRemoteConfig();
count = 0;
+ } else {
+ // wait for 3 * check-time to update remote config
+ if (count % 3 == 0) {
+ checkRemoteConfig();
+ count = 0;
+ }
}
// check processing time
wstTime = System.currentTimeMillis() - startTime;
@@ -339,7 +349,7 @@ public class ConfigManager {
}
httpPost.setEntity(HttpUtils.getEntity(request));
// request with post
- LOG.debug("Start to request {} to get config info with params
{}", url, request);
+ LOG.info("Start to request {} to get config info with params
{}", url, request);
CloseableHttpResponse response = httpClient.execute(httpPost);
String returnStr = EntityUtils.toString(response.getEntity());
if (response.getStatusLine().getStatusCode() != 200) {
@@ -366,7 +376,10 @@ public class ConfigManager {
return true;
}
// update meta configure
- configManager.updateMetaConfigInfo(proxyResponse.getMd5(),
returnStr);
+ if (configManager.updateMetaConfigInfo(proxyResponse.getMd5(),
returnStr)) {
+ ConfigManager.handshakeManagerOk.set(true);
+ LOG.info("Get meta config info and set handshake status is
ok!");
+ }
return true;
} catch (Throwable ex) {
LOG.error("Request remote manager failure", ex);
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java
index 9b098c17c2..9c7078ee48 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java
@@ -26,6 +26,7 @@ import org.apache.inlong.common.pojo.dataproxy.InLongIdObject;
import org.apache.inlong.common.pojo.dataproxy.ProxyClusterObject;
import org.apache.inlong.dataproxy.config.CommonConfigHolder;
import org.apache.inlong.dataproxy.config.ConfigHolder;
+import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
import org.apache.inlong.dataproxy.config.pojo.CacheType;
import org.apache.inlong.dataproxy.config.pojo.DataType;
@@ -211,6 +212,12 @@ public class MetaConfigHolder extends ConfigHolder {
LOG.warn("Load failed json config from {}, malformed content,
data is null", getFileName());
return false;
}
+ if
(!CommonConfigHolder.getInstance().isEnableStartupUsingLocalMetaFile()
+ && !ConfigManager.handshakeManagerOk.get()) {
+ LOG.info("Failed to load json config from {}, don't obtain
metadata from the Manager,"
+ + " and the startup via the cache file is false",
getFileName());
+ return false;
+ }
// update cache data
if (updateCacheData(clusterObj)) {
// update cache string
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/PropertiesHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/PropertiesHolder.java
index 0426f60b05..68ebd0e126 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/PropertiesHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/PropertiesHolder.java
@@ -84,7 +84,7 @@ public abstract class PropertiesHolder extends ConfigHolder {
try {
Map<String, String> loadMap = loadConfigFromFile();
if (loadMap == null || loadMap.isEmpty()) {
- LOG.info("Load changed properties {}, but no records
configured", getFileName());
+ LOG.debug("Load changed properties {}, but no records
configured", getFileName());
return false;
}
// filter blank items
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/ConfigConstants.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/ConfigConstants.java
index 0dd453e3e1..ea8f51cee1 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/ConfigConstants.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/consts/ConfigConstants.java
@@ -124,7 +124,6 @@ public class ConfigConstants {
public static final String RECOVER_THREAD_COUNT = "recover_thread_count";
public static final String MANAGER_PATH = "/inlong/manager/openapi";
- public static final String MANAGER_GET_CONFIG_PATH =
"/dataproxy/getConfig";
public static final String MANAGER_GET_ALL_CONFIG_PATH =
"/dataproxy/getAllConfig";
public static final String MANAGER_HEARTBEAT_REPORT = "/heartbeat/report";
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/SinkContext.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/SinkContext.java
index 7edc0b873e..4ebe570975 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/SinkContext.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/SinkContext.java
@@ -24,8 +24,8 @@ import org.apache.inlong.dataproxy.config.CommonConfigHolder;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
import org.apache.inlong.dataproxy.consts.AttrConstants;
import org.apache.inlong.dataproxy.metrics.DataProxyMetricItemSet;
-import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueHandler;
+import org.apache.inlong.dataproxy.sink.mq.PackProfile;
import org.apache.inlong.dataproxy.sink.mq.pulsar.PulsarHandler;
import org.apache.inlong.dataproxy.utils.BufferQueue;
@@ -275,7 +275,7 @@ public class SinkContext {
* createBufferQueue
* @return
*/
- public static BufferQueue<BatchPackProfile> createBufferQueue() {
+ public static BufferQueue<PackProfile> createBufferQueue() {
return new
BufferQueue<>(CommonConfigHolder.getInstance().getMaxBufferQueueSizeKb());
}
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java
index 40f211ec39..84a1f5c976 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java
@@ -17,10 +17,13 @@
package org.apache.inlong.dataproxy.sink.mq;
+import org.apache.inlong.common.msg.AttributeConstants;
import org.apache.inlong.dataproxy.utils.BufferQueue;
+import org.apache.inlong.sdk.commons.protocol.InlongId;
import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
import org.apache.inlong.sdk.commons.protocol.ProxyPackEvent;
+import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flume.Context;
import org.apache.flume.event.SimpleEvent;
import org.slf4j.Logger;
@@ -28,6 +31,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -50,8 +54,8 @@ public class BatchPackManager {
private final long dispatchTimeout;
private final long maxPackCount;
private final long maxPackSize;
- private BufferQueue<BatchPackProfile> dispatchQueue;
- private ConcurrentHashMap<String, BatchPackProfile> profileCache = new
ConcurrentHashMap<>();
+ private BufferQueue<PackProfile> dispatchQueue;
+ private ConcurrentHashMap<String, PackProfile> profileCache = new
ConcurrentHashMap<>();
// flag that manager need to output overtime data.
private AtomicBoolean needOutputOvertimeData = new AtomicBoolean(false);
private AtomicLong inCounter = new AtomicLong(0);
@@ -63,7 +67,7 @@ public class BatchPackManager {
* @param context
* @param dispatchQueue
*/
- public BatchPackManager(Context context, BufferQueue<BatchPackProfile>
dispatchQueue) {
+ public BatchPackManager(Context context, BufferQueue<PackProfile>
dispatchQueue) {
this.dispatchQueue = dispatchQueue;
this.dispatchTimeout = context.getLong(KEY_DISPATCH_TIMEOUT,
DEFAULT_DISPATCH_TIMEOUT);
this.maxPackCount = context.getLong(KEY_DISPATCH_MAX_PACKCOUNT,
DEFAULT_DISPATCH_MAX_PACKCOUNT);
@@ -80,7 +84,7 @@ public class BatchPackManager {
long dispatchTime = event.getMsgTime() - event.getMsgTime() %
MINUTE_MS;
String dispatchKey = eventUid + "." + dispatchTime;
// find dispatch profile
- BatchPackProfile dispatchProfile = this.profileCache.get(dispatchKey);
+ PackProfile dispatchProfile = this.profileCache.get(dispatchKey);
if (dispatchProfile == null) {
dispatchProfile = new BatchPackProfile(eventUid,
event.getInlongGroupId(), event.getInlongStreamId(),
dispatchTime);
@@ -91,7 +95,7 @@ public class BatchPackManager {
if (!addResult) {
BatchPackProfile newDispatchProfile = new
BatchPackProfile(eventUid, event.getInlongGroupId(),
event.getInlongStreamId(), dispatchTime);
- BatchPackProfile oldDispatchProfile =
this.profileCache.put(dispatchKey, newDispatchProfile);
+ PackProfile oldDispatchProfile =
this.profileCache.put(dispatchKey, newDispatchProfile);
this.dispatchQueue.acquire(oldDispatchProfile.getSize());
this.dispatchQueue.offer(oldDispatchProfile);
outCounter.addAndGet(dispatchProfile.getCount());
@@ -141,10 +145,17 @@ public class BatchPackManager {
* @param event
*/
public void addSimpleEvent(SimpleEvent event) {
- BatchPackProfile dispatchProfile =
SimpleBatchPackProfileV0.create(event);
- this.dispatchQueue.acquire(dispatchProfile.getSize());
- this.dispatchQueue.offer(dispatchProfile);
- outCounter.addAndGet(dispatchProfile.getCount());
+ Map<String, String> headers = event.getHeaders();
+ String inlongGroupId = headers.get(AttributeConstants.GROUP_ID);
+ String inlongStreamId = headers.get(AttributeConstants.STREAM_ID);
+ String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
+ long msgTime =
NumberUtils.toLong(headers.get(AttributeConstants.DATA_TIME),
System.currentTimeMillis());
+ long dispatchTime = msgTime - msgTime % MINUTE_MS;
+ SimplePackProfile profile = new SimplePackProfile(uid, inlongGroupId,
inlongStreamId, dispatchTime);
+ profile.addEvent(event, maxPackCount, maxPackSize);
+ this.dispatchQueue.acquire(profile.getSize());
+ this.dispatchQueue.offer(profile);
+ outCounter.addAndGet(profile.getCount());
inCounter.incrementAndGet();
}
@@ -163,8 +174,8 @@ public class BatchPackManager {
long createThreshold = currentTime - dispatchTimeout;
List<String> removeKeys = new ArrayList<>();
long eventCount = 0;
- for (Entry<String, BatchPackProfile> entry :
this.profileCache.entrySet()) {
- BatchPackProfile dispatchProfile = entry.getValue();
+ for (Entry<String, PackProfile> entry : this.profileCache.entrySet()) {
+ PackProfile dispatchProfile = entry.getValue();
eventCount += dispatchProfile.getCount();
if (!dispatchProfile.isTimeout(createThreshold)) {
continue;
@@ -173,7 +184,7 @@ public class BatchPackManager {
}
// output
removeKeys.forEach((key) -> {
- BatchPackProfile dispatchProfile = this.profileCache.remove(key);
+ PackProfile dispatchProfile = this.profileCache.remove(key);
if (dispatchProfile != null) {
this.dispatchQueue.acquire(dispatchProfile.getSize());
dispatchQueue.offer(dispatchProfile);
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
index 9abe78f4a0..479b26ee88 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
@@ -17,8 +17,11 @@
package org.apache.inlong.dataproxy.sink.mq;
+import org.apache.inlong.common.enums.DataProxyErrCode;
import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.flume.Event;
+
import java.util.ArrayList;
import java.util.List;
@@ -26,33 +29,21 @@ import java.util.List;
*
* DispatchProfile
*/
-public class BatchPackProfile {
-
- public static final long MINUTE_MS = 60L * 1000;
+public class BatchPackProfile extends PackProfile {
- private final String inlongGroupId;
- private final String inlongStreamId;
- private final String uid;
private List<ProxyEvent> events = new ArrayList<>();
- private long createTime = System.currentTimeMillis();
- private long count = 0;
- private long size = 0;
- private long dispatchTime;
private BatchPackProfileCallback callback;
/**
* Constructor
- *
+ *
* @param uid
* @param inlongGroupId
* @param inlongStreamId
* @param dispatchTime
*/
public BatchPackProfile(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
- this.uid = uid;
- this.inlongGroupId = inlongGroupId;
- this.inlongStreamId = inlongStreamId;
- this.dispatchTime = dispatchTime;
+ super(uid, inlongGroupId, inlongStreamId, dispatchTime);
}
/**
@@ -63,36 +54,17 @@ public class BatchPackProfile {
* @param maxPackSize
* @return
*/
- public boolean addEvent(ProxyEvent event, long maxPackCount, long
maxPackSize) {
+ public boolean addEvent(Event event, long maxPackCount, long maxPackSize) {
long eventLength = event.getBody().length;
if (count >= maxPackCount || (count > 0 && size + eventLength >
maxPackSize)) {
return false;
}
- this.events.add(event);
+ this.events.add((ProxyEvent) event);
this.count++;
this.size += eventLength;
return true;
}
- /**
- * isTimeout
- *
- * @param createThreshold
- * @return
- */
- public boolean isTimeout(long createThreshold) {
- return createThreshold >= createTime;
- }
-
- /**
- * get uid
- *
- * @return the uid
- */
- public String getUid() {
- return uid;
- }
-
/**
* get events
*
@@ -111,69 +83,6 @@ public class BatchPackProfile {
this.events = events;
}
- /**
- * get count
- *
- * @return the count
- */
- public long getCount() {
- return count;
- }
-
- /**
- * set count
- *
- * @param count the count to set
- */
- public void setCount(long count) {
- this.count = count;
- }
-
- /**
- * get size
- *
- * @return the size
- */
- public long getSize() {
- return size;
- }
-
- /**
- * set size
- *
- * @param size the size to set
- */
- public void setSize(long size) {
- this.size = size;
- }
-
- /**
- * get inlongGroupId
- *
- * @return the inlongGroupId
- */
- public String getInlongGroupId() {
- return inlongGroupId;
- }
-
- /**
- * get inlongStreamId
- *
- * @return the inlongStreamId
- */
- public String getInlongStreamId() {
- return inlongStreamId;
- }
-
- /**
- * getDispatchTime
- *
- * @return
- */
- public long getDispatchTime() {
- return dispatchTime;
- }
-
/**
* ack
*/
@@ -187,7 +96,7 @@ public class BatchPackProfile {
* fail
* @return
*/
- public void fail() {
+ public void fail(DataProxyErrCode errCode, String errMsg) {
if (callback != null) {
callback.fail();
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueClusterProducer.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueClusterProducer.java
index e97b51f5f4..46bdabdeee 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueClusterProducer.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueClusterProducer.java
@@ -90,10 +90,10 @@ public class MessageQueueClusterProducer implements
LifecycleAware {
/**
* send
*
- * @param event
+ * @param profile
*/
- public boolean send(BatchPackProfile event) {
- return this.handler.send(event);
+ public boolean send(PackProfile profile) {
+ return this.handler.send(profile);
}
/**
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueHandler.java
index c9afb26ba2..7261da72e8 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueHandler.java
@@ -45,10 +45,10 @@ public interface MessageQueueHandler {
/**
* send
- * @param event
+ * @param profile
* @return
*/
- boolean send(BatchPackProfile event);
+ boolean send(PackProfile profile);
void publishTopic(Set<String> topicSet);
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneProducer.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneProducer.java
index 2b5aff38a4..6ef2576767 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneProducer.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneProducer.java
@@ -41,7 +41,7 @@ public class MessageQueueZoneProducer {
public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueZoneProducer.class);
private static final long MAX_RESERVED_TIME = 60 * 1000L;
- private final String workerName;
+ private final MessageQueueZoneSink zoneSink;
private final MessageQueueZoneSinkContext context;
private final CacheClusterSelector cacheClusterSelector;
@@ -57,11 +57,11 @@ public class MessageQueueZoneProducer {
/**
* Constructor
*
- * @param workerName
+ * @param zoneSink
* @param context
*/
- public MessageQueueZoneProducer(String workerName,
MessageQueueZoneSinkContext context) {
- this.workerName = workerName;
+ public MessageQueueZoneProducer(MessageQueueZoneSink zoneSink,
MessageQueueZoneSinkContext context) {
+ this.zoneSink = zoneSink;
this.context = context;
this.cacheClusterSelector = context.createCacheClusterSelector();
}
@@ -71,7 +71,7 @@ public class MessageQueueZoneProducer {
*/
public void start() {
try {
- LOG.info("start MessageQueueZoneProducer:{}", workerName);
+ LOG.info("start MessageQueueZoneProducer:{}", zoneSink.getName());
this.reloadMetaConfig();
} catch (Exception e) {
LOG.error(e.getMessage(), e);
@@ -140,15 +140,15 @@ public class MessageQueueZoneProducer {
tmpProducer.stop();
}
}
- LOG.info("Clear {}'s expired cluster producer {}", workerName,
expired);
+ LOG.info("Clear {}'s expired cluster producer {}", zoneSink.getName(),
expired);
}
/**
* send
*
- * @param event
+ * @param profile
*/
- public boolean send(BatchPackProfile event) {
+ public boolean send(PackProfile profile) {
String clusterName;
MessageQueueClusterProducer clusterProducer;
readWriteLock.readLock().lock();
@@ -163,7 +163,7 @@ public class MessageQueueZoneProducer {
if (clusterProducer == null) {
continue;
}
- return clusterProducer.send(event);
+ return clusterProducer.send(profile);
} while (true);
} finally {
readWriteLock.readLock().unlock();
@@ -176,7 +176,6 @@ public class MessageQueueZoneProducer {
List<CacheClusterConfig> allConfigList =
ConfigManager.getInstance().getCachedCLusterConfig();
List<CacheClusterConfig> newConfigList =
this.cacheClusterSelector.select(allConfigList);
if (newConfigList == null || newConfigList.size() == 0) {
- LOG.info("Reload {}'s cluster info, but empty", workerName);
return;
}
// check added clusters
@@ -218,7 +217,7 @@ public class MessageQueueZoneProducer {
continue;
}
// create
- tmpCluster = new MessageQueueClusterProducer(workerName,
config, context);
+ tmpCluster = new
MessageQueueClusterProducer(zoneSink.getName(), config, context);
tmpCluster.start();
usingClusterMap.put(config.getClusterName(), tmpCluster);
usingTimeMap.put(config.getClusterName(), curTime);
@@ -264,14 +263,15 @@ public class MessageQueueZoneProducer {
if (!changed) {
return;
}
- if (ConfigManager.getInstance().isMqClusterReady()) {
+ if (zoneSink.isMqClusterStarted()) {
LOG.info("Reload {}'s cluster info, current cluster are {},
removed {}, created {}",
- workerName, lastClusterNames, needRmvs, addedItems);
+ zoneSink.getName(), lastClusterNames, needRmvs,
addedItems);
} else {
+ zoneSink.setMQClusterStarted();
+ ConfigManager.getInstance().setMqClusterReady();
LOG.info(
"Reload {}'s cluster info, and updated sink status,
current cluster are {}, removed {}, created {}",
- workerName, lastClusterNames, needRmvs, addedItems);
- ConfigManager.getInstance().updMqClusterStatus(true);
+ zoneSink.getName(), lastClusterNames, needRmvs,
addedItems);
}
} catch (Throwable e) {
LOG.error("Reload cluster info failure", e);
@@ -284,7 +284,7 @@ public class MessageQueueZoneProducer {
return;
}
LOG.info("Reload {}'s topics changed, current topics are {}, last
topics are {}",
- workerName, curTopicSet, lastRefreshTopics);
+ zoneSink.getName(), curTopicSet, lastRefreshTopics);
lastRefreshTopics.addAll(curTopicSet);
for (MessageQueueClusterProducer clusterProducer :
this.usingClusterMap.values()) {
if (clusterProducer == null) {
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSink.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSink.java
index 47a80b18d4..543a454ba6 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSink.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSink.java
@@ -49,12 +49,14 @@ public class MessageQueueZoneSink extends AbstractSink
implements Configurable,
public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueZoneSink.class);
+ private final long MQ_CLUSTER_STATUS_CHECK_DUR_MS = 2000L;
+
private Context parentContext;
private MessageQueueZoneSinkContext context;
private List<MessageQueueZoneWorker> workers = new ArrayList<>();
// message group
private BatchPackManager dispatchManager;
- private BufferQueue<BatchPackProfile> dispatchQueue;
+ private BufferQueue<PackProfile> dispatchQueue;
// scheduled thread pool
// reload
// dispatch
@@ -67,6 +69,8 @@ public class MessageQueueZoneSink extends AbstractSink
implements Configurable,
// changeListerThread
private Thread configListener;
private volatile boolean isShutdown = false;
+ // whether mq cluster connected
+ private volatile boolean mqClusterStarted = false;
/**
* configure
*
@@ -104,7 +108,7 @@ public class MessageQueueZoneSink extends AbstractSink
implements Configurable,
}, this.dispatchManager.getDispatchTimeout(),
this.dispatchManager.getDispatchTimeout(),
TimeUnit.MILLISECONDS);
// create producer
- this.zoneProducer = new MessageQueueZoneProducer(this.getName(),
this.context);
+ this.zoneProducer = new MessageQueueZoneProducer(this,
this.context);
this.zoneProducer.start();
// start configure change listener thread
this.configListener = new Thread(new ConfigChangeProcessor());
@@ -158,6 +162,16 @@ public class MessageQueueZoneSink extends AbstractSink
implements Configurable,
*/
@Override
public Status process() throws EventDeliveryException {
+ // wait mq cluster started
+ while (!mqClusterStarted) {
+ try {
+ Thread.sleep(MQ_CLUSTER_STATUS_CHECK_DUR_MS);
+ } catch (InterruptedException e1) {
+ return Status.BACKOFF;
+ } catch (Throwable e2) {
+ //
+ }
+ }
this.dispatchManager.outputOvertimeData();
Channel channel = getChannel();
Transaction tx = channel.getTransaction();
@@ -215,6 +229,14 @@ public class MessageQueueZoneSink extends AbstractSink
implements Configurable,
syncLock.notifyAll();
}
+ public boolean isMqClusterStarted() {
+ return mqClusterStarted;
+ }
+
+ public void setMQClusterStarted() {
+ this.mqClusterStarted = true;
+ }
+
private class ConfigChangeProcessor implements Runnable {
@Override
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSinkContext.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSinkContext.java
index 19398c0438..151b812ff0 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSinkContext.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSinkContext.java
@@ -17,6 +17,7 @@
package org.apache.inlong.dataproxy.sink.mq;
+import org.apache.inlong.common.enums.DataProxyErrCode;
import org.apache.inlong.dataproxy.config.CommonConfigHolder;
import org.apache.inlong.dataproxy.consts.StatConstants;
import org.apache.inlong.dataproxy.metrics.DataProxyMetricItem;
@@ -44,7 +45,7 @@ public class MessageQueueZoneSinkContext extends SinkContext {
public static final String PREFIX_PRODUCER = "producer.";
public static final String KEY_COMPRESS_TYPE = "compressType";
- private final BufferQueue<BatchPackProfile> dispatchQueue;
+ private final BufferQueue<PackProfile> dispatchQueue;
private final String proxyClusterId;
private final String nodeId;
@@ -56,7 +57,7 @@ public class MessageQueueZoneSinkContext extends SinkContext {
* Constructor
*/
public MessageQueueZoneSinkContext(String sinkName, Context context,
Channel channel,
- BufferQueue<BatchPackProfile> dispatchQueue) {
+ BufferQueue<PackProfile> dispatchQueue) {
super(sinkName, context, channel);
this.dispatchQueue = dispatchQueue;
// proxyClusterId
@@ -99,7 +100,7 @@ public class MessageQueueZoneSinkContext extends SinkContext
{
*
* @return the dispatchQueue
*/
- public BufferQueue<BatchPackProfile> getDispatchQueue() {
+ public BufferQueue<PackProfile> getDispatchQueue() {
return dispatchQueue;
}
@@ -133,24 +134,24 @@ public class MessageQueueZoneSinkContext extends
SinkContext {
/**
* addSendResultMetric
*/
- public void addSendResultMetric(BatchPackProfile currentRecord, String
mqName, String topic, boolean result,
+ public void addSendResultMetric(PackProfile currentRecord, String mqName,
String topic, boolean result,
long sendTime) {
- if (currentRecord instanceof SimpleBatchPackProfileV0) {
+ if (currentRecord instanceof SimplePackProfile) {
AuditUtils.add(AuditUtils.AUDIT_ID_DATAPROXY_SEND_SUCCESS,
- ((SimpleBatchPackProfileV0)
currentRecord).getSimpleProfile());
+ ((SimplePackProfile) currentRecord).getEvent());
return;
}
-
+ BatchPackProfile batchProfile = (BatchPackProfile) currentRecord;
Map<String, String> dimensions = new HashMap<>();
dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getProxyClusterId());
dimensions.put(DataProxyMetricItem.KEY_SOURCE_ID, "-");
dimensions.put(DataProxyMetricItem.KEY_SOURCE_DATA_ID, "-");
// metric
- fillInlongId(currentRecord, dimensions);
+ fillInlongId(batchProfile, dimensions);
dimensions.put(DataProxyMetricItem.KEY_SINK_ID, mqName);
dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, topic);
final long currentTime = System.currentTimeMillis();
- currentRecord.getEvents().forEach(event -> {
+ batchProfile.getEvents().forEach(event -> {
long msgTime = event.getMsgTime();
long auditFormatTime =
msgTime - msgTime %
CommonConfigHolder.getInstance().getAuditFormatInvlMs();
@@ -178,7 +179,7 @@ public class MessageQueueZoneSinkContext extends
SinkContext {
/**
* addSendMetric
*/
- public void addSendMetric(BatchPackProfile currentRecord, String mqName,
String topic, int sendPackSize) {
+ public void addSendMetric(PackProfile currentRecord, String mqName, String
topic, int sendPackSize) {
Map<String, String> dimensions = new HashMap<>();
dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID,
this.getProxyClusterId());
dimensions.put(DataProxyMetricItem.KEY_SOURCE_ID, "-");
@@ -224,7 +225,7 @@ public class MessageQueueZoneSinkContext extends
SinkContext {
/**
* fillInlongId
*/
- public static void fillInlongId(BatchPackProfile currentRecord,
Map<String, String> dimensions) {
+ public static void fillInlongId(PackProfile currentRecord, Map<String,
String> dimensions) {
String inlongGroupId = currentRecord.getInlongGroupId();
inlongGroupId = (StringUtils.isBlank(inlongGroupId)) ? "-" :
inlongGroupId;
String inlongStreamId = currentRecord.getInlongStreamId();
@@ -236,13 +237,15 @@ public class MessageQueueZoneSinkContext extends
SinkContext {
/**
* processSendFail
*/
- public void processSendFail(BatchPackProfile currentRecord, String mqName,
String topic, long sendTime) {
+ public void processSendFail(PackProfile currentRecord,
+ String mqName, String topic, long sendTime,
+ DataProxyErrCode errCode, String errMsg) {
if (currentRecord.isResend()) {
dispatchQueue.offer(currentRecord);
fileMetricEventInc(StatConstants.EVENT_SINK_FAILRETRY);
this.addSendResultMetric(currentRecord, mqName, topic, false,
sendTime);
} else {
- currentRecord.fail();
+ currentRecord.fail(errCode, errMsg);
fileMetricEventInc(StatConstants.EVENT_SINK_FAILDROPPED);
}
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneWorker.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneWorker.java
index d858816f35..8a2e4c5abe 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneWorker.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneWorker.java
@@ -72,19 +72,19 @@ public class MessageQueueZoneWorker extends Thread {
public void run() {
LOG.info(String.format("start MessageQueueZoneWorker:%s",
this.workerName));
while (status != LifecycleState.STOP) {
- BatchPackProfile event = null;
+ PackProfile profile = null;
try {
- event = context.getDispatchQueue().pollRecord();
- if (event == null) {
+ profile = context.getDispatchQueue().pollRecord();
+ if (profile == null) {
this.sleepOneInterval();
continue;
}
// send
- this.zoneProducer.send(event);
+ this.zoneProducer.send(profile);
} catch (Throwable e) {
LOG.error(e.getMessage(), e);
- if (event != null) {
- context.getDispatchQueue().offer(event);
+ if (profile != null) {
+ context.getDispatchQueue().offer(profile);
}
this.sleepOneInterval();
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
deleted file mode 100644
index 172341b411..0000000000
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.dataproxy.sink.mq;
-
-import org.apache.inlong.common.msg.AttributeConstants;
-import org.apache.inlong.common.msg.MsgType;
-import org.apache.inlong.dataproxy.base.SinkRspEvent;
-import org.apache.inlong.sdk.commons.protocol.InlongId;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.nio.charset.StandardCharsets;
-import java.util.Map;
-
-/**
- * SimpleBatchPackProfileV0
- *
- */
-public class OrderBatchPackProfileV0 extends BatchPackProfile {
-
- public static final Logger LOG =
LoggerFactory.getLogger(OrderBatchPackProfileV0.class);
-
- private SinkRspEvent orderProfile;
-
- /**
- * Constructor
- * @param uid
- * @param inlongGroupId
- * @param inlongStreamId
- * @param dispatchTime
- */
- public OrderBatchPackProfileV0(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
- super(uid, inlongGroupId, inlongStreamId, dispatchTime);
- }
-
- /**
- * create
- * @param event
- * @return
- */
- public static OrderBatchPackProfileV0 create(SinkRspEvent event) {
- Map<String, String> headers = event.getHeaders();
- String inlongGroupId = headers.get(AttributeConstants.GROUP_ID);;
- String inlongStreamId = headers.get(AttributeConstants.STREAM_ID);
- String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
- long msgTime =
NumberUtils.toLong(headers.get(AttributeConstants.DATA_TIME),
System.currentTimeMillis());
- long dispatchTime = msgTime - msgTime % MINUTE_MS;
- OrderBatchPackProfileV0 profile = new OrderBatchPackProfileV0(uid,
inlongGroupId, inlongStreamId,
- dispatchTime);
- profile.setCount(1);
- profile.setSize(event.getBody().length);
- profile.orderProfile = event;
- return profile;
- }
-
- /**
- * get event
- * @return the event
- */
- public SinkRspEvent getOrderProfile() {
- return orderProfile;
- }
-
- /**
- * ackOrder
- */
- public void ackOrder() {
- String sequenceId =
orderProfile.getHeaders().get(AttributeConstants.UNIQ_ID);
- if
("false".equals(orderProfile.getHeaders().get(AttributeConstants.MESSAGE_IS_ACK)))
{
- if (LOG.isDebugEnabled()) {
- LOG.debug("not need to rsp message: seqId = {}, inlongGroupId
= {}, inlongStreamId = {}",
- sequenceId, this.getInlongGroupId(),
this.getInlongStreamId());
- }
- return;
- }
- if (orderProfile.getChannel() != null &&
orderProfile.getChannel().isActive()) {
- orderProfile.getChannel().eventLoop().execute(() -> {
- if (LOG.isDebugEnabled()) {
- LOG.debug("order message rsp: seqId = {}, inlongGroupId =
{}, inlongStreamId = {}", sequenceId,
- this.getInlongGroupId(), this.getInlongStreamId());
- }
- ByteBuf binBuffer = getResponsePackage("",
MsgType.MSG_BIN_MULTI_BODY, sequenceId);
- orderProfile.getChannel().writeAndFlush(binBuffer);
- });
- }
- }
-
- /**
- * Convert String to ByteBuf
- *
- * @param backattrs
- * @param msgType message type
- * @param sequenceId sequence Id
- * @return ByteBuf
- */
- public static ByteBuf getResponsePackage(String backattrs, MsgType
msgType, String sequenceId) {
- int binTotalLen = 1 + 4 + 2 + 2;
- if (null != backattrs) {
- binTotalLen += backattrs.length();
- }
- ByteBuf binBuffer = ByteBufAllocator.DEFAULT.buffer(4 + binTotalLen);
- binBuffer.writeInt(binTotalLen);
- binBuffer.writeByte(msgType.getValue());
-
- long uniqVal = Long.parseLong(sequenceId);
- byte[] uniq = new byte[4];
- uniq[0] = (byte) ((uniqVal >> 24) & 0xFF);
- uniq[1] = (byte) ((uniqVal >> 16) & 0xFF);
- uniq[2] = (byte) ((uniqVal >> 8) & 0xFF);
- uniq[3] = (byte) (uniqVal & 0xFF);
- binBuffer.writeBytes(uniq);
-
- if (null != backattrs) {
- binBuffer.writeShort(backattrs.length());
- binBuffer.writeBytes(backattrs.getBytes(StandardCharsets.UTF_8));
- } else {
- binBuffer.writeShort(0x0);
- }
- binBuffer.writeShort(0xee01);
- return binBuffer;
- }
-}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/PackProfile.java
similarity index 59%
copy from
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
copy to
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/PackProfile.java
index 9abe78f4a0..9541091347 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/PackProfile.java
@@ -17,38 +17,33 @@
package org.apache.inlong.dataproxy.sink.mq;
-import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.inlong.common.enums.DataProxyErrCode;
-import java.util.ArrayList;
-import java.util.List;
+import org.apache.flume.Event;
/**
*
* DispatchProfile
*/
-public class BatchPackProfile {
-
- public static final long MINUTE_MS = 60L * 1000;
+public abstract class PackProfile {
private final String inlongGroupId;
private final String inlongStreamId;
+ private final long dispatchTime;
+ private final long createTime = System.currentTimeMillis();
private final String uid;
- private List<ProxyEvent> events = new ArrayList<>();
- private long createTime = System.currentTimeMillis();
- private long count = 0;
- private long size = 0;
- private long dispatchTime;
- private BatchPackProfileCallback callback;
+ protected long count = 0;
+ protected long size = 0;
/**
* Constructor
- *
+ *
* @param uid
* @param inlongGroupId
* @param inlongStreamId
* @param dispatchTime
*/
- public BatchPackProfile(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
+ public PackProfile(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
this.uid = uid;
this.inlongGroupId = inlongGroupId;
this.inlongStreamId = inlongStreamId;
@@ -56,64 +51,44 @@ public class BatchPackProfile {
}
/**
- * addEvent
- *
- * @param event
- * @param maxPackCount
- * @param maxPackSize
- * @return
- */
- public boolean addEvent(ProxyEvent event, long maxPackCount, long
maxPackSize) {
- long eventLength = event.getBody().length;
- if (count >= maxPackCount || (count > 0 && size + eventLength >
maxPackSize)) {
- return false;
- }
- this.events.add(event);
- this.count++;
- this.size += eventLength;
- return true;
- }
-
- /**
- * isTimeout
+ * get uid
*
- * @param createThreshold
- * @return
+ * @return the uid
*/
- public boolean isTimeout(long createThreshold) {
- return createThreshold >= createTime;
+ public String getUid() {
+ return uid;
}
/**
- * get uid
+ * get inlongGroupId
*
- * @return the uid
+ * @return the inlongGroupId
*/
- public String getUid() {
- return uid;
+ public String getInlongGroupId() {
+ return inlongGroupId;
}
/**
- * get events
+ * get inlongStreamId
*
- * @return the events
+ * @return the inlongStreamId
*/
- public List<ProxyEvent> getEvents() {
- return events;
+ public String getInlongStreamId() {
+ return inlongStreamId;
}
/**
- * set events
+ * getDispatchTime
*
- * @param events the events to set
+ * @return
*/
- public void setEvents(List<ProxyEvent> events) {
- this.events = events;
+ public long getDispatchTime() {
+ return dispatchTime;
}
/**
* get count
- *
+ *
* @return the count
*/
public long getCount() {
@@ -122,7 +97,7 @@ public class BatchPackProfile {
/**
* set count
- *
+ *
* @param count the count to set
*/
public void setCount(long count) {
@@ -131,7 +106,7 @@ public class BatchPackProfile {
/**
* get size
- *
+ *
* @return the size
*/
public long getSize() {
@@ -140,7 +115,7 @@ public class BatchPackProfile {
/**
* set size
- *
+ *
* @param size the size to set
*/
public void setSize(long size) {
@@ -148,73 +123,40 @@ public class BatchPackProfile {
}
/**
- * get inlongGroupId
- *
- * @return the inlongGroupId
- */
- public String getInlongGroupId() {
- return inlongGroupId;
- }
-
- /**
- * get inlongStreamId
- *
- * @return the inlongStreamId
- */
- public String getInlongStreamId() {
- return inlongStreamId;
- }
-
- /**
- * getDispatchTime
- *
+ * isTimeout
+ *
+ * @param createThreshold
* @return
*/
- public long getDispatchTime() {
- return dispatchTime;
+ public boolean isTimeout(long createThreshold) {
+ return createThreshold >= createTime;
}
/**
* ack
*/
- public void ack() {
- if (callback != null) {
- callback.ack(this.events.size());
- }
- }
+ public abstract void ack();
/**
* fail
* @return
*/
- public void fail() {
- if (callback != null) {
- callback.fail();
- }
- }
+ public abstract void fail(DataProxyErrCode errCode, String errMsg);
/**
* isResend
* @return
*/
- public boolean isResend() {
- return callback == null;
- }
-
- /**
- * get callback
- * @return the callback
- */
- public BatchPackProfileCallback getCallback() {
- return callback;
- }
+ public abstract boolean isResend();
/**
- * set callback
- * @param callback the callback to set
+ * addEvent
+ *
+ * @param event
+ * @param maxPackCount
+ * @param maxPackSize
+ * @return
*/
- public void setCallback(BatchPackProfileCallback callback) {
- this.callback = callback;
- }
+ public abstract boolean addEvent(Event event, long maxPackCount, long
maxPackSize);
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimpleBatchPackProfileV0.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimpleBatchPackProfileV0.java
deleted file mode 100644
index dca12907e6..0000000000
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimpleBatchPackProfileV0.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.dataproxy.sink.mq;
-
-import org.apache.inlong.common.msg.AttributeConstants;
-import org.apache.inlong.dataproxy.consts.ConfigConstants;
-import org.apache.inlong.dataproxy.utils.MessageUtils;
-import org.apache.inlong.sdk.commons.protocol.InlongId;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.math.NumberUtils;
-import org.apache.flume.Event;
-
-import java.util.Map;
-
-/**
- * SimpleBatchPackProfileV0
- *
- */
-public class SimpleBatchPackProfileV0 extends BatchPackProfile {
-
- private Event simpleProfile;
- private Map<String, String> properties;
-
- /**
- * Constructor
- * @param uid
- * @param inlongGroupId
- * @param inlongStreamId
- * @param dispatchTime
- */
- public SimpleBatchPackProfileV0(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
- super(uid, inlongGroupId, inlongStreamId, dispatchTime);
- }
-
- /**
- * create
- * @param event
- * @return
- */
- public static SimpleBatchPackProfileV0 create(Event event) {
- Map<String, String> headers = event.getHeaders();
- String inlongGroupId = headers.get(AttributeConstants.GROUP_ID);
- String inlongStreamId = headers.get(AttributeConstants.STREAM_ID);
- String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
- long msgTime =
NumberUtils.toLong(headers.get(AttributeConstants.DATA_TIME),
System.currentTimeMillis());
- long dispatchTime = msgTime - msgTime % MINUTE_MS;
- SimpleBatchPackProfileV0 profile = new SimpleBatchPackProfileV0(uid,
inlongGroupId, inlongStreamId,
- dispatchTime);
- profile.setCount(1);
- profile.setSize(event.getBody().length);
- profile.simpleProfile = event;
-
- String pkgVersion =
event.getHeaders().get(ConfigConstants.MSG_ENCODE_VER);
- if (StringUtils.isNotBlank(pkgVersion)) {
- profile.properties = MessageUtils.getXfsAttrs(headers, pkgVersion);
- }
- return profile;
- }
-
- /**
- * get simpleProfile
- * @return the simpleProfile
- */
- public Event getSimpleProfile() {
- return simpleProfile;
- }
-
- /**
- * get properties
- * @return the properties
- */
- public Map<String, String> getProperties() {
- return properties;
- }
-}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimplePackProfile.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimplePackProfile.java
new file mode 100644
index 0000000000..e701509822
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimplePackProfile.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.dataproxy.sink.mq;
+
+import org.apache.inlong.common.enums.DataProxyErrCode;
+import org.apache.inlong.common.monitor.LogCounter;
+import org.apache.inlong.common.msg.AttributeConstants;
+import org.apache.inlong.common.msg.MsgType;
+import org.apache.inlong.dataproxy.base.SinkRspEvent;
+import org.apache.inlong.dataproxy.source2.InLongMessageHandler;
+import org.apache.inlong.sdk.commons.protocol.InlongId;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.flume.Event;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+/**
+ * Simple Pack Profile
+ *
+ */
+public class SimplePackProfile extends PackProfile {
+
+ // log print count
+ private static final LogCounter logCounter =
+ new LogCounter(10, 100000, 30 * 1000);
+ private static final Logger logger =
LoggerFactory.getLogger(SimplePackProfile.class);
+ private static final long MINUTE_MS = 60L * 1000;
+ private boolean needRspEvent = false;
+ private Channel channel;
+ private MsgType msgType;
+ private Event event;
+
+ /**
+ * Constructor
+ * @param uid
+ * @param inlongGroupId
+ * @param inlongStreamId
+ * @param dispatchTime
+ */
+ public SimplePackProfile(String uid, String inlongGroupId, String
inlongStreamId, long dispatchTime) {
+ super(uid, inlongGroupId, inlongStreamId, dispatchTime);
+ }
+
+ @Override
+ public void ack() {
+ if (!this.needRspEvent) {
+ return;
+ }
+ responseV0Msg(DataProxyErrCode.SUCCESS, "");
+ }
+
+ @Override
+ public void fail(DataProxyErrCode errCode, String errMsg) {
+ if (!needRspEvent) {
+ return;
+ }
+ responseV0Msg(errCode, errMsg);
+ }
+
+ @Override
+ public boolean isResend() {
+ return !needRspEvent;
+ }
+
+ @Override
+ public boolean addEvent(Event event, long maxPackCount, long maxPackSize) {
+ setCount(1);
+ setSize(event.getBody().length);
+ if (event instanceof SinkRspEvent) {
+ SinkRspEvent rspEvent = (SinkRspEvent) event;
+ this.needRspEvent = true;
+ this.event = rspEvent.getEvent();
+ this.channel = rspEvent.getChannel();
+ this.msgType = rspEvent.getMsgType();
+ } else {
+ this.event = event;
+ this.needRspEvent = false;
+ }
+ return true;
+ }
+
+ /**
+ * create
+ * @param event
+ * @return
+ */
+ public static SimplePackProfile create(Event event) {
+ Map<String, String> headers = event.getHeaders();
+ String inlongGroupId = headers.get(AttributeConstants.GROUP_ID);
+ String inlongStreamId = headers.get(AttributeConstants.STREAM_ID);
+ String uid = InlongId.generateUid(inlongGroupId, inlongStreamId);
+ long msgTime =
NumberUtils.toLong(headers.get(AttributeConstants.DATA_TIME),
System.currentTimeMillis());
+ long dispatchTime = msgTime - msgTime % MINUTE_MS;
+ SimplePackProfile profile = new SimplePackProfile(uid, inlongGroupId,
inlongStreamId,
+ dispatchTime);
+ profile.setCount(1);
+ profile.setSize(event.getBody().length);
+ if (event instanceof SinkRspEvent) {
+ SinkRspEvent rspEvent = (SinkRspEvent) event;
+ profile.needRspEvent = true;
+ profile.event = rspEvent.getEvent();
+ profile.channel = rspEvent.getChannel();
+ profile.msgType = rspEvent.getMsgType();
+ } else {
+ profile.event = event;
+ }
+ return profile;
+ }
+
+ public Event getEvent() {
+ return event;
+ }
+
+ /**
+ * get properties
+ * @return the properties
+ */
+ public Map<String, String> getProperties() {
+ return event.getHeaders();
+ }
+
+ /**
+ * Return response to client in source
+ */
+ private void responseV0Msg(DataProxyErrCode errCode, String errMsg) {
+ try {
+ String uid = event.getHeaders().get(AttributeConstants.UNIQ_ID);
+ if
("false".equals(event.getHeaders().get(AttributeConstants.MESSAGE_IS_ACK))) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("not need to rsp message: seqId = {},
inlongGroupId = {}, inlongStreamId = {}",
+ uid, this.getInlongGroupId(),
this.getInlongStreamId());
+ }
+ return;
+ }
+ // check channel status
+ if (channel == null || !channel.isWritable()) {
+ if (logCounter.shouldPrint()) {
+ logger.warn("Prepare send msg but channel full,
msgType={}, attr={}, channel={}",
+ msgType, event.getHeaders(), channel);
+ }
+ return;
+ }
+ // build return attribute string
+ StringBuilder strBuff = new StringBuilder(512);
+ if (errCode != DataProxyErrCode.SUCCESS) {
+ strBuff.append(AttributeConstants.MESSAGE_PROCESS_ERRCODE)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errCode.getErrCodeStr());
+ if (StringUtils.isNotEmpty(errMsg)) {
+
strBuff.append(AttributeConstants.SEPARATOR).append(AttributeConstants.MESSAGE_PROCESS_ERRMSG)
+
.append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errMsg);
+ }
+ }
+ // build and send response message
+ ByteBuf retData;
+ if (MsgType.MSG_BIN_MULTI_BODY.equals(msgType)) {
+ retData =
InLongMessageHandler.buildBinMsgRspPackage(strBuff.toString(),
Long.parseLong(uid));
+ } else {
+ retData = InLongMessageHandler.buildTxtMsgRspPackage(msgType,
strBuff.toString());
+ }
+ strBuff.delete(0, strBuff.length());
+ if (channel == null || !channel.isWritable()) {
+ // release allocated ByteBuf
+ retData.release();
+ if (logCounter.shouldPrint()) {
+ logger.warn("Send msg but channel full, attr={},
channel={}", event.getHeaders(), channel);
+ }
+ return;
+ }
+ channel.writeAndFlush(strBuff);
+ } catch (Throwable e) {
+ //
+ if (logCounter.shouldPrint()) {
+ logger.warn("Send msg but failure, attr={}",
event.getHeaders(), e);
+ }
+ }
+ }
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java
index 5e28a4d407..b8eef4f7ca 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java
@@ -17,6 +17,7 @@
package org.apache.inlong.dataproxy.sink.mq.kafka;
+import org.apache.inlong.common.enums.DataProxyErrCode;
import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
@@ -25,10 +26,9 @@ import org.apache.inlong.dataproxy.sink.common.EventHandler;
import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueHandler;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueZoneSinkContext;
-import org.apache.inlong.dataproxy.sink.mq.OrderBatchPackProfileV0;
-import org.apache.inlong.dataproxy.sink.mq.SimpleBatchPackProfileV0;
+import org.apache.inlong.dataproxy.sink.mq.PackProfile;
+import org.apache.inlong.dataproxy.sink.mq.SimplePackProfile;
-import org.apache.commons.collections.MapUtils;
import org.apache.flume.Context;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
@@ -109,57 +109,56 @@ public class KafkaHandler implements MessageQueueHandler {
/**
* send
- * @param event
+ * @param profile
* @return
*/
@Override
- public boolean send(BatchPackProfile event) {
+ public boolean send(PackProfile profile) {
try {
// idConfig
IdTopicConfig idConfig =
ConfigManager.getInstance().getIdTopicConfig(
- event.getInlongGroupId(), event.getInlongStreamId());
+ profile.getInlongGroupId(), profile.getInlongStreamId());
if (idConfig == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOUID);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+
profile.fail(DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE, "");
return false;
}
String topic = idConfig.getTopicName();
if (topic == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOTOPIC);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+ profile.fail(DataProxyErrCode.TOPIC_IS_BLANK, "");
return false;
}
// create producer failed
if (producer == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOPRODUCER);
- sinkContext.processSendFail(event, clusterName, topic, 0);
+ sinkContext.processSendFail(profile, clusterName, topic, 0,
DataProxyErrCode.PRODUCER_IS_NULL, "");
return false;
}
// send
- if (event instanceof SimpleBatchPackProfileV0) {
- this.sendSimpleProfileV0((SimpleBatchPackProfileV0) event,
idConfig, topic);
- } else if (event instanceof OrderBatchPackProfileV0) {
- this.sendOrderProfileV0((OrderBatchPackProfileV0) event,
idConfig, topic);
+ if (profile instanceof SimplePackProfile) {
+ this.sendSimplePackProfile((SimplePackProfile) profile,
idConfig, topic);
} else {
- this.sendProfileV1(event, idConfig, topic);
+ this.sendBatchPackProfile((BatchPackProfile) profile,
idConfig, topic);
}
return true;
- } catch (Exception e) {
+ } catch (Exception ex) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SENDEXCEPT);
- sinkContext.processSendFail(event, clusterName, event.getUid(), 0);
- LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(profile, clusterName,
profile.getUid(), 0,
+ DataProxyErrCode.SEND_REQUEST_TO_MQ_FAILURE,
ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
return false;
}
}
/**
- * sendProfileV1
+ * send BatchPackProfile
*/
- private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
+ private void sendBatchPackProfile(BatchPackProfile batchProfile,
IdTopicConfig idConfig,
String topic) throws Exception {
EventHandler handler = handlerLocal.get();
if (handler == null) {
@@ -167,12 +166,12 @@ public class KafkaHandler implements MessageQueueHandler {
handlerLocal.set(handler);
}
// headers
- Map<String, String> headers = handler.parseHeader(idConfig, event,
sinkContext.getNodeId(),
+ Map<String, String> headers = handler.parseHeader(idConfig,
batchProfile, sinkContext.getNodeId(),
sinkContext.getCompressType());
// compress
- byte[] bodyBytes = handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ byte[] bodyBytes = handler.parseBody(idConfig, batchProfile,
sinkContext.getCompressType());
// metric
- sinkContext.addSendMetric(event, clusterName, topic, bodyBytes.length);
+ sinkContext.addSendMetric(batchProfile, clusterName, topic,
bodyBytes.length);
// sendAsync
long sendTime = System.currentTimeMillis();
@@ -190,13 +189,14 @@ public class KafkaHandler implements MessageQueueHandler {
public void onCompletion(RecordMetadata arg0, Exception ex) {
if (ex != null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
- LOG.error("Send ProfileV1 to Kafka failure", ex);
+ sinkContext.processSendFail(batchProfile, clusterName,
topic, sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
+ LOG.error("Send BatchPackProfile to Kafka failure", ex);
} else {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
+ sinkContext.addSendResultMetric(batchProfile, clusterName,
topic, true, sendTime);
+
sinkContext.getDispatchQueue().release(batchProfile.getSize());
+ batchProfile.ack();
}
}
};
@@ -204,60 +204,16 @@ public class KafkaHandler implements MessageQueueHandler {
}
/**
- * sendSimpleProfileV0
+ * send SimplePackProfile
*/
- private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ private void sendSimplePackProfile(SimplePackProfile simpleProfile,
IdTopicConfig idConfig,
String topic) throws Exception {
// headers
- Map<String, String> headers = event.getProperties();
- if (MapUtils.isEmpty(headers)) {
- headers = event.getSimpleProfile().getHeaders();
- }
+ Map<String, String> headers = simpleProfile.getProperties();
// body
- byte[] bodyBytes = event.getSimpleProfile().getBody();
- // metric
- sinkContext.addSendMetric(event, clusterName, topic, bodyBytes.length);
- // sendAsync
- long sendTime = System.currentTimeMillis();
-
- // prepare ProducerRecord
- ProducerRecord<String, byte[]> producerRecord = new
ProducerRecord<>(topic, bodyBytes);
- // add headers
- headers.forEach((key, value) -> {
- producerRecord.headers().add(key, value.getBytes());
- });
-
- // callback
- Callback callback = new Callback() {
-
- @Override
- public void onCompletion(RecordMetadata arg0, Exception ex) {
- if (ex != null) {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
- LOG.error("Send SimpleProfileV0 to Kafka failure", ex);
- } else {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
- }
- }
- };
- producer.send(producerRecord, callback);
- }
-
- /**
- * sendOrderProfileV0
- */
- private void sendOrderProfileV0(OrderBatchPackProfileV0 event,
IdTopicConfig idConfig,
- String topic) throws Exception {
- // headers
- Map<String, String> headers = event.getOrderProfile().getHeaders();
- // compress
- byte[] bodyBytes = event.getOrderProfile().getBody();
+ byte[] bodyBytes = simpleProfile.getEvent().getBody();
// metric
- sinkContext.addSendMetric(event, clusterName, topic, bodyBytes.length);
+ sinkContext.addSendMetric(simpleProfile, clusterName, topic,
bodyBytes.length);
// sendAsync
long sendTime = System.currentTimeMillis();
@@ -275,13 +231,14 @@ public class KafkaHandler implements MessageQueueHandler {
public void onCompletion(RecordMetadata arg0, Exception ex) {
if (ex != null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
- LOG.error("Send OrderProfileV0 to Kafka failure", ex);
+ sinkContext.processSendFail(simpleProfile, clusterName,
topic, sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
+ LOG.error("Send SimplePackProfile to Kafka failure", ex);
} else {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
+ sinkContext.addSendResultMetric(simpleProfile,
clusterName, topic, true, sendTime);
+
sinkContext.getDispatchQueue().release(simpleProfile.getSize());
+ simpleProfile.ack();
}
}
};
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java
index 1800c5550a..0765197431 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java
@@ -17,6 +17,7 @@
package org.apache.inlong.dataproxy.sink.mq.pulsar;
+import org.apache.inlong.common.enums.DataProxyErrCode;
import org.apache.inlong.common.monitor.LogCounter;
import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
@@ -26,10 +27,9 @@ import org.apache.inlong.dataproxy.sink.common.EventHandler;
import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueHandler;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueZoneSinkContext;
-import org.apache.inlong.dataproxy.sink.mq.OrderBatchPackProfileV0;
-import org.apache.inlong.dataproxy.sink.mq.SimpleBatchPackProfileV0;
+import org.apache.inlong.dataproxy.sink.mq.PackProfile;
+import org.apache.inlong.dataproxy.sink.mq.SimplePackProfile;
-import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
@@ -197,29 +197,29 @@ public class PulsarHandler implements MessageQueueHandler
{
/**
* send
- * @param event
+ * @param profile
* @return
*/
@Override
- public boolean send(BatchPackProfile event) {
+ public boolean send(PackProfile profile) {
try {
// idConfig
IdTopicConfig idConfig =
ConfigManager.getInstance().getIdTopicConfig(
- event.getInlongGroupId(), event.getInlongStreamId());
+ profile.getInlongGroupId(), profile.getInlongStreamId());
if (idConfig == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOUID);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+
profile.fail(DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE, "");
return false;
}
// topic
String producerTopic = idConfig.getPulsarTopicName(tenant,
namespace);
if (producerTopic == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOTOPIC);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+ profile.fail(DataProxyErrCode.TOPIC_IS_BLANK, "");
return false;
}
// get producer
@@ -247,22 +247,22 @@ public class PulsarHandler implements MessageQueueHandler
{
// create producer failed
if (producer == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOPRODUCER);
- sinkContext.processSendFail(event, clusterName, producerTopic,
0);
+ sinkContext.processSendFail(profile, clusterName,
producerTopic, 0,
+ DataProxyErrCode.PRODUCER_IS_NULL, "");
return false;
}
// send
- if (event instanceof SimpleBatchPackProfileV0) {
- this.sendSimpleProfileV0((SimpleBatchPackProfileV0) event,
idConfig, producer, producerTopic);
- } else if (event instanceof OrderBatchPackProfileV0) {
- this.sendOrderProfileV0((OrderBatchPackProfileV0) event,
idConfig, producer, producerTopic);
+ if (profile instanceof SimplePackProfile) {
+ this.sendSimplePackProfile((SimplePackProfile) profile,
idConfig, producer, producerTopic);
} else {
- this.sendProfileV1(event, idConfig, producer, producerTopic);
+ this.sendBatchPackProfile((BatchPackProfile) profile,
idConfig, producer, producerTopic);
}
return true;
- } catch (Exception e) {
+ } catch (Exception ex) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SENDEXCEPT);
- sinkContext.processSendFail(event, clusterName, event.getUid(), 0);
- LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(profile, clusterName,
profile.getUid(), 0,
+ DataProxyErrCode.SEND_REQUEST_TO_MQ_FAILURE,
ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
return false;
}
}
@@ -291,9 +291,9 @@ public class PulsarHandler implements MessageQueueHandler {
}
/**
- * sendProfileV1
+ * send BatchPackProfile
*/
- private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
Producer<byte[]> producer,
+ private void sendBatchPackProfile(BatchPackProfile batchProfile,
IdTopicConfig idConfig, Producer<byte[]> producer,
String producerTopic) throws Exception {
EventHandler handler = handlerLocal.get();
if (handler == null) {
@@ -301,12 +301,12 @@ public class PulsarHandler implements MessageQueueHandler
{
handlerLocal.set(handler);
}
// headers
- Map<String, String> headers = handler.parseHeader(idConfig, event,
sinkContext.getNodeId(),
+ Map<String, String> headers = handler.parseHeader(idConfig,
batchProfile, sinkContext.getNodeId(),
sinkContext.getCompressType());
// compress
- byte[] bodyBytes = handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ byte[] bodyBytes = handler.parseBody(idConfig, batchProfile,
sinkContext.getCompressType());
// metric
- sinkContext.addSendMetric(event, clusterName, producerTopic,
bodyBytes.length);
+ sinkContext.addSendMetric(batchProfile, clusterName, producerTopic,
bodyBytes.length);
// sendAsync
long sendTime = System.currentTimeMillis();
CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
@@ -315,32 +315,30 @@ public class PulsarHandler implements MessageQueueHandler
{
future.whenCompleteAsync((msgId, ex) -> {
if (ex != null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, producerTopic,
sendTime);
+ sinkContext.processSendFail(batchProfile, clusterName,
producerTopic, sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
LOG.error("Send ProfileV1 to Pulsar failure", ex);
} else {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName,
producerTopic, true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
+ sinkContext.addSendResultMetric(batchProfile, clusterName,
producerTopic, true, sendTime);
+ sinkContext.getDispatchQueue().release(batchProfile.getSize());
+ batchProfile.ack();
}
});
}
/**
- * sendSimpleProfileV0
+ * send SimplePackProfile
*/
- private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ private void sendSimplePackProfile(SimplePackProfile simpleProfile,
IdTopicConfig idConfig,
Producer<byte[]> producer,
String producerTopic) throws Exception {
// headers
- Map<String, String> headers = event.getProperties();
- if (MapUtils.isEmpty(headers)) {
- headers = event.getSimpleProfile().getHeaders();
- }
+ Map<String, String> headers = simpleProfile.getProperties();
// body
- byte[] bodyBytes = event.getSimpleProfile().getBody();
+ byte[] bodyBytes = simpleProfile.getEvent().getBody();
// metric
- sinkContext.addSendMetric(event, clusterName, producerTopic,
bodyBytes.length);
+ sinkContext.addSendMetric(simpleProfile, clusterName, producerTopic,
bodyBytes.length);
// sendAsync
long sendTime = System.currentTimeMillis();
CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
@@ -349,44 +347,14 @@ public class PulsarHandler implements MessageQueueHandler
{
future.whenCompleteAsync((msgId, ex) -> {
if (ex != null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, producerTopic,
sendTime);
+ sinkContext.processSendFail(simpleProfile, clusterName,
producerTopic, sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
LOG.error("Send SimpleProfileV0 to Pulsar failure", ex);
} else {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName,
producerTopic, true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
- }
- });
- }
-
- /**
- * sendOrderProfileV0
- */
- private void sendOrderProfileV0(OrderBatchPackProfileV0 event,
IdTopicConfig idConfig, Producer<byte[]> producer,
- String producerTopic) throws Exception {
- // headers
- Map<String, String> headers = event.getOrderProfile().getHeaders();
- // compress
- byte[] bodyBytes = event.getOrderProfile().getBody();
- // metric
- sinkContext.addSendMetric(event, clusterName, producerTopic,
bodyBytes.length);
- // sendAsync
- long sendTime = System.currentTimeMillis();
- CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
- .value(bodyBytes).sendAsync();
- // callback
- future.whenCompleteAsync((msgId, ex) -> {
- if (ex != null) {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, producerTopic,
sendTime);
- LOG.error("Send OrderProfileV0 to Pulsar failure", ex);
- } else {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName,
producerTopic, true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
- event.ackOrder();
+ sinkContext.addSendResultMetric(simpleProfile, clusterName,
producerTopic, true, sendTime);
+
sinkContext.getDispatchQueue().release(simpleProfile.getSize());
+ simpleProfile.ack();
}
});
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java
index bd02c7a566..54da8c283b 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java
@@ -17,6 +17,7 @@
package org.apache.inlong.dataproxy.sink.mq.tube;
+import org.apache.inlong.common.enums.DataProxyErrCode;
import org.apache.inlong.common.monitor.LogCounter;
import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
@@ -28,8 +29,8 @@ import org.apache.inlong.dataproxy.sink.common.TubeUtils;
import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueHandler;
import org.apache.inlong.dataproxy.sink.mq.MessageQueueZoneSinkContext;
-import org.apache.inlong.dataproxy.sink.mq.OrderBatchPackProfileV0;
-import org.apache.inlong.dataproxy.sink.mq.SimpleBatchPackProfileV0;
+import org.apache.inlong.dataproxy.sink.mq.PackProfile;
+import org.apache.inlong.dataproxy.sink.mq.SimplePackProfile;
import org.apache.inlong.tubemq.client.config.TubeClientConfig;
import org.apache.inlong.tubemq.client.exception.TubeClientException;
import org.apache.inlong.tubemq.client.factory.TubeMultiSessionFactory;
@@ -71,8 +72,8 @@ public class TubeHandler implements MessageQueueHandler {
// tube producer
private TubeMultiSessionFactory sessionFactory;
private MessageProducer producer;
- private Set<String> topicSet = new HashSet<>();
- private ThreadLocal<EventHandler> handlerLocal = new ThreadLocal<>();
+ private final Set<String> topicSet = new HashSet<>();
+ private final ThreadLocal<EventHandler> handlerLocal = new ThreadLocal<>();
/**
* init
@@ -112,6 +113,7 @@ public class TubeHandler implements MessageQueueHandler {
Set<String> published;
try {
published = producer.publish(topicSet);
+ topicSet.addAll(published);
LOG.info("Publish topics to {}, need publish are {}, published are
{}",
this.clusterName, topicSet, published);
} catch (Throwable e) {
@@ -178,30 +180,31 @@ public class TubeHandler implements MessageQueueHandler {
/**
* send
*/
- public boolean send(BatchPackProfile event) {
+ public boolean send(PackProfile profile) {
try {
// idConfig
IdTopicConfig idConfig =
ConfigManager.getInstance().getIdTopicConfig(
- event.getInlongGroupId(), event.getInlongStreamId());
+ profile.getInlongGroupId(), profile.getInlongStreamId());
if (idConfig == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOUID);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+
profile.fail(DataProxyErrCode.GROUPID_OR_STREAMID_NOT_CONFIGURE, "");
return false;
}
String topic = idConfig.getTopicName();
if (topic == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOTOPIC);
- sinkContext.addSendResultMetric(event, clusterName,
event.getUid(), false, 0);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.fail();
+ sinkContext.addSendResultMetric(profile, clusterName,
profile.getUid(), false, 0);
+ sinkContext.getDispatchQueue().release(profile.getSize());
+ profile.fail(DataProxyErrCode.TOPIC_IS_BLANK, "");
return false;
}
// create producer failed
if (producer == null) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_NOPRODUCER);
- sinkContext.processSendFail(event, clusterName, topic, 0);
+ sinkContext.processSendFail(profile, clusterName, topic, 0,
+ DataProxyErrCode.PRODUCER_IS_NULL, "");
LOG.error("producer is null");
return false;
}
@@ -211,26 +214,25 @@ public class TubeHandler implements MessageQueueHandler {
this.topicSet.add(topic);
}
// send
- if (event instanceof SimpleBatchPackProfileV0) {
- this.sendSimpleProfileV0((SimpleBatchPackProfileV0) event,
idConfig, topic);
- } else if (event instanceof OrderBatchPackProfileV0) {
- this.sendOrderProfileV0((OrderBatchPackProfileV0) event,
idConfig, topic);
+ if (profile instanceof SimplePackProfile) {
+ this.sendSimplePackProfile((SimplePackProfile) profile,
idConfig, topic);
} else {
- this.sendProfileV1(event, idConfig, topic);
+ this.sendBatchPackProfile((BatchPackProfile) profile,
idConfig, topic);
}
return true;
- } catch (Exception e) {
+ } catch (Exception ex) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SENDEXCEPT);
- sinkContext.processSendFail(event, clusterName, event.getUid(), 0);
- LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(profile, clusterName,
profile.getUid(), 0,
+ DataProxyErrCode.SEND_REQUEST_TO_MQ_FAILURE,
ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
return false;
}
}
/**
- * sendProfileV1
+ * send BatchPackProfile
*/
- private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
+ private void sendBatchPackProfile(BatchPackProfile batchProfile,
IdTopicConfig idConfig,
String topic) throws Exception {
EventHandler handler = handlerLocal.get();
if (handler == null) {
@@ -238,12 +240,12 @@ public class TubeHandler implements MessageQueueHandler {
handlerLocal.set(handler);
}
// headers
- Map<String, String> headers = handler.parseHeader(idConfig, event,
sinkContext.getNodeId(),
+ Map<String, String> headers = handler.parseHeader(idConfig,
batchProfile, sinkContext.getNodeId(),
sinkContext.getCompressType());
// compress
- byte[] bodyBytes = handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ byte[] bodyBytes = handler.parseBody(idConfig, batchProfile,
sinkContext.getCompressType());
// metric
- sinkContext.addSendMetric(event, clusterName, topic, bodyBytes.length);
+ sinkContext.addSendMetric(batchProfile, clusterName, topic,
bodyBytes.length);
// sendAsync
Message message = new Message(topic, bodyBytes);
// add headers
@@ -257,15 +259,16 @@ public class TubeHandler implements MessageQueueHandler {
@Override
public void onMessageSent(MessageSentResult result) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
+ sinkContext.addSendResultMetric(batchProfile, clusterName,
topic, true, sendTime);
+ sinkContext.getDispatchQueue().release(batchProfile.getSize());
+ batchProfile.ack();
}
@Override
public void onException(Throwable ex) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
+ sinkContext.processSendFail(batchProfile, clusterName, topic,
sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
if (logCounter.shouldPrint()) {
LOG.error("Send ProfileV1 to tube failure", ex);
}
@@ -277,12 +280,12 @@ public class TubeHandler implements MessageQueueHandler {
/**
* sendSimpleProfileV0
*/
- private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ private void sendSimplePackProfile(SimplePackProfile simpleProfile,
IdTopicConfig idConfig,
String topic) throws Exception {
// build message
- Message message = TubeUtils.buildMessage(topic,
event.getSimpleProfile());
+ Message message = TubeUtils.buildMessage(topic,
simpleProfile.getEvent());
// metric
- sinkContext.addSendMetric(event, clusterName, topic,
event.getSimpleProfile().getBody().length);
+ sinkContext.addSendMetric(simpleProfile, clusterName, topic,
simpleProfile.getEvent().getBody().length);
// callback
long sendTime = System.currentTimeMillis();
MessageSentCallback callback = new MessageSentCallback() {
@@ -290,15 +293,16 @@ public class TubeHandler implements MessageQueueHandler {
@Override
public void onMessageSent(MessageSentResult result) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
+ sinkContext.addSendResultMetric(simpleProfile, clusterName,
topic, true, sendTime);
+
sinkContext.getDispatchQueue().release(simpleProfile.getSize());
+ simpleProfile.ack();
}
@Override
public void onException(Throwable ex) {
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
+ sinkContext.processSendFail(simpleProfile, clusterName, topic,
sendTime,
+ DataProxyErrCode.MQ_RETURN_ERROR, ex.getMessage());
if (logCounter.shouldPrint()) {
LOG.error("Send SimpleProfileV0 to tube failure", ex);
}
@@ -306,44 +310,4 @@ public class TubeHandler implements MessageQueueHandler {
};
producer.sendMessage(message, callback);
}
-
- /**
- * sendOrderProfileV0
- */
- private void sendOrderProfileV0(OrderBatchPackProfileV0 event,
IdTopicConfig idConfig, String topic)
- throws Exception {
- // headers
- Map<String, String> headers = event.getOrderProfile().getHeaders();
- // compress
- byte[] bodyBytes = event.getOrderProfile().getBody();
- // metric
- sinkContext.addSendMetric(event, clusterName, topic, bodyBytes.length);
- // sendAsync
- Message message = new Message(topic, bodyBytes);
- // add headers
- headers.forEach(message::setAttrKeyVal);
- // callback
- long sendTime = System.currentTimeMillis();
- MessageSentCallback callback = new MessageSentCallback() {
-
- @Override
- public void onMessageSent(MessageSentResult result) {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_SUCCESS);
- sinkContext.addSendResultMetric(event, clusterName, topic,
true, sendTime);
- sinkContext.getDispatchQueue().release(event.getSize());
- event.ack();
- event.ackOrder();
- }
-
- @Override
- public void onException(Throwable ex) {
-
sinkContext.fileMetricEventInc(StatConstants.EVENT_SINK_RECEIVEEXCEPT);
- sinkContext.processSendFail(event, clusterName, topic,
sendTime);
- if (logCounter.shouldPrint()) {
- LOG.error("Send OrderProfileV0 to tube failure", ex);
- }
- }
- };
- producer.sendMessage(message, callback);
- }
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
index 8909f3b3a5..d030eeb1a8 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/InLongMessageHandler.java
@@ -149,7 +149,8 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
throw new Exception("Unknown V0 message type, type = " +
msgTypeValue);
} else if (MsgType.MSG_HEARTBEAT == msgType) {
// send response message
- flushV0MsgPackage(source, channel,
buildHeartBeatMsgRspPackage(), MsgType.MSG_HEARTBEAT.name());
+ flushV0MsgPackage(source, channel,
+ buildHeartBeatMsgRspPackage(),
MsgType.MSG_HEARTBEAT.name());
return;
} else if (MsgType.MSG_BIN_HEARTBEAT == msgType) {
procBinHeartbeatMsg(source, channel, cb, totalDataLen);
@@ -524,8 +525,8 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
+ bodyLen + BIN_HB_ATTRLEN_SIZE, attrData, 0, attrLen);
}
// build and send response message
- flushV0MsgPackage(source, channel, buildHBRspPackage(attrData,
version, 0),
- MsgType.MSG_BIN_HEARTBEAT.name());
+ flushV0MsgPackage(source, channel,
+ buildHBRspPackage(attrData, version, 0),
MsgType.MSG_BIN_HEARTBEAT.name());
}
/**
@@ -535,7 +536,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
* @param uniqVal sequence Id
* @return ByteBuf
*/
- private ByteBuf buildBinMsgRspPackage(String attrs, long uniqVal) {
+ public static ByteBuf buildBinMsgRspPackage(String attrs, long uniqVal) {
// calculate total length
// binTotalLen = mstType + uniq + attrsLen + attrs + magic
int binTotalLen = 1 + 4 + 2 + 2;
@@ -569,7 +570,7 @@ public class InLongMessageHandler extends
ChannelInboundHandlerAdapter {
* @param attrs the return attribute
* @return ByteBuf
*/
- private ByteBuf buildTxtMsgRspPackage(MsgType msgType, String attrs) {
+ public static ByteBuf buildTxtMsgRspPackage(MsgType msgType, String attrs)
{
int attrsLen = 0;
int bodyLen = 0;
if (attrs != null) {
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
index eaa451de37..b99d28cddc 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source2/v0msg/AbsV0MsgCodec.java
@@ -192,6 +192,7 @@ public abstract class AbsV0MsgCodec {
headers.put(ConfigConstants.MSG_COUNTER_KEY, String.valueOf(msgCount));
headers.put(ConfigConstants.MSG_ENCODE_VER,
InLongMsgVer.INLONG_V0.getName());
headers.put(AttributeConstants.RCV_TIME, String.valueOf(msgRcvTime));
+ headers.put(AttributeConstants.UNIQ_ID, String.valueOf(uniq));
// add extra key-value information
String pkgTimeStr = attrMap.get(ConfigConstants.PKG_TIME_KEY);
if (StringUtils.isBlank(pkgTimeStr)) {
diff --git
a/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestCommonConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestCommonConfigHolder.java
index 2a7be2668f..930eccc700 100644
---
a/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestCommonConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestCommonConfigHolder.java
@@ -37,6 +37,6 @@ public class TestCommonConfigHolder {
Assert.assertTrue(CommonConfigHolder.getInstance().isEnableWhiteList());
assertEquals("DataProxy",
CommonConfigHolder.getInstance().getProperties().get(MetricListener.KEY_METRIC_DOMAINS));
- assertEquals(50,
CommonConfigHolder.getInstance().getMetaConfigSyncInvlMs());
+ assertEquals(10000,
CommonConfigHolder.getInstance().getMetaConfigSyncInvlMs());
}
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestMetaConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestMetaConfigHolder.java
index 9bc0f0c4b4..5faf4499c7 100644
---
a/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestMetaConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/config/holder/TestMetaConfigHolder.java
@@ -17,6 +17,7 @@
package org.apache.inlong.dataproxy.config.holder;
+import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
import org.junit.Assert;
@@ -31,6 +32,7 @@ public class TestMetaConfigHolder {
@Test
public void test() {
+ ConfigManager.handshakeManagerOk.set(true);
MetaConfigHolder metaConfigHolder = new MetaConfigHolder();
boolean result = metaConfigHolder.loadFromFileToHolder();
Assert.assertTrue(result);
diff --git
a/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
b/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
index 9f58a5467c..49d0e277c0 100644
--- a/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
+++ b/inlong-dataproxy/dataproxy-source/src/test/resources/common.properties
@@ -19,7 +19,9 @@ proxy.cluster.name=proxy_inlong5th_sz
metricDomains=DataProxy
metricDomains.DataProxy.domainListeners=org.apache.inlong.dataproxy.metrics.prometheus.PrometheusMetricListener
metricDomains.DataProxy.snapshotInterval=60000
+# whether to startup using the local metadata.json file without connecting to
the Manager
+startup.using.local.meta.file.enable=true
proxy.enable.whitelist=true
-meta.config.sync.interval.ms=50
+meta.config.sync.interval.ms=10000
