This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new d95c2977f [INLONG-6153][DataProxy] Optimize the sink architecture
(#6219)
d95c2977f is described below
commit d95c2977f0e2dcc501e5f5e85cdfdce5103dd6b2
Author: 卢春亮 <[email protected]>
AuthorDate: Mon Nov 21 16:16:43 2022 +0800
[INLONG-6153][DataProxy] Optimize the sink architecture (#6219)
---
.../config/holder/IdTopicConfigHolder.java | 27 +-
.../dataproxy/sink/common/DefaultEventHandler.java | 116 +++++++
.../inlong/dataproxy/sink/common/EventHandler.java | 43 +++
.../inlong/dataproxy/sink/common/SinkContext.java | 248 ++++++++++++++
.../inlong/dataproxy/sink/mq/BatchPackManager.java | 209 ++++++++++++
.../inlong/dataproxy/sink/mq/BatchPackProfile.java | 219 +++++++++++++
.../sink/mq/BatchPackProfileCallback.java | 61 ++++
.../sink/mq/MessageQueueClusterProducer.java | 129 ++++++++
.../dataproxy/sink/mq/MessageQueueHandler.java | 50 +++
.../sink/mq/MessageQueueZoneProducer.java | 164 +++++++++
.../dataproxy/sink/mq/MessageQueueZoneSink.java | 167 ++++++++++
.../sink/mq/MessageQueueZoneSinkContext.java | 265 +++++++++++++++
.../dataproxy/sink/mq/MessageQueueZoneWorker.java | 108 ++++++
.../dataproxy/sink/mq/OrderBatchPackProfileV0.java | 105 ++++++
.../sink/mq/SimpleBatchPackProfileV0.java | 73 +++++
.../dataproxy/sink/mq/kafka/KafkaHandler.java | 267 +++++++++++++++
.../dataproxy/sink/mq/pulsar/PulsarHandler.java | 365 +++++++++++++++++++++
.../inlong/dataproxy/sink/mq/tube/TubeHandler.java | 307 +++++++++++++++++
18 files changed, 2913 insertions(+), 10 deletions(-)
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
index 497db56d0..5ace034bc 100644
---
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/IdTopicConfigHolder.java
@@ -17,8 +17,14 @@
package org.apache.inlong.dataproxy.config.holder;
-import static
org.apache.inlong.dataproxy.config.loader.ConfigLoader.RELOAD_INTERVAL;
-import static
org.apache.inlong.dataproxy.config.loader.IdTopicConfigLoader.IDTOPIC_CONFIG_TYPE;
+import org.apache.commons.lang.ClassUtils;
+import org.apache.flume.Context;
+import org.apache.flume.conf.Configurable;
+import org.apache.inlong.dataproxy.config.loader.ContextIdTopicConfigLoader;
+import org.apache.inlong.dataproxy.config.loader.IdTopicConfigLoader;
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Date;
@@ -28,14 +34,8 @@ import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
-import org.apache.commons.lang3.ClassUtils;
-import org.apache.flume.Context;
-import org.apache.flume.conf.Configurable;
-import org.apache.inlong.dataproxy.config.loader.ContextIdTopicConfigLoader;
-import org.apache.inlong.dataproxy.config.loader.IdTopicConfigLoader;
-import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static
org.apache.inlong.dataproxy.config.loader.ConfigLoader.RELOAD_INTERVAL;
+import static
org.apache.inlong.dataproxy.config.loader.IdTopicConfigLoader.IDTOPIC_CONFIG_TYPE;
/**
*
@@ -160,4 +160,11 @@ public class IdTopicConfigHolder implements Configurable {
}
return null;
}
+
+ /**
+ * getIdConfig
+ */
+ public IdTopicConfig getIdConfig(String uid) {
+ return this.configMap.get(uid);
+ }
}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/DefaultEventHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/DefaultEventHandler.java
new file mode 100644
index 000000000..22ae6113a
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/DefaultEventHandler.java
@@ -0,0 +1,116 @@
+/**
+ * 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.common;
+
+import com.google.protobuf.ByteString;
+
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
+import org.apache.inlong.sdk.commons.protocol.EventConstants;
+import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.INLONG_COMPRESSED_TYPE;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MapFieldEntry;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObj;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObjs;
+import org.apache.inlong.sdk.commons.utils.GzipUtils;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.inlong.sdk.commons.protocol.EventConstants.HEADER_CACHE_VERSION_1;
+import static
org.apache.inlong.sdk.commons.protocol.EventConstants.HEADER_KEY_VERSION;
+
+/**
+ * DefaultEventHandler
+ *
+ */
+public class DefaultEventHandler implements EventHandler {
+
+ /**
+ * parseHeader
+ */
+ @Override
+ public Map<String, String> parseHeader(IdTopicConfig idConfig,
BatchPackProfile profile, String nodeId,
+ INLONG_COMPRESSED_TYPE compressType) {
+ Map<String, String> headers = new HashMap<>();
+ // version int32 protocol version, the value is 1
+ headers.put(HEADER_KEY_VERSION, HEADER_CACHE_VERSION_1);
+ // inlongGroupId string inlongGroupId
+ headers.put(EventConstants.INLONG_GROUP_ID,
profile.getInlongGroupId());
+ // inlongStreamId string inlongStreamId
+ headers.put(EventConstants.INLONG_STREAM_ID,
profile.getInlongStreamId());
+ // proxyName string proxy node id, IP or conainer name
+ headers.put(EventConstants.HEADER_KEY_PROXY_NAME, nodeId);
+ // packTime int64 pack time, milliseconds
+ headers.put(EventConstants.HEADER_KEY_PACK_TIME,
String.valueOf(System.currentTimeMillis()));
+ // msgCount int32 message count
+ headers.put(EventConstants.HEADER_KEY_MSG_COUNT,
String.valueOf(profile.getEvents().size()));
+ // srcLength int32 total length of raw messages body
+ headers.put(EventConstants.HEADER_KEY_SRC_LENGTH,
String.valueOf(profile.getSize()));
+ // compressType int
+ // compress type of body data
+ // INLONG_NO_COMPRESS = 0,
+ // INLONG_GZ = 1,
+ // INLONG_SNAPPY = 2
+ headers.put(EventConstants.HEADER_KEY_COMPRESS_TYPE,
+ String.valueOf(compressType.getNumber()));
+ // messageKey string partition hash key, optional
+ return headers;
+ }
+
+ /**
+ * parseBody
+ */
+ @Override
+ public byte[] parseBody(IdTopicConfig idConfig, BatchPackProfile profile,
INLONG_COMPRESSED_TYPE compressType)
+ throws IOException {
+ List<ProxyEvent> events = profile.getEvents();
+ // encode
+ MessageObjs.Builder objs = MessageObjs.newBuilder();
+ for (ProxyEvent event : events) {
+ MessageObj.Builder builder = MessageObj.newBuilder();
+ builder.setMsgTime(event.getMsgTime());
+ builder.setSourceIp(event.getSourceIp());
+ event.getHeaders().forEach((key, value) -> {
+
builder.addParams(MapFieldEntry.newBuilder().setKey(key).setValue(value));
+ });
+ builder.setBody(ByteString.copyFrom(event.getBody()));
+ objs.addMsgs(builder.build());
+ }
+ byte[] srcBytes = objs.build().toByteArray();
+ // compress
+ byte[] compressBytes = null;
+ switch (compressType) {
+ case INLONG_SNAPPY :
+ compressBytes = Snappy.compress(srcBytes);
+ break;
+ case INLONG_GZ :
+ compressBytes = GzipUtils.compress(srcBytes);
+ break;
+ case INLONG_NO_COMPRESS :
+ default :
+ compressBytes = srcBytes;
+ break;
+ }
+ return compressBytes;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/EventHandler.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/EventHandler.java
new file mode 100644
index 000000000..3a2d023b5
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/EventHandler.java
@@ -0,0 +1,43 @@
+/**
+ * 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.common;
+
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+import org.apache.inlong.dataproxy.sink.mq.BatchPackProfile;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.INLONG_COMPRESSED_TYPE;
+
+import java.util.Map;
+
+/**
+ * IEventHandler
+ *
+ */
+public interface EventHandler {
+
+ /**
+ * parseHeader
+ */
+ Map<String, String> parseHeader(IdTopicConfig idConfig, BatchPackProfile
profile, String nodeId,
+ INLONG_COMPRESSED_TYPE compressType) throws Exception;
+
+ /**
+ * parseBody
+ */
+ byte[] parseBody(IdTopicConfig idConfig, BatchPackProfile profile,
INLONG_COMPRESSED_TYPE compressType)
+ throws Exception;
+}
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
new file mode 100644
index 000000000..11298296e
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/common/SinkContext.java
@@ -0,0 +1,248 @@
+/*
+ * 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.common;
+
+import org.apache.commons.lang.ClassUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.inlong.common.metric.MetricRegister;
+import org.apache.inlong.dataproxy.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
+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.pulsar.PulsarHandler;
+import org.apache.inlong.dataproxy.utils.BufferQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+import java.util.Timer;
+import java.util.TimerTask;
+
+/**
+ * SinkContext
+ */
+public class SinkContext {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(SinkContext.class);
+
+ public static final String KEY_MAX_THREADS = "maxThreads";
+ public static final String KEY_PROCESSINTERVAL = "processInterval";
+ public static final String KEY_RELOADINTERVAL = "reloadInterval";
+ public static final String KEY_EVENT_HANDLER = "eventHandler";
+ public static final String KEY_MESSAGE_QUEUE_HANDLER =
"messageQueueHandler";
+ public static final String KEY_MAX_BUFFERQUEUE_SIZE_KB =
"maxBufferQueueSizeKb";
+ public static final int DEFAULT_MAX_BUFFERQUEUE_SIZE_KB = 128 * 1024;
+
+ protected final String clusterId;
+ protected final String sinkName;
+ protected final Context sinkContext;
+
+ protected final Channel channel;
+ //
+ protected final int maxThreads;
+ protected final long processInterval;
+ protected final long reloadInterval;
+ //
+ protected final DataProxyMetricItemSet metricItemSet;
+ protected Timer reloadTimer;
+
+ /**
+ * Constructor
+ */
+ public SinkContext(String sinkName, Context context, Channel channel) {
+ this.sinkName = sinkName;
+ this.sinkContext = context;
+ this.channel = channel;
+ this.clusterId =
CommonPropertiesHolder.getString(CommonPropertiesHolder.KEY_PROXY_CLUSTER_NAME);
+ this.maxThreads = sinkContext.getInteger(KEY_MAX_THREADS, 10);
+ this.processInterval = sinkContext.getInteger(KEY_PROCESSINTERVAL,
100);
+ this.reloadInterval = sinkContext.getLong(KEY_RELOADINTERVAL, 60000L);
+ //
+ this.metricItemSet = new DataProxyMetricItemSet(sinkName);
+ MetricRegister.register(this.metricItemSet);
+ }
+
+ /**
+ * start
+ */
+ public void start() {
+ try {
+ this.reload();
+ this.setReloadTimer();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * close
+ */
+ public void close() {
+ try {
+ this.reloadTimer.cancel();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * setReloadTimer
+ */
+ protected void setReloadTimer() {
+ reloadTimer = new Timer(true);
+ TimerTask task = new TimerTask() {
+
+ public void run() {
+ reload();
+ }
+ };
+ reloadTimer.schedule(task, new Date(System.currentTimeMillis() +
reloadInterval), reloadInterval);
+ }
+
+ /**
+ * reload
+ */
+ public void reload() {
+ }
+
+ /**
+ * get clusterId
+ *
+ * @return the clusterId
+ */
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ /**
+ * get sinkName
+ *
+ * @return the sinkName
+ */
+ public String getSinkName() {
+ return sinkName;
+ }
+
+ /**
+ * get sinkContext
+ *
+ * @return the sinkContext
+ */
+ public Context getSinkContext() {
+ return sinkContext;
+ }
+
+ /**
+ * get channel
+ *
+ * @return the channel
+ */
+ public Channel getChannel() {
+ return channel;
+ }
+
+ /**
+ * get maxThreads
+ *
+ * @return the maxThreads
+ */
+ public int getMaxThreads() {
+ return maxThreads;
+ }
+
+ /**
+ * get processInterval
+ *
+ * @return the processInterval
+ */
+ public long getProcessInterval() {
+ return processInterval;
+ }
+
+ /**
+ * get reloadInterval
+ *
+ * @return the reloadInterval
+ */
+ public long getReloadInterval() {
+ return reloadInterval;
+ }
+
+ /**
+ * get metricItemSet
+ *
+ * @return the metricItemSet
+ */
+ public DataProxyMetricItemSet getMetricItemSet() {
+ return metricItemSet;
+ }
+
+ /**
+ * createEventHandler
+ */
+ public EventHandler createEventHandler() {
+ // IEventHandler
+ String eventHandlerClass =
CommonPropertiesHolder.getString(KEY_EVENT_HANDLER,
+ DefaultEventHandler.class.getName());
+ try {
+ Class<?> handlerClass = ClassUtils.getClass(eventHandlerClass);
+ Object handlerObject =
handlerClass.getDeclaredConstructor().newInstance();
+ if (handlerObject instanceof EventHandler) {
+ EventHandler handler = (EventHandler) handlerObject;
+ return handler;
+ }
+ } catch (Throwable t) {
+ LOG.error("Fail to init EventHandler,handlerClass:{},error:{}",
+ eventHandlerClass, t.getMessage(), t);
+ }
+ return null;
+ }
+
+ /**
+ * createMessageQueueHandler
+ */
+ public MessageQueueHandler createMessageQueueHandler(CacheClusterConfig
config) {
+ String strHandlerClass =
config.getParams().getOrDefault(KEY_MESSAGE_QUEUE_HANDLER,
+ PulsarHandler.class.getName());
+ try {
+ Class<?> handlerClass = ClassUtils.getClass(strHandlerClass);
+ Object handlerObject =
handlerClass.getDeclaredConstructor().newInstance();
+ if (handlerObject instanceof MessageQueueHandler) {
+ MessageQueueHandler handler = (MessageQueueHandler)
handlerObject;
+ return handler;
+ }
+ } catch (Throwable t) {
+ LOG.error("Fail to init
MessageQueueHandler,handlerClass:{},error:{}",
+ strHandlerClass, t.getMessage(), t);
+ }
+ return null;
+ }
+
+ /**
+ * createBufferQueue
+ * @return
+ */
+ public static BufferQueue<BatchPackProfile> createBufferQueue() {
+ int maxBufferQueueSizeKb =
CommonPropertiesHolder.getInteger(KEY_MAX_BUFFERQUEUE_SIZE_KB,
+ DEFAULT_MAX_BUFFERQUEUE_SIZE_KB);
+ BufferQueue<BatchPackProfile> dispatchQueue = new
BufferQueue<BatchPackProfile>(maxBufferQueueSizeKb);
+ return dispatchQueue;
+ }
+}
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
new file mode 100644
index 000000000..fc99b3c05
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackManager.java
@@ -0,0 +1,209 @@
+/**
+ * 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.flume.Context;
+import org.apache.inlong.dataproxy.utils.BufferQueue;
+import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.inlong.sdk.commons.protocol.ProxyPackEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * BatchPackManager
+ */
+public class BatchPackManager {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(BatchPackManager.class);
+ public static final String KEY_DISPATCH_TIMEOUT = "dispatchTimeout";
+ public static final String KEY_DISPATCH_MAX_PACKCOUNT =
"dispatchMaxPackCount";
+ public static final String KEY_DISPATCH_MAX_PACKSIZE =
"dispatchMaxPackSize";
+ public static final long DEFAULT_DISPATCH_TIMEOUT = 2000;
+ public static final long DEFAULT_DISPATCH_MAX_PACKCOUNT = 256;
+ public static final long DEFAULT_DISPATCH_MAX_PACKSIZE = 327680;
+ public static final long MINUTE_MS = 60L * 1000;
+
+ private final long dispatchTimeout;
+ private final long maxPackCount;
+ private final long maxPackSize;
+ private BufferQueue<BatchPackProfile> dispatchQueue;
+ private ConcurrentHashMap<String, BatchPackProfile> profileCache = new
ConcurrentHashMap<>();
+ // flag that manager need to output overtime data.
+ private AtomicBoolean needOutputOvertimeData = new AtomicBoolean(false);
+ private AtomicLong inCounter = new AtomicLong(0);
+ private AtomicLong outCounter = new AtomicLong(0);
+
+ /**
+ * Constructor
+ *
+ * @param context
+ * @param dispatchQueue
+ */
+ public BatchPackManager(Context context, BufferQueue<BatchPackProfile>
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);
+ this.maxPackSize = context.getLong(KEY_DISPATCH_MAX_PACKSIZE,
DEFAULT_DISPATCH_MAX_PACKSIZE);
+ }
+
+ /**
+ * addEvent
+ *
+ * @param event
+ */
+ public void addEvent(ProxyEvent event) {
+ // parse
+ String eventUid = event.getUid();
+ long dispatchTime = event.getMsgTime() - event.getMsgTime() %
MINUTE_MS;
+ String dispatchKey = eventUid + "." + dispatchTime;
+ // find dispatch profile
+ BatchPackProfile dispatchProfile = this.profileCache.get(dispatchKey);
+ if (dispatchProfile == null) {
+ dispatchProfile = new BatchPackProfile(eventUid,
event.getInlongGroupId(), event.getInlongStreamId(),
+ dispatchTime);
+ this.profileCache.put(dispatchKey, dispatchProfile);
+ }
+ // add event
+ boolean addResult = dispatchProfile.addEvent(event, maxPackCount,
maxPackSize);
+ if (!addResult) {
+ BatchPackProfile newDispatchProfile = new
BatchPackProfile(eventUid, event.getInlongGroupId(),
+ event.getInlongStreamId(), dispatchTime);
+ BatchPackProfile oldDispatchProfile =
this.profileCache.put(dispatchKey, newDispatchProfile);
+ this.dispatchQueue.acquire(oldDispatchProfile.getSize());
+ this.dispatchQueue.offer(oldDispatchProfile);
+ outCounter.addAndGet(dispatchProfile.getCount());
+ newDispatchProfile.addEvent(event, maxPackCount, maxPackSize);
+ }
+ inCounter.incrementAndGet();
+ }
+
+ /**
+ * addPackEvent
+ * @param packEvent
+ */
+ public void addPackEvent(ProxyPackEvent packEvent) {
+ String eventUid = packEvent.getUid();
+ long dispatchTime = packEvent.getMsgTime() - packEvent.getMsgTime() %
MINUTE_MS;
+ BatchPackProfile dispatchProfile = new BatchPackProfile(eventUid,
packEvent.getInlongGroupId(),
+ packEvent.getInlongStreamId(), dispatchTime);
+ // callback
+ BatchPackProfileCallback callback = new
BatchPackProfileCallback(packEvent.getEvents().size(),
+ packEvent.getCallback());
+ dispatchProfile.setCallback(callback);
+ // offer queue
+ for (ProxyEvent event : packEvent.getEvents()) {
+ inCounter.incrementAndGet();
+ boolean addResult = dispatchProfile.addEvent(event, maxPackCount,
maxPackSize);
+ // dispatch profile is full
+ if (!addResult) {
+ outCounter.addAndGet(dispatchProfile.getCount());
+ this.dispatchQueue.acquire(dispatchProfile.getSize());
+ this.dispatchQueue.offer(dispatchProfile);
+ dispatchProfile = new BatchPackProfile(eventUid,
event.getInlongGroupId(), event.getInlongStreamId(),
+ dispatchTime);
+ dispatchProfile.setCallback(callback);
+ dispatchProfile.addEvent(event, maxPackCount, maxPackSize);
+ }
+ }
+ // last dispatch profile
+ if (dispatchProfile.getEvents().size() > 0) {
+ outCounter.addAndGet(dispatchProfile.getCount());
+ this.dispatchQueue.acquire(dispatchProfile.getSize());
+ this.dispatchQueue.offer(dispatchProfile);
+ }
+ }
+
+ /**
+ * outputOvertimeData
+ *
+ * @return
+ */
+ public void outputOvertimeData() {
+ if (!needOutputOvertimeData.getAndSet(false)) {
+ return;
+ }
+ LOG.info("start to outputOvertimeData
profileCacheSize:{},dispatchQueueSize:{}",
+ profileCache.size(), dispatchQueue.size());
+ long currentTime = System.currentTimeMillis();
+ long createThreshold = currentTime - dispatchTimeout;
+ List<String> removeKeys = new ArrayList<>();
+ long eventCount = 0;
+ for (Entry<String, BatchPackProfile> entry :
this.profileCache.entrySet()) {
+ BatchPackProfile dispatchProfile = entry.getValue();
+ eventCount += dispatchProfile.getCount();
+ if (!dispatchProfile.isTimeout(createThreshold)) {
+ continue;
+ }
+ removeKeys.add(entry.getKey());
+ }
+ // output
+ removeKeys.forEach((key) -> {
+ BatchPackProfile dispatchProfile = this.profileCache.remove(key);
+ if (dispatchProfile != null) {
+ this.dispatchQueue.acquire(dispatchProfile.getSize());
+ dispatchQueue.offer(dispatchProfile);
+ outCounter.addAndGet(dispatchProfile.getCount());
+ }
+ });
+ LOG.info("end to outputOvertimeData
profileCacheSize:{},dispatchQueueSize:{},eventCount:{},"
+ + "inCounter:{},outCounter:{}",
+ profileCache.size(), dispatchQueue.size(), eventCount,
+ inCounter.getAndSet(0), outCounter.getAndSet(0));
+ }
+
+ /**
+ * get dispatchTimeout
+ *
+ * @return the dispatchTimeout
+ */
+ public long getDispatchTimeout() {
+ return dispatchTimeout;
+ }
+
+ /**
+ * get maxPackCount
+ *
+ * @return the maxPackCount
+ */
+ public long getMaxPackCount() {
+ return maxPackCount;
+ }
+
+ /**
+ * get maxPackSize
+ *
+ * @return the maxPackSize
+ */
+ public long getMaxPackSize() {
+ return maxPackSize;
+ }
+
+ /**
+ * setNeedOutputOvertimeData
+ */
+ public void setNeedOutputOvertimeData() {
+ this.needOutputOvertimeData.getAndSet(true);
+ }
+}
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
new file mode 100644
index 000000000..a4dfb3a11
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfile.java
@@ -0,0 +1,219 @@
+/**
+ * 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.sdk.commons.protocol.ProxyEvent;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ *
+ * DispatchProfile
+ */
+public class BatchPackProfile {
+ public static final long MINUTE_MS = 60L * 1000;
+
+ 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;
+ }
+
+ /**
+ * 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
+ *
+ * @param createThreshold
+ * @return
+ */
+ public boolean isTimeout(long createThreshold) {
+ return createThreshold >= createTime;
+ }
+
+ /**
+ * get uid
+ *
+ * @return the uid
+ */
+ public String getUid() {
+ return uid;
+ }
+
+ /**
+ * get events
+ *
+ * @return the events
+ */
+ public List<ProxyEvent> getEvents() {
+ return events;
+ }
+
+ /**
+ * set events
+ *
+ * @param events the events to set
+ */
+ public void setEvents(List<ProxyEvent> events) {
+ 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
+ */
+ public void ack() {
+ if (callback != null) {
+ callback.ack(this.events.size());
+ }
+ }
+
+ /**
+ * fail
+ * @return
+ */
+ public void fail() {
+ if (callback != null) {
+ callback.fail();
+ }
+ }
+
+ /**
+ * isResend
+ * @return
+ */
+ public boolean isResend() {
+ return callback == null;
+ }
+
+ /**
+ * get callback
+ * @return the callback
+ */
+ public BatchPackProfileCallback getCallback() {
+ return callback;
+ }
+
+ /**
+ * set callback
+ * @param callback the callback to set
+ */
+ public void setCallback(BatchPackProfileCallback callback) {
+ this.callback = callback;
+ }
+
+}
diff --git
a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfileCallback.java
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfileCallback.java
new file mode 100644
index 000000000..098bd9359
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/BatchPackProfileCallback.java
@@ -0,0 +1,61 @@
+/**
+ * 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.sdk.commons.protocol.ProxySdk.ResultCode;
+import org.apache.inlong.sdk.commons.protocol.SourceCallback;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * DispatchProfileCallback
+ *
+ */
+public class BatchPackProfileCallback {
+
+ private AtomicInteger ackingCount;
+ private SourceCallback callback;
+
+ /**
+ * Constructor
+ * @param totalCount
+ * @param callback
+ */
+ public BatchPackProfileCallback(int totalCount, SourceCallback callback) {
+ this.ackingCount = new AtomicInteger(totalCount);
+ this.callback = callback;
+ }
+
+ /**
+ * ack
+ * @param eventCount
+ */
+ public void ack(int eventCount) {
+ int currentCount = this.ackingCount.addAndGet(-eventCount);
+ if (currentCount <= 0) {
+ this.callback.callback(ResultCode.SUCCUSS);
+ }
+ }
+
+ /**
+ * fail
+ */
+ public void fail() {
+ this.callback.callback(ResultCode.ERR_REJECT);
+ }
+}
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
new file mode 100644
index 000000000..aeedefe6b
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueClusterProducer.java
@@ -0,0 +1,129 @@
+/**
+ * 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.flume.lifecycle.LifecycleAware;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MessageQueueClusterProducer
+ */
+public class MessageQueueClusterProducer implements LifecycleAware {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueClusterProducer.class);
+
+ private final String workerName;
+ private final CacheClusterConfig config;
+ private final MessageQueueZoneSinkContext sinkContext;
+ private final String cacheClusterName;
+ private LifecycleState state;
+
+ private MessageQueueHandler handler;
+
+ /**
+ * Constructor
+ *
+ * @param workerName
+ * @param config
+ * @param context
+ */
+ public MessageQueueClusterProducer(String workerName, CacheClusterConfig
config,
+ MessageQueueZoneSinkContext context) {
+ this.workerName = workerName;
+ this.config = config;
+ this.sinkContext = context;
+ this.state = LifecycleState.IDLE;
+ this.cacheClusterName = config.getClusterName();
+ this.handler = this.sinkContext.createMessageQueueHandler(config);
+ this.handler.init(config, context);
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ this.state = LifecycleState.START;
+ this.handler.start();
+ }
+
+ /**
+ * stop
+ */
+ @Override
+ public void stop() {
+ this.state = LifecycleState.STOP;
+ this.handler.stop();
+ }
+
+ /**
+ * getLifecycleState
+ *
+ * @return
+ */
+ @Override
+ public LifecycleState getLifecycleState() {
+ return state;
+ }
+
+ /**
+ * send
+ *
+ * @param event
+ */
+ public boolean send(BatchPackProfile event) {
+ return this.handler.send(event);
+ }
+
+ /**
+ * get cacheClusterName
+ *
+ * @return the cacheClusterName
+ */
+ public String getCacheClusterName() {
+ return cacheClusterName;
+ }
+
+ /**
+ * get workerName
+ * @return the workerName
+ */
+ public String getWorkerName() {
+ return workerName;
+ }
+
+ /**
+ * get state
+ * @return the state
+ */
+ public LifecycleState getState() {
+ return state;
+ }
+
+ /**
+ * get config
+ * @return the config
+ */
+ public CacheClusterConfig getConfig() {
+ return config;
+ }
+
+}
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
new file mode 100644
index 000000000..4b8417dd7
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueHandler.java
@@ -0,0 +1,50 @@
+/**
+ * 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.dataproxy.config.pojo.CacheClusterConfig;
+
+/**
+ * MessageQueueHandler
+ */
+public interface MessageQueueHandler {
+
+ /**
+ * init
+ * @param config
+ * @param sinkContext
+ */
+ void init(CacheClusterConfig config, MessageQueueZoneSinkContext
sinkContext);
+
+ /**
+ * start
+ */
+ void start();
+
+ /**
+ * stop
+ */
+ void stop();
+
+ /**
+ * send
+ * @param event
+ * @return
+ */
+ boolean send(BatchPackProfile event);
+}
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
new file mode 100644
index 000000000..bc85b6097
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneProducer.java
@@ -0,0 +1,164 @@
+/**
+ * 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.dataproxy.config.pojo.CacheClusterConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ *
+ * MessageQueueZoneProducer
+ */
+public class MessageQueueZoneProducer {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueZoneProducer.class);
+ public static final int MAX_INDEX = Integer.MAX_VALUE / 2;
+
+ private final String workerName;
+ private final MessageQueueZoneSinkContext context;
+ private Timer reloadTimer;
+
+ private List<MessageQueueClusterProducer> clusterList = new ArrayList<>();
+ private List<MessageQueueClusterProducer> deletingClusterList = new
ArrayList<>();
+
+ private AtomicInteger clusterIndex = new AtomicInteger(0);
+
+ /**
+ * Constructor
+ *
+ * @param workerName
+ * @param context
+ */
+ public MessageQueueZoneProducer(String workerName,
MessageQueueZoneSinkContext context) {
+ this.workerName = workerName;
+ this.context = context;
+ }
+
+ /**
+ * start
+ */
+ public void start() {
+ try {
+ LOG.info("start MessageQueueZoneProducer:{}", workerName);
+ this.reload();
+ this.setReloadTimer();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * close
+ */
+ public void close() {
+ try {
+ this.reloadTimer.cancel();
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ for (MessageQueueClusterProducer cluster : this.clusterList) {
+ cluster.stop();
+ }
+ }
+
+ /**
+ * setReloadTimer
+ */
+ private void setReloadTimer() {
+ reloadTimer = new Timer(true);
+ TimerTask task = new TimerTask() {
+
+ public void run() {
+ reload();
+ }
+ };
+ reloadTimer.schedule(task, new Date(System.currentTimeMillis() +
context.getReloadInterval()),
+ context.getReloadInterval());
+ }
+
+ /**
+ * reload
+ */
+ public void reload() {
+ try {
+ // stop deleted cluster
+ deletingClusterList.forEach(item -> {
+ item.stop();
+ });
+ deletingClusterList.clear();
+ // update cluster list
+ List<CacheClusterConfig> configList =
this.context.getCacheHolder().getConfigList();
+ List<MessageQueueClusterProducer> newClusterList = new
ArrayList<>(configList.size());
+ // prepare
+ Set<String> newClusterNames = new HashSet<>();
+ configList.forEach(item -> {
+ newClusterNames.add(item.getClusterName());
+ });
+ Set<String> oldClusterNames = new HashSet<>();
+ clusterList.forEach(item -> {
+ oldClusterNames.add(item.getCacheClusterName());
+ });
+ // add
+ for (CacheClusterConfig config : configList) {
+ if (!oldClusterNames.contains(config.getClusterName())) {
+ MessageQueueClusterProducer cluster = new
MessageQueueClusterProducer(workerName, config, context);
+ cluster.start();
+ newClusterList.add(cluster);
+ }
+ }
+ // remove
+ for (MessageQueueClusterProducer cluster : this.clusterList) {
+ if (newClusterNames.contains(cluster.getCacheClusterName())) {
+ newClusterList.add(cluster);
+ } else {
+ deletingClusterList.add(cluster);
+ }
+ }
+ this.clusterList = newClusterList;
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * send
+ *
+ * @param event
+ */
+ public boolean send(BatchPackProfile event) {
+ int currentIndex = clusterIndex.getAndIncrement();
+ if (currentIndex > MAX_INDEX) {
+ clusterIndex.set(0);
+ }
+ List<MessageQueueClusterProducer> currentClusterList =
this.clusterList;
+ int currentSize = currentClusterList.size();
+ int realIndex = currentIndex % currentSize;
+ MessageQueueClusterProducer clusterProducer =
currentClusterList.get(realIndex);
+ return clusterProducer.send(event);
+ }
+}
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
new file mode 100644
index 000000000..d261e7dcb
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSink.java
@@ -0,0 +1,167 @@
+/**
+ * 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.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+import org.apache.inlong.dataproxy.sink.common.SinkContext;
+import org.apache.inlong.dataproxy.utils.BufferQueue;
+import org.apache.inlong.sdk.commons.protocol.ProxyEvent;
+import org.apache.inlong.sdk.commons.protocol.ProxyPackEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * MessageQueueZoneSink
+ */
+public class MessageQueueZoneSink extends AbstractSink implements Configurable
{
+
+ public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueZoneSink.class);
+
+ private Context parentContext;
+ private MessageQueueZoneSinkContext context;
+ private List<MessageQueueZoneWorker> workers = new ArrayList<>();
+ // message group
+ private BatchPackManager dispatchManager;
+ private BufferQueue<BatchPackProfile> dispatchQueue;
+ // scheduled thread pool
+ // reload
+ // dispatch
+ private ScheduledExecutorService scheduledPool;
+
+ /**
+ * configure
+ *
+ * @param context
+ */
+ @Override
+ public void configure(Context context) {
+ LOG.info("start to configure:{}, context:{}.",
this.getClass().getSimpleName(), context.toString());
+ this.parentContext = context;
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ try {
+ this.dispatchQueue = SinkContext.createBufferQueue();
+ this.context = new MessageQueueZoneSinkContext(getName(),
parentContext, getChannel(), this.dispatchQueue);
+ if (getChannel() == null) {
+ LOG.error("channel is null");
+ }
+ this.context.start();
+ this.dispatchManager = new BatchPackManager(parentContext,
dispatchQueue);
+ this.scheduledPool = Executors.newScheduledThreadPool(2);
+ // dispatch
+ this.scheduledPool.scheduleWithFixedDelay(new Runnable() {
+
+ public void run() {
+ dispatchManager.setNeedOutputOvertimeData();
+ }
+ }, this.dispatchManager.getDispatchTimeout(),
this.dispatchManager.getDispatchTimeout(),
+ TimeUnit.MILLISECONDS);
+ // create worker
+ for (int i = 0; i < context.getMaxThreads(); i++) {
+ MessageQueueZoneWorker worker = new
MessageQueueZoneWorker(this.getName(), i, context);
+ worker.start();
+ this.workers.add(worker);
+ }
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ super.start();
+ }
+
+ /**
+ * stop
+ */
+ @Override
+ public void stop() {
+ for (MessageQueueZoneWorker worker : workers) {
+ try {
+ worker.close();
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ this.context.close();
+ super.stop();
+ }
+
+ /**
+ * process
+ *
+ * @return Status
+ * @throws EventDeliveryException
+ */
+ @Override
+ public Status process() throws EventDeliveryException {
+ this.dispatchManager.outputOvertimeData();
+ Channel channel = getChannel();
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ try {
+ Event event = channel.take();
+ // no data
+ if (event == null) {
+ tx.commit();
+ return Status.BACKOFF;
+ }
+ // ProxyEvent
+ if (event instanceof ProxyEvent) {
+ ProxyEvent proxyEvent = (ProxyEvent) event;
+ this.dispatchManager.addEvent(proxyEvent);
+ tx.commit();
+ return Status.READY;
+ }
+ // ProxyPackEvent
+ if (event instanceof ProxyPackEvent) {
+ ProxyPackEvent packEvent = (ProxyPackEvent) event;
+ this.dispatchManager.addPackEvent(packEvent);
+ tx.commit();
+ return Status.READY;
+ }
+ tx.commit();
+ this.context.addSendFailMetric();
+ return Status.READY;
+ } catch (Throwable t) {
+ LOG.error("Process event failed!" + this.getName(), t);
+ try {
+ tx.rollback();
+ } catch (Throwable e) {
+ LOG.error("Channel take transaction rollback exception:" +
getName(), e);
+ }
+ return Status.BACKOFF;
+ } finally {
+ tx.close();
+ }
+ }
+}
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
new file mode 100644
index 000000000..50a766a43
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneSinkContext.java
@@ -0,0 +1,265 @@
+/**
+ * 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.commons.lang3.StringUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.inlong.dataproxy.config.holder.CacheClusterConfigHolder;
+import org.apache.inlong.dataproxy.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.dataproxy.config.holder.IdTopicConfigHolder;
+import org.apache.inlong.dataproxy.metrics.DataProxyMetricItem;
+import org.apache.inlong.dataproxy.metrics.audit.AuditUtils;
+import org.apache.inlong.dataproxy.sink.common.SinkContext;
+import org.apache.inlong.dataproxy.utils.BufferQueue;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.INLONG_COMPRESSED_TYPE;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ *
+ * MessageQueueZoneSinkContext
+ */
+public class MessageQueueZoneSinkContext extends SinkContext {
+
+ public static final String KEY_NODE_ID = "nodeId";
+ public static final String PREFIX_PRODUCER = "producer.";
+ public static final String KEY_COMPRESS_TYPE = "compressType";
+
+ private final BufferQueue<BatchPackProfile> dispatchQueue;
+
+ private final String proxyClusterId;
+ private final String nodeId;
+ private final Context producerContext;
+ //
+ private final IdTopicConfigHolder idTopicHolder;
+ private final CacheClusterConfigHolder cacheHolder;
+ private final INLONG_COMPRESSED_TYPE compressType;
+
+ /**
+ * Constructor
+ */
+ public MessageQueueZoneSinkContext(String sinkName, Context context,
Channel channel,
+ BufferQueue<BatchPackProfile> dispatchQueue) {
+ super(sinkName, context, channel);
+ this.dispatchQueue = dispatchQueue;
+ // proxyClusterId
+ this.proxyClusterId =
CommonPropertiesHolder.getString(CommonPropertiesHolder.KEY_PROXY_CLUSTER_NAME);
+ // nodeId
+ this.nodeId = CommonPropertiesHolder.getString(KEY_NODE_ID,
"127.0.0.1");
+ // compressionType
+ String strCompressionType =
CommonPropertiesHolder.getString(KEY_COMPRESS_TYPE,
+ INLONG_COMPRESSED_TYPE.INLONG_SNAPPY.name());
+ this.compressType = INLONG_COMPRESSED_TYPE.valueOf(strCompressionType);
+ // producerContext
+ Map<String, String> producerParams =
context.getSubProperties(PREFIX_PRODUCER);
+ this.producerContext = new Context(producerParams);
+ // idTopicHolder
+ Context commonPropertiesContext = new
Context(CommonPropertiesHolder.get());
+ this.idTopicHolder = new IdTopicConfigHolder();
+ this.idTopicHolder.configure(commonPropertiesContext);
+ // cacheHolder
+ this.cacheHolder = new CacheClusterConfigHolder();
+ this.cacheHolder.configure(commonPropertiesContext);
+ }
+
+ /**
+ * start
+ */
+ public void start() {
+ super.start();
+ this.idTopicHolder.start();
+ this.cacheHolder.start();
+ }
+
+ /**
+ * close
+ */
+ public void close() {
+ super.close();
+ this.idTopicHolder.close();
+ this.cacheHolder.close();
+ }
+
+ /**
+ * get proxyClusterId
+ *
+ * @return the proxyClusterId
+ */
+ public String getProxyClusterId() {
+ return proxyClusterId;
+ }
+
+ /**
+ * get dispatchQueue
+ *
+ * @return the dispatchQueue
+ */
+ public BufferQueue<BatchPackProfile> getDispatchQueue() {
+ return dispatchQueue;
+ }
+
+ /**
+ * get producerContext
+ *
+ * @return the producerContext
+ */
+ public Context getProducerContext() {
+ return producerContext;
+ }
+
+ /**
+ * get idTopicHolder
+ *
+ * @return the idTopicHolder
+ */
+ public IdTopicConfigHolder getIdTopicHolder() {
+ return idTopicHolder;
+ }
+
+ /**
+ * get cacheHolder
+ *
+ * @return the cacheHolder
+ */
+ public CacheClusterConfigHolder getCacheHolder() {
+ return cacheHolder;
+ }
+
+ /**
+ * get compressType
+ *
+ * @return the compressType
+ */
+ public INLONG_COMPRESSED_TYPE getCompressType() {
+ return compressType;
+ }
+
+ /**
+ * get nodeId
+ *
+ * @return the nodeId
+ */
+ public String getNodeId() {
+ return nodeId;
+ }
+
+ /**
+ * addSendResultMetric
+ */
+ public void addSendResultMetric(BatchPackProfile currentRecord, String
topic, boolean result, long sendTime) {
+ 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);
+ dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, topic);
+ final long currentTime = System.currentTimeMillis();
+ currentRecord.getEvents().forEach(event -> {
+ long msgTime = event.getMsgTime();
+ long auditFormatTime = msgTime - msgTime %
CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(DataProxyMetricItem.KEY_MESSAGE_TIME,
String.valueOf(auditFormatTime));
+ DataProxyMetricItem metricItem =
this.getMetricItemSet().findMetricItem(dimensions);
+ if (result) {
+ metricItem.sendSuccessCount.addAndGet(1);
+ metricItem.sendSuccessSize.addAndGet(event.getBody().length);
+ if (sendTime > 0) {
+ long sinkDuration = currentTime - sendTime;
+ long nodeDuration = currentTime - event.getSourceTime();
+ long wholeDuration = currentTime - msgTime;
+ metricItem.sinkDuration.addAndGet(sinkDuration);
+ metricItem.nodeDuration.addAndGet(nodeDuration);
+ metricItem.wholeDuration.addAndGet(wholeDuration);
+ }
+ AuditUtils.add(AuditUtils.AUDIT_ID_DATAPROXY_SEND_SUCCESS,
event);
+ } else {
+ metricItem.sendFailCount.addAndGet(1);
+ metricItem.sendFailSize.addAndGet(event.getBody().length);
+ }
+ });
+ }
+
+ /**
+ * addSendMetric
+ */
+ public void addSendMetric(BatchPackProfile currentRecord, String topic) {
+ 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);
+ dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, topic);
+ long msgTime = currentRecord.getDispatchTime();
+ long auditFormatTime = msgTime - msgTime %
CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(DataProxyMetricItem.KEY_MESSAGE_TIME,
String.valueOf(auditFormatTime));
+ DataProxyMetricItem metricItem =
this.getMetricItemSet().findMetricItem(dimensions);
+ long count = currentRecord.getCount();
+ long size = currentRecord.getSize();
+ metricItem.sendCount.addAndGet(count);
+ metricItem.sendSize.addAndGet(size);
+ }
+
+ /**
+ * addReadFailMetric
+ */
+ public void addSendFailMetric() {
+ 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, "-");
+ dimensions.put(DataProxyMetricItem.KEY_INLONG_GROUP_ID, "-");
+ dimensions.put(DataProxyMetricItem.KEY_INLONG_STREAM_ID, "-");
+ dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getSinkName());
+ dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, "-");
+ long msgTime = System.currentTimeMillis();
+ long auditFormatTime = msgTime - msgTime %
CommonPropertiesHolder.getAuditFormatInterval();
+ dimensions.put(DataProxyMetricItem.KEY_MESSAGE_TIME,
String.valueOf(auditFormatTime));
+ DataProxyMetricItem metricItem =
this.getMetricItemSet().findMetricItem(dimensions);
+ metricItem.sendFailCount.incrementAndGet();
+ metricItem.sendFailSize.incrementAndGet();
+ }
+
+ /**
+ * fillInlongId
+ */
+ public static void fillInlongId(BatchPackProfile currentRecord,
Map<String, String> dimensions) {
+ String inlongGroupId = currentRecord.getInlongGroupId();
+ inlongGroupId = (StringUtils.isBlank(inlongGroupId)) ? "-" :
inlongGroupId;
+ String inlongStreamId = currentRecord.getInlongStreamId();
+ inlongStreamId = (StringUtils.isBlank(inlongStreamId)) ? "-" :
inlongStreamId;
+ dimensions.put(DataProxyMetricItem.KEY_INLONG_GROUP_ID, inlongGroupId);
+ dimensions.put(DataProxyMetricItem.KEY_INLONG_STREAM_ID,
inlongStreamId);
+ }
+
+ /**
+ * processSendFail
+ */
+ public void processSendFail(BatchPackProfile currentRecord, String topic,
long sendTime) {
+ if (currentRecord.isResend()) {
+ dispatchQueue.offer(currentRecord);
+ this.addSendResultMetric(currentRecord, topic, false, sendTime);
+ } else {
+ currentRecord.fail();
+ }
+ }
+}
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
new file mode 100644
index 000000000..e6b876799
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/MessageQueueZoneWorker.java
@@ -0,0 +1,108 @@
+/**
+ * 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.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * MessageQueueZoneWorker
+ */
+public class MessageQueueZoneWorker extends Thread {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(MessageQueueZoneWorker.class);
+
+ private final String workerName;
+ private final MessageQueueZoneSinkContext context;
+
+ private MessageQueueZoneProducer zoneProducer;
+ private LifecycleState status;
+
+ /**
+ * Constructor
+ *
+ * @param sinkName
+ * @param workerIndex
+ * @param context
+ */
+ public MessageQueueZoneWorker(String sinkName, int workerIndex,
MessageQueueZoneSinkContext context) {
+ super();
+ this.workerName = sinkName + "-worker-" + workerIndex;
+ this.context = context;
+ this.zoneProducer = new MessageQueueZoneProducer(workerName,
this.context);
+ this.status = LifecycleState.IDLE;
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ this.zoneProducer.start();
+ this.status = LifecycleState.START;
+ super.start();
+ }
+
+ /**
+ *
+ * close
+ */
+ public void close() {
+ // close all producers
+ this.zoneProducer.close();
+ this.status = LifecycleState.STOP;
+ }
+
+ /**
+ * run
+ */
+ @Override
+ public void run() {
+ LOG.info(String.format("start MessageQueueZoneWorker:%s",
this.workerName));
+ while (status != LifecycleState.STOP) {
+ BatchPackProfile event = null;
+ try {
+ event = context.getDispatchQueue().pollRecord();
+ if (event == null) {
+ this.sleepOneInterval();
+ continue;
+ }
+ // send
+ this.zoneProducer.send(event);
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ if (event != null) {
+ context.getDispatchQueue().offer(event);
+ }
+ this.sleepOneInterval();
+ }
+ }
+ }
+
+ /**
+ * sleepOneInterval
+ */
+ private void sleepOneInterval() {
+ try {
+ Thread.sleep(context.getProcessInterval());
+ } catch (InterruptedException e1) {
+ LOG.error(e1.getMessage(), e1);
+ }
+ }
+}
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
new file mode 100644
index 000000000..8322bd786
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/OrderBatchPackProfileV0.java
@@ -0,0 +1,105 @@
+/**
+ * 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.commons.lang3.math.NumberUtils;
+import org.apache.inlong.dataproxy.base.OrderEvent;
+import org.apache.inlong.dataproxy.consts.AttributeConstants;
+import org.apache.inlong.dataproxy.source.MsgType;
+import org.apache.inlong.dataproxy.utils.MessageUtils;
+import org.apache.inlong.sdk.commons.protocol.InlongId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import io.netty.buffer.ByteBuf;
+
+/**
+ * SimpleBatchPackProfileV0
+ *
+ */
+public class OrderBatchPackProfileV0 extends BatchPackProfile {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(OrderBatchPackProfileV0.class);
+
+ private OrderEvent 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(OrderEvent 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 OrderEvent 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.getCtx() != null &&
orderProfile.getCtx().channel().isActive()) {
+ orderProfile.getCtx().channel().eventLoop().execute(() -> {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("order message rsp: seqId = {}, inlongGroupId =
{}, inlongStreamId = {}", sequenceId,
+ this.getInlongGroupId(), this.getInlongStreamId());
+ }
+ ByteBuf binBuffer = MessageUtils.getResponsePackage("",
MsgType.MSG_BIN_MULTI_BODY, sequenceId);
+ orderProfile.getCtx().writeAndFlush(binBuffer);
+ });
+ }
+ }
+}
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
new file mode 100644
index 000000000..8b2c82d3b
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimpleBatchPackProfileV0.java
@@ -0,0 +1,73 @@
+/**
+ * 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.commons.lang3.math.NumberUtils;
+import org.apache.flume.Event;
+import org.apache.inlong.dataproxy.consts.AttributeConstants;
+import org.apache.inlong.sdk.commons.protocol.InlongId;
+
+import java.util.Map;
+
+/**
+ * SimpleBatchPackProfileV0
+ *
+ */
+public class SimpleBatchPackProfileV0 extends BatchPackProfile {
+
+ private Event simpleProfile;
+
+ /**
+ * 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;
+ return profile;
+ }
+
+ /**
+ * get simpleProfile
+ * @return the simpleProfile
+ */
+ public Event getSimpleProfile() {
+ return simpleProfile;
+ }
+}
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
new file mode 100644
index 000000000..7d030393d
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/kafka/KafkaHandler.java
@@ -0,0 +1,267 @@
+/**
+ * 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.kafka;
+
+import org.apache.flume.Context;
+import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+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.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * KafkaHandler
+ *
+ */
+public class KafkaHandler implements MessageQueueHandler {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(KafkaHandler.class);
+
+ private CacheClusterConfig config;
+ private MessageQueueZoneSinkContext sinkContext;
+
+ // kafka producer
+ private KafkaProducer<String, byte[]> producer;
+ private EventHandler handler;
+
+ /**
+ * init
+ * @param config
+ * @param sinkContext
+ */
+ @Override
+ public void init(CacheClusterConfig config, MessageQueueZoneSinkContext
sinkContext) {
+ this.config = config;
+ this.sinkContext = sinkContext;
+ this.handler = this.sinkContext.createEventHandler();
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ // create kafka producer
+ try {
+ // prepare configuration
+ Properties props = new Properties();
+ Context context = this.sinkContext.getProducerContext();
+ props.putAll(context.getParameters());
+ props.putAll(config.getParams());
+ LOG.info("try to create kafka client:{}", props);
+ producer = new KafkaProducer<>(props, new StringSerializer(), new
ByteArraySerializer());
+ LOG.info("create new producer success:{}", producer);
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * stop
+ */
+ @Override
+ public void stop() {
+ // kafka producer
+ this.producer.close();
+ }
+
+ /**
+ * send
+ * @param event
+ * @return
+ */
+ @Override
+ public boolean send(BatchPackProfile event) {
+ try {
+ // idConfig
+ IdTopicConfig idConfig =
sinkContext.getIdTopicHolder().getIdConfig(event.getUid());
+ if (idConfig == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ String topic = idConfig.getTopicName();
+ if (topic == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ // metric
+ sinkContext.addSendMetric(event, topic);
+ // create producer failed
+ if (producer == null) {
+ sinkContext.processSendFail(event, topic, 0);
+ return false;
+ }
+ // send
+ if (event instanceof SimpleBatchPackProfileV0) {
+ this.sendSimpleProfileV0((SimpleBatchPackProfileV0) event,
idConfig, topic);
+ } else if (event instanceof OrderBatchPackProfileV0) {
+ this.sendOrderProfileV0((OrderBatchPackProfileV0) event,
idConfig, topic);
+ } else {
+ this.sendProfileV1(event, idConfig, topic);
+ }
+ return true;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(event, event.getUid(), 0);
+ return false;
+ }
+ }
+
+ /**
+ * sendProfileV1
+ */
+ private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
+ String topic) throws Exception {
+ // headers
+ Map<String, String> headers = this.handler.parseHeader(idConfig,
event, sinkContext.getNodeId(),
+ sinkContext.getCompressType());
+ // compress
+ byte[] bodyBytes = this.handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ // 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) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ if (event.isResend()) {
+ sinkContext.processSendFail(event, topic, sendTime);
+ } else {
+ event.fail();
+ }
+ } else {
+ sinkContext.addSendResultMetric(event, topic, true,
sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ }
+ }
+ };
+ producer.send(producerRecord, callback);
+ }
+
+ /**
+ * sendSimpleProfileV0
+ */
+ private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ String topic) throws Exception {
+ // headers
+ Map<String, String> headers = event.getSimpleProfile().getHeaders();
+ // compress
+ byte[] bodyBytes = event.getSimpleProfile().getBody();
+ // 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) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ if (event.isResend()) {
+ sinkContext.processSendFail(event, topic, sendTime);
+ } else {
+ event.fail();
+ }
+ } else {
+ sinkContext.addSendResultMetric(event, 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();
+ // 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) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ if (event.isResend()) {
+ sinkContext.processSendFail(event, topic, sendTime);
+ } else {
+ event.fail();
+ }
+ } else {
+ sinkContext.addSendResultMetric(event, topic, true,
sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ }
+ }
+ };
+ producer.send(producerRecord, callback);
+ }
+}
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
new file mode 100644
index 000000000..cfc7bb0df
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/pulsar/PulsarHandler.java
@@ -0,0 +1,365 @@
+/**
+ * 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.pulsar;
+
+import org.apache.flume.Context;
+import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+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.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.api.CompressionType;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerAccessMode;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SizeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.security.SecureRandom;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * PulsarHandler
+ */
+public class PulsarHandler implements MessageQueueHandler {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(PulsarHandler.class);
+
+ public static final String KEY_TENANT = "tenant";
+ public static final String KEY_NAMESPACE = "namespace";
+
+ public static final String KEY_SERVICE_URL = "serviceUrl";
+ public static final String KEY_AUTHENTICATION = "authentication";
+
+ public static final String KEY_ENABLEBATCHING = "enableBatching";
+ public static final String KEY_BATCHINGMAXBYTES = "batchingMaxBytes";
+ public static final String KEY_BATCHINGMAXMESSAGES = "batchingMaxMessages";
+ public static final String KEY_BATCHINGMAXPUBLISHDELAY =
"batchingMaxPublishDelay";
+ public static final String KEY_MAXPENDINGMESSAGES = "maxPendingMessages";
+ public static final String KEY_MAXPENDINGMESSAGESACROSSPARTITIONS =
"maxPendingMessagesAcrossPartitions";
+ public static final String KEY_SENDTIMEOUT = "sendTimeout";
+ public static final String KEY_COMPRESSIONTYPE = "compressionType";
+ public static final String KEY_BLOCKIFQUEUEFULL = "blockIfQueueFull";
+ public static final String
KEY_ROUNDROBINROUTERBATCHINGPARTITIONSWITCHFREQUENCY = "roundRobinRouter"
+ + "BatchingPartitionSwitchFrequency";
+
+ public static final String KEY_IOTHREADS = "ioThreads";
+ public static final String KEY_MEMORYLIMIT = "memoryLimit";
+ public static final String KEY_CONNECTIONSPERBROKER =
"connectionsPerBroker";
+
+ private CacheClusterConfig config;
+ private MessageQueueZoneSinkContext sinkContext;
+
+ private String tenant;
+ private String namespace;
+ private EventHandler handler;
+
+ /**
+ * pulsar client
+ */
+ private PulsarClient client;
+ private ProducerBuilder<byte[]> baseBuilder;
+
+ private Map<String, Producer<byte[]>> producerMap = new
ConcurrentHashMap<>();
+
+ /**
+ * init
+ * @param config
+ * @param sinkContext
+ */
+ public void init(CacheClusterConfig config, MessageQueueZoneSinkContext
sinkContext) {
+ this.config = config;
+ this.sinkContext = sinkContext;
+ this.tenant = config.getParams().get(KEY_TENANT);
+ this.namespace = config.getParams().get(KEY_NAMESPACE);
+ this.handler = this.sinkContext.createEventHandler();
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ // create pulsar client
+ try {
+ String serviceUrl = config.getParams().get(KEY_SERVICE_URL);
+ String authentication = config.getParams().get(KEY_AUTHENTICATION);
+ Context context = sinkContext.getProducerContext();
+ this.client = PulsarClient.builder()
+ .serviceUrl(serviceUrl)
+
.authentication(AuthenticationFactory.token(authentication))
+ .ioThreads(context.getInteger(KEY_IOTHREADS, 1))
+ .memoryLimit(context.getLong(KEY_MEMORYLIMIT,
1073741824L), SizeUnit.BYTES)
+
.connectionsPerBroker(context.getInteger(KEY_CONNECTIONSPERBROKER, 10))
+ .build();
+ this.baseBuilder = client.newProducer();
+ // Map<String, Object> builderConf = new HashMap<>();
+ // builderConf.putAll(context.getParameters());
+ this.baseBuilder
+ .sendTimeout(context.getInteger(KEY_SENDTIMEOUT, 0),
TimeUnit.MILLISECONDS)
+
.maxPendingMessages(context.getInteger(KEY_MAXPENDINGMESSAGES, 500))
+ .maxPendingMessagesAcrossPartitions(
+
context.getInteger(KEY_MAXPENDINGMESSAGESACROSSPARTITIONS, 60000));
+ this.baseBuilder
+
.batchingMaxMessages(context.getInteger(KEY_BATCHINGMAXMESSAGES, 500))
+
.batchingMaxPublishDelay(context.getInteger(KEY_BATCHINGMAXPUBLISHDELAY, 100),
+ TimeUnit.MILLISECONDS)
+ .batchingMaxBytes(context.getInteger(KEY_BATCHINGMAXBYTES,
131072));
+ this.baseBuilder
+ .accessMode(ProducerAccessMode.Shared)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .blockIfQueueFull(context.getBoolean(KEY_BLOCKIFQUEUEFULL,
true));
+ this.baseBuilder
+ .roundRobinRouterBatchingPartitionSwitchFrequency(
+
context.getInteger(KEY_ROUNDROBINROUTERBATCHINGPARTITIONSWITCHFREQUENCY, 60))
+ .enableBatching(context.getBoolean(KEY_ENABLEBATCHING,
true))
+ .compressionType(this.getPulsarCompressionType());
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * stop
+ */
+ @Override
+ public void stop() {
+ for (Entry<String, Producer<byte[]>> entry :
this.producerMap.entrySet()) {
+ try {
+ entry.getValue().close();
+ } catch (PulsarClientException e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ try {
+ this.client.close();
+ } catch (PulsarClientException e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * send
+ * @param event
+ * @return
+ */
+ @Override
+ public boolean send(BatchPackProfile event) {
+ try {
+ // idConfig
+ IdTopicConfig idConfig =
sinkContext.getIdTopicHolder().getIdConfig(event.getUid());
+ if (idConfig == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ String baseTopic = idConfig.getTopicName();
+ if (baseTopic == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ // topic
+ String producerTopic = this.getProducerTopic(baseTopic);
+ if (producerTopic == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.fail();
+ return false;
+ }
+ // metric
+ sinkContext.addSendMetric(event, producerTopic);
+ // get producer
+ Producer<byte[]> producer = this.producerMap.get(producerTopic);
+ if (producer == null) {
+ try {
+ LOG.info("try to new a object for topic " + producerTopic);
+ SecureRandom secureRandom = new SecureRandom(
+ (producerTopic +
System.currentTimeMillis()).getBytes());
+ String producerName = producerTopic + "-" +
secureRandom.nextLong();
+ producer = baseBuilder.clone().topic(producerTopic)
+ .producerName(producerName)
+ .create();
+ LOG.info("create new producer success:{}",
producer.getProducerName());
+ Producer<byte[]> oldProducer =
this.producerMap.putIfAbsent(producerTopic, producer);
+ if (oldProducer != null) {
+ producer.close();
+ LOG.info("close producer success:{}",
producer.getProducerName());
+ producer = oldProducer;
+ }
+ } catch (Throwable ex) {
+ LOG.error("create new producer failed", ex);
+ }
+ }
+ // create producer failed
+ if (producer == null) {
+ sinkContext.processSendFail(event, producerTopic, 0);
+ 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);
+ } else {
+ this.sendProfileV1(event, idConfig, producer, producerTopic);
+ }
+ return true;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(event, event.getUid(), 0);
+ return false;
+ }
+ }
+
+ /**
+ * getProducerTopic
+ */
+ private String getProducerTopic(String baseTopic) {
+ StringBuilder builder = new StringBuilder();
+ if (tenant != null) {
+ builder.append(tenant).append("/");
+ }
+ if (namespace != null) {
+ builder.append(namespace).append("/");
+ }
+ builder.append(baseTopic);
+ return builder.toString();
+ }
+
+ /**
+ * getPulsarCompressionType
+ *
+ * @return CompressionType
+ */
+ private CompressionType getPulsarCompressionType() {
+ Context context = sinkContext.getProducerContext();
+ String type = context.getString(KEY_COMPRESSIONTYPE,
CompressionType.SNAPPY.name());
+ switch (type) {
+ case "LZ4" :
+ return CompressionType.LZ4;
+ case "NONE" :
+ return CompressionType.NONE;
+ case "ZLIB" :
+ return CompressionType.ZLIB;
+ case "ZSTD" :
+ return CompressionType.ZSTD;
+ case "SNAPPY" :
+ return CompressionType.SNAPPY;
+ default :
+ return CompressionType.NONE;
+ }
+ }
+
+ /**
+ * sendProfileV1
+ */
+ private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
Producer<byte[]> producer,
+ String producerTopic) throws Exception {
+ // headers
+ Map<String, String> headers = this.handler.parseHeader(idConfig,
event, sinkContext.getNodeId(),
+ sinkContext.getCompressType());
+ // compress
+ byte[] bodyBytes = this.handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ // sendAsync
+ long sendTime = System.currentTimeMillis();
+ CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
+ .value(bodyBytes).sendAsync();
+ // callback
+ future.whenCompleteAsync((msgId, ex) -> {
+ if (ex != null) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, producerTopic, sendTime);
+ } else {
+ sinkContext.addSendResultMetric(event, producerTopic, true,
sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ }
+ });
+ }
+
+ /**
+ * sendSimpleProfileV0
+ */
+ private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ Producer<byte[]> producer,
+ String producerTopic) throws Exception {
+ // headers
+ Map<String, String> headers = event.getSimpleProfile().getHeaders();
+ // compress
+ byte[] bodyBytes = event.getSimpleProfile().getBody();
+ // sendAsync
+ long sendTime = System.currentTimeMillis();
+ CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
+ .value(bodyBytes).sendAsync();
+ // callback
+ future.whenCompleteAsync((msgId, ex) -> {
+ if (ex != null) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, producerTopic, sendTime);
+ } else {
+ sinkContext.addSendResultMetric(event, 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();
+ // sendAsync
+ long sendTime = System.currentTimeMillis();
+ CompletableFuture<MessageId> future =
producer.newMessage().properties(headers)
+ .value(bodyBytes).sendAsync();
+ // callback
+ future.whenCompleteAsync((msgId, ex) -> {
+ if (ex != null) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, producerTopic, sendTime);
+ } else {
+ sinkContext.addSendResultMetric(event, producerTopic, true,
sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ event.ackOrder();
+ }
+ });
+ }
+}
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
new file mode 100644
index 000000000..1b1471dbb
--- /dev/null
+++
b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/tube/TubeHandler.java
@@ -0,0 +1,307 @@
+/**
+ * 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.tube;
+
+import org.apache.flume.Context;
+import org.apache.inlong.dataproxy.config.pojo.CacheClusterConfig;
+import org.apache.inlong.dataproxy.config.pojo.IdTopicConfig;
+import org.apache.inlong.dataproxy.consts.ConfigConstants;
+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.tubemq.client.config.TubeClientConfig;
+import org.apache.inlong.tubemq.client.exception.TubeClientException;
+import org.apache.inlong.tubemq.client.factory.TubeMultiSessionFactory;
+import org.apache.inlong.tubemq.client.producer.MessageProducer;
+import org.apache.inlong.tubemq.client.producer.MessageSentCallback;
+import org.apache.inlong.tubemq.client.producer.MessageSentResult;
+import org.apache.inlong.tubemq.corebase.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * TubeHandler
+ */
+public class TubeHandler implements MessageQueueHandler {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(TubeHandler.class);
+ private static String MASTER_HOST_PORT_LIST = "master-host-port-list";
+
+ private CacheClusterConfig config;
+ private MessageQueueZoneSinkContext sinkContext;
+
+ // parameter
+ private String masterHostAndPortList;
+ private long linkMaxAllowedDelayedMsgCount;
+ private long sessionWarnDelayedMsgCount;
+ private long sessionMaxAllowedDelayedMsgCount;
+ private long nettyWriteBufferHighWaterMark;
+ // tube producer
+ private TubeMultiSessionFactory sessionFactory;
+ private MessageProducer producer;
+ private Set<String> topicSet = new HashSet<>();
+ private EventHandler handler;
+
+ /**
+ * init
+ * @param config
+ * @param sinkContext
+ */
+ @Override
+ public void init(CacheClusterConfig config, MessageQueueZoneSinkContext
sinkContext) {
+ this.config = config;
+ this.sinkContext = sinkContext;
+ this.handler = this.sinkContext.createEventHandler();
+ }
+
+ /**
+ * start
+ */
+ @Override
+ public void start() {
+ // create tube producer
+ try {
+ // prepare configuration
+ TubeClientConfig conf = initTubeConfig();
+ LOG.info("try to create producer:{}", conf.toJsonString());
+ this.sessionFactory = new TubeMultiSessionFactory(conf);
+ this.producer = sessionFactory.createProducer();
+ LOG.info("create new producer success:{}", producer);
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * initTubeConfig
+ * @return
+ *
+ * @throws Exception
+ */
+ private TubeClientConfig initTubeConfig() throws Exception {
+ // get parameter
+ Context context = sinkContext.getProducerContext();
+ Context configContext = new Context(context.getParameters());
+ configContext.putAll(this.config.getParams());
+ masterHostAndPortList = configContext.getString(MASTER_HOST_PORT_LIST);
+ linkMaxAllowedDelayedMsgCount =
configContext.getLong(ConfigConstants.LINK_MAX_ALLOWED_DELAYED_MSG_COUNT,
+ 80000L);
+ sessionWarnDelayedMsgCount =
configContext.getLong(ConfigConstants.SESSION_WARN_DELAYED_MSG_COUNT,
+ 2000000L);
+ sessionMaxAllowedDelayedMsgCount = configContext.getLong(
+ ConfigConstants.SESSION_MAX_ALLOWED_DELAYED_MSG_COUNT,
+ 4000000L);
+ nettyWriteBufferHighWaterMark =
configContext.getLong(ConfigConstants.NETTY_WRITE_BUFFER_HIGH_WATER_MARK,
+ 15 * 1024 * 1024L);
+ // config
+ final TubeClientConfig tubeClientConfig = new
TubeClientConfig(this.masterHostAndPortList);
+
tubeClientConfig.setLinkMaxAllowedDelayedMsgCount(linkMaxAllowedDelayedMsgCount);
+
tubeClientConfig.setSessionWarnDelayedMsgCount(sessionWarnDelayedMsgCount);
+
tubeClientConfig.setSessionMaxAllowedDelayedMsgCount(sessionMaxAllowedDelayedMsgCount);
+
tubeClientConfig.setNettyWriteBufferHighWaterMark(nettyWriteBufferHighWaterMark);
+ tubeClientConfig.setHeartbeatPeriodMs(15000L);
+ tubeClientConfig.setRpcTimeoutMs(20000L);
+
+ return tubeClientConfig;
+ }
+
+ /**
+ * stop
+ */
+ @Override
+ public void stop() {
+ // producer
+ if (this.producer != null) {
+ try {
+ this.producer.shutdown();
+ } catch (Throwable e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ if (this.sessionFactory != null) {
+ try {
+ this.sessionFactory.shutdown();
+ } catch (TubeClientException e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ }
+
+ /**
+ * send
+ */
+ public boolean send(BatchPackProfile event) {
+ try {
+ // idConfig
+ IdTopicConfig idConfig =
sinkContext.getIdTopicHolder().getIdConfig(event.getUid());
+ if (idConfig == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ String topic = idConfig.getTopicName();
+ if (topic == null) {
+ sinkContext.addSendResultMetric(event, event.getUid(), false,
0);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ return false;
+ }
+ // metric
+ sinkContext.addSendMetric(event, topic);
+ // publish
+ if (!this.topicSet.contains(topic)) {
+ this.producer.publish(topic);
+ this.topicSet.add(topic);
+ }
+ // create producer failed
+ if (producer == null) {
+ sinkContext.processSendFail(event, topic, 0);
+ return false;
+ }
+ // send
+ if (event instanceof SimpleBatchPackProfileV0) {
+ this.sendSimpleProfileV0((SimpleBatchPackProfileV0) event,
idConfig, topic);
+ } else if (event instanceof OrderBatchPackProfileV0) {
+ this.sendOrderProfileV0((OrderBatchPackProfileV0) event,
idConfig, topic);
+ } else {
+ this.sendProfileV1(event, idConfig, topic);
+ }
+ return true;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ sinkContext.processSendFail(event, event.getUid(), 0);
+ return false;
+ }
+ }
+
+ /**
+ * sendProfileV1
+ */
+ private void sendProfileV1(BatchPackProfile event, IdTopicConfig idConfig,
+ String topic) throws Exception {
+ // headers
+ Map<String, String> headers = this.handler.parseHeader(idConfig,
event, sinkContext.getNodeId(),
+ sinkContext.getCompressType());
+ // compress
+ byte[] bodyBytes = this.handler.parseBody(idConfig, event,
sinkContext.getCompressType());
+ // sendAsync
+ Message message = new Message(topic, bodyBytes);
+ // add headers
+ headers.forEach((key, value) -> {
+ message.setAttrKeyVal(key, value);
+ });
+ // callback
+ long sendTime = System.currentTimeMillis();
+ MessageSentCallback callback = new MessageSentCallback() {
+
+ @Override
+ public void onMessageSent(MessageSentResult result) {
+ sinkContext.addSendResultMetric(event, topic, true, sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ }
+
+ @Override
+ public void onException(Throwable ex) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, topic, sendTime);
+ }
+ };
+ producer.sendMessage(message, callback);
+ }
+
+ /**
+ * sendSimpleProfileV0
+ */
+ private void sendSimpleProfileV0(SimpleBatchPackProfileV0 event,
IdTopicConfig idConfig,
+ String topic) throws Exception {
+ // headers
+ Map<String, String> headers = event.getSimpleProfile().getHeaders();
+ // compress
+ byte[] bodyBytes = event.getSimpleProfile().getBody();
+ // sendAsync
+ Message message = new Message(topic, bodyBytes);
+ // add headers
+ headers.forEach((key, value) -> {
+ message.setAttrKeyVal(key, value);
+ });
+ // callback
+ long sendTime = System.currentTimeMillis();
+ MessageSentCallback callback = new MessageSentCallback() {
+
+ @Override
+ public void onMessageSent(MessageSentResult result) {
+ sinkContext.addSendResultMetric(event, topic, true, sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ }
+
+ @Override
+ public void onException(Throwable ex) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, topic, sendTime);
+ }
+ };
+ 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();
+ // sendAsync
+ Message message = new Message(topic, bodyBytes);
+ // add headers
+ headers.forEach((key, value) -> {
+ message.setAttrKeyVal(key, value);
+ });
+ // callback
+ long sendTime = System.currentTimeMillis();
+ MessageSentCallback callback = new MessageSentCallback() {
+
+ @Override
+ public void onMessageSent(MessageSentResult result) {
+ sinkContext.addSendResultMetric(event, topic, true, sendTime);
+ sinkContext.getDispatchQueue().release(event.getSize());
+ event.ack();
+ event.ackOrder();
+ }
+
+ @Override
+ public void onException(Throwable ex) {
+ LOG.error("Send fail:{}", ex.getMessage());
+ LOG.error(ex.getMessage(), ex);
+ sinkContext.processSendFail(event, topic, sendTime);
+ }
+ };
+ producer.sendMessage(message, callback);
+ }
+}
