You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2022/03/25 11:25:48 UTC
[incubator-inlong] branch master updated: [INLONG-3352][DataProxy] Fix DataProxy keeps trying to send messages that were sent failed (#3357)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 5fac2f2 [INLONG-3352][DataProxy] Fix DataProxy keeps trying to send messages that were sent failed (#3357)
5fac2f2 is described below
commit 5fac2f2cfec8da870005822786877ad6dd9e285c
Author: baomingyu <ba...@163.com>
AuthorDate: Fri Mar 25 19:25:44 2022 +0800
[INLONG-3352][DataProxy] Fix DataProxy keeps trying to send messages that were sent failed (#3357)
---
.../config/pojo/ThirdPartyClusterConfig.java | 9 +-
.../apache/inlong/dataproxy/sink/EventStat.java | 27 +-
.../apache/inlong/dataproxy/sink/PulsarSink.java | 440 ++++++++-------------
.../dataproxy/sink/pulsar/PulsarClientService.java | 63 +--
.../inlong/dataproxy/sink/pulsar/SinkTask.java | 270 +++++++++++++
.../dataproxy/source/ServerMessageHandler.java | 38 +-
.../inlong/dataproxy/utils/MessageUtils.java | 25 ++
.../src/main/resources/dataproxy-pulsar.conf | 110 ++++++
.../dataproxy-source/src/main/resources/flume.conf | 162 --------
.../dataproxy-source/src/main/resources/log4j2.xml | 132 +++++++
10 files changed, 762 insertions(+), 514 deletions(-)
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
index db34271..27639e6 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/pojo/ThirdPartyClusterConfig.java
@@ -55,8 +55,8 @@ public class ThirdPartyClusterConfig extends Context {
private static final String MAX_BATCHING_PUBLISH_DELAY_MILLIS =
"max_batching_publish_delay_millis";
private static final String EVENT_QUEUE_SIZE = "event_queue_size";
-
private static final String BAD_EVENT_QUEUE_SIZE = "bad_event_queue_size";
+ private static final String MAX_RETRY_SEND_TIMES = "max_retry_send_times";
private static final String SLA_METRIC_SINK = "sla_metric_sink";
@@ -87,7 +87,6 @@ public class ThirdPartyClusterConfig extends Context {
private static final String STAT_INTERVAL_SEC = "stat_interval_sec";
private static final String LOG_EVERY_N_EVENTS = "log_every_n_events";
private static final String CLIENT_ID_CACHE = "client_id_cache";
- private static final String RETRY_CNT = "retry_currentSuccSendedCnt";
public static String PULSAR_DEFAULT_AUTH_TYPE = "token";
private static final int DEFAULT_CLIENT_TIMEOUT_SECOND = 30;
@@ -101,7 +100,6 @@ public class ThirdPartyClusterConfig extends Context {
private static final int DEFAULT_MAX_BATCHING_MESSAGES = 1000;
private static final int DEFAULT_MAX_BATCHING_BYTES = 128 * 1024;
private static final long DEFAULT_MAX_BATCHING_PUBLISH_DELAY_MILLIS = 1L;
- private static final int DEFAULT_RETRY_CNT = -1;
private static final int DEFAULT_LOG_EVERY_N_EVENTS = 100000;
private static final int DEFAULT_STAT_INTERVAL_SEC = 60;
private static final int DEFAULT_THREAD_NUM = 4;
@@ -112,6 +110,7 @@ public class ThirdPartyClusterConfig extends Context {
private static final int DEFAULT_PULSAR_IO_THREADS = Math.max(1, SystemPropertyUtil
.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
private static final int DEFAULT_CONNECTIONS_PRE_BROKER = 1;
+ private static final int DEFAULT_MAX_RETRY_SEND_TIMES = 16;
private static final boolean DEFAULT_SLA_METRIC_SINK = false;
private static final String DEFAULT_LOG_TOPIC = "teg_manager";
@@ -225,8 +224,8 @@ public class ThirdPartyClusterConfig extends Context {
return getLong(RETRY_INTERVAL_WHEN_SEND_ERROR_MILL, DEFAULT_RETRY_INTERVAL_WHEN_SEND_ERROR_MILL);
}
- public int getRetryCnt() {
- return getInteger(RETRY_CNT, DEFAULT_RETRY_CNT);
+ public int getMaxRetryCnt() {
+ return getInteger(MAX_RETRY_SEND_TIMES, DEFAULT_MAX_RETRY_SEND_TIMES);
}
public int getStatIntervalSec() {
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/EventStat.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/EventStat.java
index 0c67153..5c23111 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/EventStat.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/EventStat.java
@@ -17,20 +17,29 @@
package org.apache.inlong.dataproxy.sink;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.flume.Event;
+import org.apache.inlong.dataproxy.base.OrderEvent;
+import org.apache.inlong.dataproxy.utils.MessageUtils;
public class EventStat {
+
+ private static long RETRY_INTERVAL_MS = 1 * 1000L;
private Event event;
- private int myRetryCnt;
+ private AtomicInteger myRetryCnt = new AtomicInteger(0);
+ private boolean isOrderMessage = false;
public EventStat(Event event) {
this.event = event;
- this.myRetryCnt = 0;
+ this.isOrderMessage = MessageUtils
+ .isSyncSendForOrder(event) && (event instanceof OrderEvent);
}
public EventStat(Event event, int retryCnt) {
this.event = event;
- this.myRetryCnt = retryCnt;
+ this.myRetryCnt.set(retryCnt);
+ this.isOrderMessage = MessageUtils
+ .isSyncSendForOrder(event) && (event instanceof OrderEvent);
}
public Event getEvent() {
@@ -42,15 +51,19 @@ public class EventStat {
}
public int getRetryCnt() {
- return myRetryCnt;
+ return myRetryCnt.get();
}
public void setRetryCnt(int retryCnt) {
- this.myRetryCnt = retryCnt;
+ this.myRetryCnt.set(retryCnt);
}
public void incRetryCnt() {
- this.myRetryCnt++;
+ this.myRetryCnt.incrementAndGet();
+ }
+
+ public boolean isOrderMessage() {
+ return isOrderMessage;
}
public boolean shouldDrop() {
@@ -59,6 +72,6 @@ public class EventStat {
public void reset() {
this.event = null;
- this.myRetryCnt = 0;
+ this.myRetryCnt.set(0);
}
}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
index 645c32a..db81416 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
@@ -25,6 +25,7 @@ import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.handler.codec.TooLongFrameException;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
@@ -39,6 +40,7 @@ import org.apache.inlong.common.monitor.MonitorIndex;
import org.apache.inlong.common.monitor.MonitorIndexExt;
import org.apache.inlong.common.reporpter.StreamConfigLogMetric;
import org.apache.inlong.dataproxy.base.HighPriorityThreadFactory;
+import org.apache.inlong.dataproxy.base.OrderEvent;
import org.apache.inlong.dataproxy.config.ConfigManager;
import org.apache.inlong.dataproxy.config.holder.ConfigUpdateCallback;
import org.apache.inlong.dataproxy.config.pojo.ThirdPartyClusterConfig;
@@ -49,10 +51,13 @@ import org.apache.inlong.dataproxy.metrics.DataProxyMetricItemSet;
import org.apache.inlong.dataproxy.sink.pulsar.CreatePulsarClientCallBack;
import org.apache.inlong.dataproxy.sink.pulsar.PulsarClientService;
import org.apache.inlong.dataproxy.sink.pulsar.SendMessageCallBack;
+import org.apache.inlong.dataproxy.sink.pulsar.SinkTask;
import org.apache.inlong.dataproxy.utils.FailoverChannelProcessorHolder;
+import org.apache.inlong.dataproxy.utils.MessageUtils;
import org.apache.inlong.dataproxy.utils.NetworkUtils;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
+import org.apache.pulsar.client.api.PulsarClientException.NotFoundException;
import org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError;
import org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException;
import org.slf4j.Logger;
@@ -62,16 +67,14 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
-import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
/**
- * use pulsarSink need adding such config, if these ara not config in flume.conf, PulsarSink will
- * use default value.
+ * use pulsarSink need adding such config, if these ara not config in dataproxy-pulsar.conf,
+ * PulsarSink will use default value.
* prefix of pulsar sink config in flume.conf like this XXX.sinks.XXX.property
* and properties are may these Configurations:
* type (*): value must be 'org.apache.inlong.dataproxy.sink.PulsarSink'
@@ -96,10 +99,6 @@ public class PulsarSink extends AbstractSink implements Configurable,
private static String TOPIC = "topic";
/*
- * default value
- */
- private static int BATCH_SIZE = 10000;
- /*
* for log
*/
private RateLimiter diskRateLimiter;
@@ -107,33 +106,19 @@ public class PulsarSink extends AbstractSink implements Configurable,
/*
* for stat
*/
+ private static AtomicLong totalPulsarSuccSendCnt = new AtomicLong(0);
+ private static AtomicLong totalPulsarSuccSendSize = new AtomicLong(0);
private AtomicLong currentSuccessSendCnt = new AtomicLong(0);
private AtomicLong lastSuccessSendCnt = new AtomicLong(0);
private long t1 = System.currentTimeMillis();
private long t2 = 0L;
- private static AtomicLong totalPulsarSuccSendCnt = new AtomicLong(0);
- private static AtomicLong totalPulsarSuccSendSize = new AtomicLong(0);
- /*
- * for control
- */
- private boolean overflow = false;
- private LinkedBlockingQueue<EventStat> resendQueue;
+ private AtomicInteger processIndex = new AtomicInteger(0);
private int maxMonitorCnt = 300000;
- private long logCounter = 0;
-
private final AtomicLong currentInFlightCount = new AtomicLong(0);
- private static ConcurrentHashMap<String, Long> illegalTopicMap =
- new ConcurrentHashMap<String, Long>();
-
- /*
- * whether the SendTask thread can send data to pulsar
- */
- private volatile boolean canSend = false;
-
/*
* Control whether the SinkRunner thread can read data from the Channel
*/
@@ -142,30 +127,36 @@ public class PulsarSink extends AbstractSink implements Configurable,
/*
* log tools
*/
- private static final LogCounter logPrinterA = new LogCounter(10, 100000, 60 * 1000);
private static final LogCounter logPrinterB = new LogCounter(10, 100000, 60 * 1000);
private static final LogCounter logPrinterC = new LogCounter(10, 100000, 60 * 1000);
+ private SinkCounter sinkCounter;
+ /*
+ * message queue and retry
+ */
private int eventQueueSize = 10000;
private int badEventQueueSize = 10000;
+ private int maxRetrySendCnt = 16;
/*
* send thread pool
*/
- private Thread[] sinkThreadPool;
+ private SinkTask[] sinkThreadPool;
private int sinkThreadPoolSize;
private PulsarClientService pulsarClientService;
- private LinkedBlockingQueue<Event> eventQueue;
private static final String SEPARATOR = "#";
- private boolean isNewMetricOn = true;
- private boolean keepOrder = false;
+ /*
+ * statistic info log
+ */
private MonitorIndex monitorIndex;
private MonitorIndexExt monitorIndexExt;
- private SinkCounter sinkCounter;
+ /*
+ * report error log
+ */
private StreamConfigLogMetric streamConfigLogMetric;
private String localIp;
@@ -219,8 +210,6 @@ public class PulsarSink extends AbstractSink implements Configurable,
@Override
public void configure(Context context) {
logger.info("PulsarSink started and context = {}", context.toString());
- isNewMetricOn = context.getBoolean("new-metric-on", true);
- keepOrder = context.getBoolean("keep-order", false);
maxMonitorCnt = context.getInteger("max-monitor-cnt", 300000);
configManager = ConfigManager.getInstance();
@@ -228,12 +217,7 @@ public class PulsarSink extends AbstractSink implements Configurable,
pulsarCluster = configManager.getThirdPartyClusterUrl2Token();
pulsarConfig = configManager.getThirdPartyClusterConfig(); //pulsar common config
commonProperties = configManager.getCommonProperties();
- if (keepOrder) {
- logger.info("This is order pulsar sink!");
- sinkThreadPoolSize = 1;
- } else {
- sinkThreadPoolSize = pulsarConfig.getThreadNum();
- }
+ sinkThreadPoolSize = pulsarConfig.getThreadNum();
if (sinkThreadPoolSize <= 0) {
sinkThreadPoolSize = 1;
}
@@ -274,16 +258,11 @@ public class PulsarSink extends AbstractSink implements Configurable,
}
}
});
+ maxRetrySendCnt = pulsarConfig.getMaxRetryCnt();
badEventQueueSize = pulsarConfig.getBadEventQueueSize();
- resendQueue = new LinkedBlockingQueue<EventStat>(badEventQueueSize);
-
Preconditions.checkArgument(pulsarConfig.getThreadNum() > 0, "threadNum must be > 0");
-
- sinkThreadPool = new Thread[sinkThreadPoolSize];
-
+ sinkThreadPool = new SinkTask[sinkThreadPoolSize];
eventQueueSize = pulsarConfig.getEventQueueSize();
- eventQueue = new LinkedBlockingQueue<Event>(eventQueueSize);
-
if (pulsarConfig.getDiskIoRatePerSec() != 0) {
diskRateLimiter = RateLimiter.create(pulsarConfig.getDiskIoRatePerSec());
}
@@ -362,46 +341,46 @@ public class PulsarSink extends AbstractSink implements Configurable,
@Override
public void start() {
- logger.info("pulsar sink starting...");
+ logger.info("[{}] pulsar sink starting...", getName());
+ //register metrics
this.dimensions = new HashMap<>();
this.dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID, "DataProxy");
this.dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getName());
- //register metrics
this.metricItemSet = new DataProxyMetricItemSet(this.getName());
MetricRegister.register(metricItemSet);
sinkCounter.start();
pulsarClientService.initCreateConnection(this);
+
int statIntervalSec = pulsarConfig.getStatIntervalSec();
Preconditions.checkArgument(statIntervalSec >= 0, "statIntervalSec must be >= 0");
if (statIntervalSec > 0) {
/*
* switch for lots of metrics
*/
- if (isNewMetricOn) {
- monitorIndex = new MonitorIndex("Pulsar_Sink", statIntervalSec, maxMonitorCnt);
- }
+ monitorIndex = new MonitorIndex("Pulsar_Sink", statIntervalSec, maxMonitorCnt);
monitorIndexExt = new MonitorIndexExt("Pulsar_Sink_monitors#" + this.getName(),
statIntervalSec, maxMonitorCnt);
}
super.start();
- this.canSend = true;
- this.canTake = true;
try {
- initTopicSet(pulsarClientService,
- new HashSet<String>(topicProperties.values()));
+ initTopicSet(pulsarClientService, new HashSet<String>(topicProperties.values()));
} catch (Exception e) {
logger.info("pulsar sink start publish topic fail.", e);
}
- for (int i = 0; i < sinkThreadPool.length; i++) {
- sinkThreadPool[i] = new Thread(new SinkTask(pulsarClientService, i), getName()
- + "_pulsar_sink_sender-" + i);
+ for (int i = 0; i < sinkThreadPoolSize; i++) {
+ sinkThreadPool[i] = new SinkTask(pulsarClientService, this,
+ eventQueueSize / sinkThreadPoolSize,
+ badEventQueueSize / sinkThreadPoolSize, i, true);
+ sinkThreadPool[i].setName(getName() + "_pulsar_sink_sender-" + i);
sinkThreadPool[i].start();
}
- logger.debug("pulsar sink started");
+
+ this.canTake = true;
+ logger.info("[{}] Pulsar sink started", getName());
}
@Override
@@ -409,7 +388,7 @@ public class PulsarSink extends AbstractSink implements Configurable,
logger.info("pulsar sink stopping");
this.canTake = false;
int waitCount = 0;
- while (eventQueue.size() != 0 && waitCount++ < 10) {
+ while (isAllSendFinished() && waitCount++ < 10) {
try {
Thread.currentThread().sleep(1000);
} catch (InterruptedException e) {
@@ -417,7 +396,6 @@ public class PulsarSink extends AbstractSink implements Configurable,
break;
}
}
- this.canSend = false;
if (pulsarConfig.getStatIntervalSec() > 0) {
try {
monitorIndex.shutDown();
@@ -429,8 +407,9 @@ public class PulsarSink extends AbstractSink implements Configurable,
pulsarClientService.close();
}
if (sinkThreadPool != null) {
- for (Thread thread : sinkThreadPool) {
+ for (SinkTask thread : sinkThreadPool) {
if (thread != null) {
+ thread.close();
thread.interrupt();
}
}
@@ -445,6 +424,15 @@ public class PulsarSink extends AbstractSink implements Configurable,
logger.debug("pulsar sink stopped. Metrics:{}", sinkCounter);
}
+ private boolean isAllSendFinished() {
+ for (int i = 0; i < sinkThreadPoolSize; i++) {
+ if (!sinkThreadPool[i].isAllSendFinished()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
@Override
public Status process() throws EventDeliveryException {
if (!this.canTake) {
@@ -461,28 +449,22 @@ public class PulsarSink extends AbstractSink implements Configurable,
if (diskRateLimiter != null) {
diskRateLimiter.acquire(event.getBody().length);
}
- if (!eventQueue.offer(event, 3 * 1000, TimeUnit.MILLISECONDS)) {
+ if (!processEvent(new EventStat(event))) {
logger.info("[{}] Channel --> Queue(has no enough space,current code point) "
+ "--> pulsar,Check if pulsar server or network is ok.(if this situation "
+ "last long time it will cause memoryChannel full and fileChannel write.)", getName());
tx.rollback();
// metric
- if (event.getHeaders().containsKey(TOPIC)) {
- dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, event.getHeaders().get(TOPIC));
- } else {
- dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, "");
- }
+ dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID,
+ event.getHeaders().getOrDefault(TOPIC, ""));
DataProxyMetricItem metricItem = this.metricItemSet.findMetricItem(dimensions);
metricItem.readFailCount.incrementAndGet();
metricItem.readFailSize.addAndGet(event.getBody().length);
} else {
tx.commit();
// metric
- if (event.getHeaders().containsKey(TOPIC)) {
- dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, event.getHeaders().get(TOPIC));
- } else {
- dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID, "");
- }
+ dimensions.put(DataProxyMetricItem.KEY_SINK_DATA_ID,
+ event.getHeaders().getOrDefault(TOPIC, ""));
DataProxyMetricItem metricItem = this.metricItemSet.findMetricItem(dimensions);
metricItem.readSuccessCount.incrementAndGet();
metricItem.readFailSize.addAndGet(event.getBody().length);
@@ -505,7 +487,7 @@ public class PulsarSink extends AbstractSink implements Configurable,
return status;
}
- private void editStatistic(final Event event, String keyPostfix, String msgId) {
+ private void editStatistic(final Event event, String keyPostfix, boolean isOrder) {
String topic = "";
String streamId = "";
String nodeIp = null;
@@ -552,16 +534,15 @@ public class PulsarSink extends AbstractSink implements Configurable,
.get(ConfigConstants.MSG_COUNTER_KEY));
}
- /*
- * SINK_INTF#pulsarsink1#topic#streamId#clientIp#busIP#pkgTime#successCnt
- * #packcnt
- * #packsize#failCnt
- */
+ String orderType = "non-order";
+ if (isOrder) {
+ orderType = "order";
+ }
StringBuilder newbase = new StringBuilder();
newbase.append(this.getName()).append(SEPARATOR).append(topic).append(SEPARATOR)
.append(streamId).append(SEPARATOR).append(nodeIp)
.append(SEPARATOR).append(NetworkUtils.getLocalIp())
- .append(SEPARATOR).append(SEPARATOR)
+ .append(SEPARATOR).append(orderType).append(SEPARATOR)
.append(event.getHeaders().get(ConfigConstants.PKG_TIME_KEY));
long messageSize = event.getBody().length;
@@ -632,17 +613,21 @@ public class PulsarSink extends AbstractSink implements Configurable,
metricItem.sendCount.incrementAndGet();
metricItem.sendSize.addAndGet(eventStat.getEvent().getBody().length);
monitorIndexExt.incrementAndGet("PULSAR_SINK_SUCCESS");
- editStatistic(eventStat.getEvent(), null, result.toString());
+ editStatistic(eventStat.getEvent(), null, eventStat.isOrderMessage());
}
@Override
- public void handleMessageSendException(String topic, EventStat eventStat, Object e) {
+ public void handleMessageSendException(String topic, EventStat eventStat, Object e) {
monitorIndexExt.incrementAndGet("PULSAR_SINK_EXP");
- if (e instanceof TooLongFrameException) {
- PulsarSink.this.overflow = true;
+ boolean needRetry = true;
+ if (e instanceof NotFoundException) {
+ logger.error("NotFoundException topic {}, message will be discard!", topic, e);
+ needRetry = false;
+ } else if (e instanceof TooLongFrameException) {
+ logger.error("Send failed TooLongFrameException! {}{}", getName(), e);
} else if (e instanceof ProducerQueueIsFullError) {
- PulsarSink.this.overflow = true;
+ logger.error("Send failed ProducerQueueIsFullError! {}{}", getName(), e);
} else if (!(e instanceof AlreadyClosedException
|| e instanceof PulsarClientException.NotConnectedException
|| e instanceof TopicTerminatedException)) {
@@ -650,47 +635,91 @@ public class PulsarSink extends AbstractSink implements Configurable,
logger.error("Send failed!{}{}", getName(), e);
}
if (eventStat.getRetryCnt() == 0) {
- editStatistic(eventStat.getEvent(), "failure", "");
+ editStatistic(eventStat.getEvent(), "failure", eventStat.isOrderMessage());
}
}
- eventStat.incRetryCnt();
Map<String, String> dimensions = getNewDimension(DataProxyMetricItem.KEY_SINK_DATA_ID,
topic);
DataProxyMetricItem metricItem = this.metricItemSet.findMetricItem(dimensions);
metricItem.sendFailCount.incrementAndGet();
metricItem.sendFailSize.addAndGet(eventStat.getEvent().getBody().length);
- resendEvent(eventStat, true);
+ eventStat.incRetryCnt();
+ if (!eventStat.isOrderMessage() && needRetry) {
+ processResendEvent(eventStat);
+ }
}
- /**
- * Resend the data and store the data in the memory cache.
- * @param es
- * @param isDecrement
- */
- private void resendEvent(EventStat es, boolean isDecrement) {
+ private Map getNewDimension(String otherKey, String value) {
+ Map dimensions = new HashMap<>();
+ dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID, "DataProxy");
+ dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getName());
+ dimensions.put(otherKey, value);
+ return dimensions;
+ }
+
+ private boolean processEvent(EventStat eventStat) {
+ boolean result = true;
+ if (eventStat == null || eventStat.getEvent() == null) {
+ return result;
+ }
+ Event event = eventStat.getEvent();
+ if (MessageUtils.isSyncSendForOrder(event) && (event instanceof OrderEvent)) {
+ String partitionKey = event.getHeaders().get(AttributeConstants.MESSAGE_PARTITION_KEY);
+ SinkTask sinkTask =
+ sinkThreadPool[Math.abs(partitionKey.hashCode()) % sinkThreadPoolSize];
+ result = sinkTask.processEvent(eventStat);
+ } else {
+ int num = 0;
+ do {
+ int index = processIndex.getAndIncrement();
+ SinkTask sinkTask = sinkThreadPool[index % sinkThreadPoolSize];
+ if (sinkTask != null) {
+ result = sinkTask.processEvent(eventStat);
+ if (result) {
+ break;
+ }
+ }
+ num++;
+ } while (num < sinkThreadPoolSize);
+ }
+ return result;
+ }
+
+ private void processResendEvent(EventStat eventStat) {
try {
- if (isDecrement) {
- currentInFlightCount.decrementAndGet();
- }
- if (es == null || es.getEvent() == null) {
+ if (eventStat == null || eventStat.getEvent() == null) {
+ logger.warn("processResendEvent eventStat is null!");
return;
}
+ if (eventStat.getRetryCnt() == 1) {
+ currentInFlightCount.decrementAndGet();
+ }
/*
* If the failure requires retransmission to pulsar,
* the sid needs to be removed before retransmission.
*/
if (pulsarConfig.getClientIdCache()) {
- String clientId = es.getEvent().getHeaders().get(ConfigConstants.SEQUENCE_ID);
+ String clientId = eventStat.getEvent().getHeaders().get(ConfigConstants.SEQUENCE_ID);
if (clientId != null && agentIdCache.asMap().containsKey(clientId)) {
agentIdCache.invalidate(clientId);
}
}
- if (!resendQueue.offer(es)) {
- /*
- * If the retry queue is full, it will re-enter the channel; if it is not full,
- * it will be taken out of the resendQueue in sinktask
- */
- FailoverChannelProcessorHolder.getChannelProcessor().processEvent(es.getEvent());
+ boolean result = false;
+ int num = 0;
+ do {
+ int index = processIndex.getAndIncrement();
+ SinkTask sinkTask = sinkThreadPool[index % sinkThreadPoolSize];
+ if (sinkTask != null) {
+ result = sinkTask.processReSendEvent(eventStat);
+ if (result) {
+ break;
+ }
+ }
+ num++;
+ } while (num < sinkThreadPoolSize);
+ if (!result) {
+ FailoverChannelProcessorHolder.getChannelProcessor()
+ .processEvent(eventStat.getEvent());
if (logPrinterC.shouldPrint()) {
logger.error(getName() + " Channel --> pulsar --> ResendQueue(full) "
+ "-->FailOverChannelProcessor(current code point), "
@@ -701,18 +730,33 @@ public class PulsarSink extends AbstractSink implements Configurable,
monitorIndexExt.incrementAndGet("PULSAR_SINK_DROPPED");
if (logPrinterC.shouldPrint()) {
logger.error(getName() + " Discard msg because put events to both of "
- + "queue and fileChannel fail,current resendQueue.size = "
- + resendQueue.size(), throwable);
+ + "queue and fileChannel fail", throwable);
}
}
}
- private Map getNewDimension(String otherKey, String value) {
- Map dimensions = new HashMap<>();
- dimensions.put(DataProxyMetricItem.KEY_CLUSTER_ID, "DataProxy");
- dimensions.put(DataProxyMetricItem.KEY_SINK_ID, this.getName());
- dimensions.put(otherKey, value);
- return dimensions;
+ public LoadingCache<String, Long> getAgentIdCache() {
+ return agentIdCache;
+ }
+
+ public Map<String, String> getTopicsProperties() {
+ return topicProperties;
+ }
+
+ public SinkCounter getSinkCounter() {
+ return sinkCounter;
+ }
+
+ public AtomicLong getCurrentInFlightCount() {
+ return currentInFlightCount;
+ }
+
+ public ThirdPartyClusterConfig getPulsarConfig() {
+ return pulsarConfig;
+ }
+
+ public int getMaxRetrySendCnt() {
+ return maxRetrySendCnt;
}
static class PulsarPerformanceTask implements Runnable {
@@ -737,174 +781,4 @@ public class PulsarSink extends AbstractSink implements Configurable,
}
}
}
-
- class SinkTask implements Runnable {
-
- private PulsarClientService pulsarClientService;
-
- private int poolIndex = 0;
-
- public SinkTask(PulsarClientService pulsarClientService, int poolIndex) {
- this.pulsarClientService = pulsarClientService;
- this.poolIndex = poolIndex;
- }
-
- @Override
- public void run() {
- logger.info("Sink task {} started.", Thread.currentThread().getName());
- while (canSend) {
- boolean decrementFlag = false;
- Event event = null;
- EventStat eventStat = null;
- String topic = null;
- try {
- if (PulsarSink.this.overflow) {
- PulsarSink.this.overflow = false;
- Thread.currentThread().sleep(10);
- }
- if (!resendQueue.isEmpty()) {
- /*
- * Send the data in the retry queue first
- */
- eventStat = resendQueue.poll();
- if (eventStat != null) {
- event = eventStat.getEvent();
- // logger.warn("Resend event: {}", event.toString());
- if (event.getHeaders().containsKey(TOPIC)) {
- topic = event.getHeaders().get(TOPIC);
- }
- }
- } else {
- if (currentInFlightCount.get() > BATCH_SIZE) {
- /*
- * Under the condition that the number of unresponsive messages
- * is greater than 1w, the number of unresponsive messages sent
- * to pulsar will be printed periodically
- */
- logCounter++;
- if (logCounter == 1 || logCounter % 100000 == 0) {
- logger.info(getName()
- + " currentInFlightCount={} resendQueue"
- + ".size={}",
- currentInFlightCount.get(), resendQueue.size());
- }
- if (logCounter > Long.MAX_VALUE - 10) {
- logCounter = 0;
- }
- }
- event = eventQueue.take();
- eventStat = new EventStat(event);
- sinkCounter.incrementEventDrainAttemptCount();
- if (event.getHeaders().containsKey(TOPIC)) {
- topic = event.getHeaders().get(TOPIC);
- }
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Event is {}, topic = {} ", event, topic);
- }
-
- if (event == null) {
- continue;
- }
-
- if (topic == null || topic.equals("")) {
- logger.warn("no topic specified in event header, just skip this event");
- continue;
- }
-
- Long expireTime = illegalTopicMap.get(topic);
- if (expireTime != null) {
- long currentTime = System.currentTimeMillis();
- if (expireTime > currentTime) {
- /*
- * If the exception-channel is configured, put the exception-channel,
- * otherwise discard the data
- */
- continue;
- } else {
- /*
- * Illegal has expired, there is no need to put it in the map
- */
- illegalTopicMap.remove(topic);
- }
- }
-
- String clientId = event.getHeaders().get(ConfigConstants.SEQUENCE_ID);
- final EventStat es = eventStat;
-
- boolean hasKey = false;
- if (pulsarConfig.getClientIdCache() && clientId != null) {
- hasKey = agentIdCache.asMap().containsKey(clientId);
- }
-
- if (pulsarConfig.getClientIdCache() && clientId != null && hasKey) {
- agentIdCache.put(clientId, System.currentTimeMillis());
- if (logPrinterA.shouldPrint()) {
- logger.info("{} agent package {} existed,just discard.",
- getName(), clientId);
- }
- } else {
- if (pulsarConfig.getClientIdCache() && clientId != null) {
- agentIdCache.put(clientId, System.currentTimeMillis());
- }
- boolean sendResult = pulsarClientService.sendMessage(poolIndex, topic,
- event, PulsarSink.this, es);
- /*
- * handle producer is current is null
- */
- if (!sendResult) {
- illegalTopicMap.put(topic, System.currentTimeMillis() + 30 * 1000);
- continue;
- }
- currentInFlightCount.incrementAndGet();
- decrementFlag = true;
- }
- /*
- * No exception is thrown, after a complete one-time sending,
- * the topic can be deleted from the illegal list
- */
- illegalTopicMap.remove(topic);
- } catch (InterruptedException e) {
- logger.info("Thread {} has been interrupted!", Thread.currentThread().getName());
- return;
- } catch (Throwable t) {
- if (t instanceof PulsarClientException) {
- String message = t.getMessage();
- if (message != null && (message.contains("No available queue for topic")
- || message.contains("The brokers of topic are all forbidden"))) {
- illegalTopicMap.put(topic, System.currentTimeMillis() + 60 * 1000);
- logger.info("IllegalTopicMap.put " + topic);
- continue;
- } else {
- try {
- /*
- * The exception of pulsar will cause the sending thread to block
- * and prevent further pressure on pulsar. Here you should pay
- * attention to the type of exception to prevent the error of
- * a topic from affecting the global
- */
- Thread.sleep(100);
- } catch (InterruptedException e) {
- //ignore..
- }
- }
- }
- if (logPrinterA.shouldPrint()) {
- logger.error("Sink task fail to send the message, decrementFlag="
- + decrementFlag
- + ",sink.name="
- + Thread.currentThread().getName()
- + ",event.headers="
- + eventStat.getEvent().getHeaders(), t);
- }
- /*
- * producer.sendMessage is abnormal,
- * so currentInFlightCount is not added,
- * so there is no need to subtract
- */
- resendEvent(eventStat, decrementFlag);
- }
- }
- }
- }
}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
index f7df0dd..c57fe84 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
@@ -22,7 +22,6 @@ import io.netty.buffer.ByteBuf;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Event;
import org.apache.flume.FlumeException;
-import org.apache.inlong.common.monitor.LogCounter;
import org.apache.inlong.common.reporpter.ConfigLogTypeEnum;
import org.apache.inlong.common.reporpter.StreamConfigLogMetric;
import org.apache.inlong.dataproxy.base.OrderEvent;
@@ -42,6 +41,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.NotFoundException;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,18 +59,20 @@ public class PulsarClientService {
private static final Logger logger = LoggerFactory.getLogger(PulsarClientService.class);
- private static final LogCounter logPrinterA = new LogCounter(10, 100000, 60 * 1000);
-
/*
* for pulsar client
*/
private Map<String, String> pulsarUrl2token;
private String authType;
+
/*
* for producer
*/
- private Integer sendTimeout; // in millsec
+ /*
+ * unit mills
+ */
+ private Integer sendTimeout;
private Integer clientTimeout;
private boolean enableBatch = true;
private boolean blockIfQueueFull = true;
@@ -124,8 +126,8 @@ public class PulsarClientService {
maxBatchingMessages = pulsarConfig.getMaxBatchingMessages();
maxBatchingBytes = pulsarConfig.getMaxBatchingBytes();
maxBatchingPublishDelayMillis = pulsarConfig.getMaxBatchingPublishDelayMillis();
- producerInfoMap = new ConcurrentHashMap<String, List<TopicProducerInfo>>();
- topicSendIndexMap = new ConcurrentHashMap<String, AtomicLong>();
+ producerInfoMap = new ConcurrentHashMap<>();
+ topicSendIndexMap = new ConcurrentHashMap<>();
localIp = NetworkUtils.getLocalIp();
}
@@ -155,20 +157,14 @@ public class PulsarClientService {
public boolean sendMessage(int poolIndex, String topic, Event event,
SendMessageCallBack sendMessageCallBack, EventStat es) {
TopicProducerInfo producerInfo = null;
+ boolean result = false;
final String inlongStreamId = getInlongStreamId(event);
final String inlongGroupId = getInlongGroupId(event);
try {
producerInfo = getProducerInfo(poolIndex, topic, inlongGroupId, inlongStreamId);
} catch (Exception e) {
producerInfo = null;
- if (logPrinterA.shouldPrint()) {
- /*
- * If it is not an IllegalTopicException,
- * the producer may be null,
- * causing the sendMessage part to report a null pointer later
- */
- logger.error("Get producer failed!", e);
- }
+ logger.error("Get producer failed! topic = {}", topic, e);
if (streamConfigLogMetric != null) {
streamConfigLogMetric.updateConfigLog(inlongGroupId,
inlongStreamId, StreamConfigLogMetric.CONFIG_LOG_PULSAR_PRODUCER,
@@ -187,9 +183,9 @@ public class PulsarClientService {
* put it back into the illegal map
*/
checkAndResponse(event, inlongGroupId, inlongStreamId);
- sendMessageCallBack.handleMessageSendException(topic, es, new Exception("producer "
- + " info is null"));
- return false;
+ sendMessageCallBack.handleMessageSendException(topic, es,
+ new NotFoundException("producer info is null"));
+ return true;
}
Map<String, String> proMap = new HashMap<>();
@@ -199,12 +195,13 @@ public class PulsarClientService {
TopicProducerInfo forCallBackP = producerInfo;
Producer producer = producerInfo.getProducer(poolIndex);
if (producer == null) {
+ logger.warn("Get producer is null! topic = {}", topic);
checkAndResponse(event, inlongGroupId, inlongStreamId);
- sendMessageCallBack.handleMessageSendException(topic, es, new Exception("producer is "
+ sendMessageCallBack.handleMessageSendException(topic, es, new NotFoundException("producer is "
+ "null"));
- return false;
+ return true;
}
- if (MessageUtils.isSyncSendForOrder(event) && (event instanceof OrderEvent)) {
+ if (es.isOrderMessage()) {
String partitionKey = event.getHeaders().get(AttributeConstants.MESSAGE_PARTITION_KEY);
try {
MessageId msgId = producer.newMessage().key(partitionKey)
@@ -213,6 +210,7 @@ public class PulsarClientService {
sendMessageCallBack.handleMessageSendSuccess(topic, msgId, es);
AuditUtils.add(AuditUtils.AUDIT_ID_DATAPROXY_SEND_SUCCESS, event);
forCallBackP.setCanUseSend(true);
+ result = true;
} catch (PulsarClientException ex) {
if (streamConfigLogMetric != null) {
streamConfigLogMetric.updateConfigLog(inlongGroupId,
@@ -221,8 +219,18 @@ public class PulsarClientService {
}
forCallBackP.setCanUseSend(false);
sendMessageCallBack.handleMessageSendException(topic, es, ex);
+ result = false;
+ if (ex instanceof NotFoundException) {
+ result = true;
+ }
+ }
+ /*
+ * avoid client timeout
+ */
+ logger.debug("es.getRetryCnt() {}", es.getRetryCnt());
+ if (es.getRetryCnt() == 0 || es.getRetryCnt() == 1) {
+ sendResponse((OrderEvent)event, inlongGroupId, inlongStreamId);
}
- sendResponse((OrderEvent)event, inlongGroupId, inlongStreamId);
} else {
producer.newMessage().properties(proMap).value(event.getBody())
.sendAsync().thenAccept((msgId) -> {
@@ -239,8 +247,9 @@ public class PulsarClientService {
sendMessageCallBack.handleMessageSendException(topic, es, e);
return null;
});
+ result = true;
}
- return true;
+ return result;
}
private void checkAndResponse(Event event, String inlongGroupId, String inlongStreamId) {
@@ -258,7 +267,7 @@ public class PulsarClientService {
if ("false".equals(orderEvent.getHeaders().get(AttributeConstants.MESSAGE_IS_ACK))) {
if (logger.isDebugEnabled()) {
logger.debug("Not need to rsp message: seqId = {}, inlongGroupId = {}, "
- + "inlongStreamId = {}",inlongGroupId, inlongStreamId, sequenceId);
+ + "inlongStreamId = {}",sequenceId, inlongGroupId, inlongStreamId);
}
return;
}
@@ -266,7 +275,7 @@ public class PulsarClientService {
orderEvent.getCtx().channel().eventLoop().execute(() -> {
if (logger.isDebugEnabled()) {
logger.debug("order message rsp: seqId = {}, inlongGroupId = {}, "
- + "inlongStreamId = {}",inlongGroupId, inlongStreamId, sequenceId);
+ + "inlongStreamId = {}",sequenceId, inlongGroupId, inlongStreamId);
}
ByteBuf binBuffer = MessageUtils.getResponsePackage("",
MsgType.MSG_BIN_MULTI_BODY, sequenceId);
@@ -288,7 +297,7 @@ public class PulsarClientService {
pulsarClients = new ConcurrentHashMap<>();
pulsarUrl2token = ConfigManager.getInstance().getThirdPartyClusterUrl2Token();
Preconditions.checkState(!pulsarUrl2token.isEmpty(), "No pulsar server url specified");
- logger.debug("number of pulsarcluster is {}", pulsarUrl2token.size());
+ logger.debug("number of pulsar cluster is {}", pulsarUrl2token.size());
for (Map.Entry<String, String> info : pulsarUrl2token.entrySet()) {
try {
if (logger.isDebugEnabled()) {
@@ -305,13 +314,13 @@ public class PulsarClientService {
StreamConfigLogMetric.CONFIG_LOG_PULSAR_CLIENT,
ConfigLogTypeEnum.ERROR, e.toString());
}
- logger.error("create connnection error in metasink, "
+ logger.error("create connection error in Pulsar sink, "
+ "maybe pulsar master set error, please re-check.url{}, ex1 {}",
info.getKey(),
e.getMessage());
} catch (Throwable e) {
callBack.handleCreateClientException(info.getKey());
- logger.error("create connnection error in metasink, "
+ logger.error("create connection error in pulsar sink, "
+ "maybe pulsar master set error/shutdown in progress, please "
+ "re-check. url{}, ex2 {}",
info.getKey(),
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SinkTask.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SinkTask.java
new file mode 100644
index 0000000..3157f0d
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SinkTask.java
@@ -0,0 +1,270 @@
+/*
+ * 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.pulsar;
+
+import com.google.common.cache.LoadingCache;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flume.Event;
+import org.apache.flume.instrumentation.SinkCounter;
+import org.apache.inlong.common.monitor.LogCounter;
+import org.apache.inlong.dataproxy.config.pojo.ThirdPartyClusterConfig;
+import org.apache.inlong.dataproxy.consts.AttributeConstants;
+import org.apache.inlong.dataproxy.consts.ConfigConstants;
+import org.apache.inlong.dataproxy.sink.EventStat;
+import org.apache.inlong.dataproxy.sink.PulsarSink;
+import org.apache.inlong.dataproxy.utils.MessageUtils;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SinkTask extends Thread {
+
+ private static final Logger logger = LoggerFactory.getLogger(SinkTask.class);
+
+ private static final LogCounter logPrinterA = new LogCounter(10, 100000, 60 * 1000);
+
+ private static String TOPIC = "topic";
+
+ /*
+ * default value
+ */
+ private static int BATCH_SIZE = 10000;
+
+ private PulsarClientService pulsarClientService;
+
+ private PulsarSink pulsarSink;
+
+ private long logCounter = 0;
+
+ private int poolIndex = 0;
+
+ private LinkedBlockingQueue<EventStat> eventQueue;
+
+ private LinkedBlockingQueue<EventStat> resendQueue;
+
+ private AtomicLong currentInFlightCount;
+
+ private SinkCounter sinkCounter;
+
+ private LoadingCache<String, Long> agentIdCache;
+
+ private ThirdPartyClusterConfig pulsarConfig;
+
+ private int maxRetrySendCnt;
+ /*
+ * whether the SendTask thread can send data to pulsar
+ */
+ private volatile boolean canSend = false;
+
+ public SinkTask(PulsarClientService pulsarClientService, PulsarSink pulsarSink,
+ int eventQueueSize,
+ int badEventQueueSize, int poolIndex, boolean canSend) {
+ this.pulsarClientService = pulsarClientService;
+ this.pulsarSink = pulsarSink;
+ this.poolIndex = poolIndex;
+ this.canSend = canSend;
+ this.currentInFlightCount = pulsarSink.getCurrentInFlightCount();
+ this.sinkCounter = pulsarSink.getSinkCounter();
+ this.agentIdCache = pulsarSink.getAgentIdCache();
+ this.pulsarConfig = pulsarSink.getPulsarConfig();
+ this.maxRetrySendCnt = pulsarSink.getMaxRetrySendCnt();
+ eventQueue = new LinkedBlockingQueue<>(eventQueueSize);
+ resendQueue = new LinkedBlockingQueue<>(badEventQueueSize);
+ }
+
+ public boolean processEvent(EventStat eventStat) {
+ try {
+ return eventQueue.offer(eventStat, 3 * 1000, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ logger.error("InterruptedException e", e);
+ }
+ return false;
+ }
+
+ public boolean processReSendEvent(EventStat eventStat) {
+ return resendQueue.offer(eventStat);
+ }
+
+ public boolean isAllSendFinished() {
+ return eventQueue.size() == 0;
+ }
+
+ public void close() {
+ canSend = false;
+ }
+
+ @Override
+ public void run() {
+ logger.info("Sink task {} started.", Thread.currentThread().getName());
+ while (canSend) {
+ boolean decrementFlag = false;
+ Event event = null;
+ EventStat eventStat = null;
+ String topic = null;
+ try {
+ if (!resendQueue.isEmpty()) {
+ /*
+ * Send the data in the retry queue first
+ */
+ eventStat = resendQueue.poll();
+ if (eventStat != null) {
+ event = eventStat.getEvent();
+ }
+ } else {
+ if (currentInFlightCount.get() > BATCH_SIZE) {
+ /*
+ * Under the condition that the number of unresponsive messages
+ * is greater than 1w, the number of unresponsive messages sent
+ * to pulsar will be printed periodically
+ */
+ logCounter++;
+ if (logCounter == 1 || logCounter % 100000 == 0) {
+ logger.info(getName()
+ + " currentInFlightCount={} resendQueue"
+ + ".size={}",
+ currentInFlightCount.get(), resendQueue.size());
+ }
+ if (logCounter > Long.MAX_VALUE - 10) {
+ logCounter = 0;
+ }
+ }
+ eventStat = eventQueue.take();
+ sinkCounter.incrementEventDrainAttemptCount();
+ event = eventStat.getEvent();
+ }
+
+ /*
+ * get topic
+ */
+ if (event.getHeaders().containsKey(TOPIC)) {
+ topic = event.getHeaders().get(TOPIC);
+ }
+ if (StringUtils.isEmpty(topic)) {
+ String groupId = event.getHeaders().get(AttributeConstants.GROUP_ID);
+ String streamId = event.getHeaders().get(AttributeConstants.STREAM_ID);
+ topic = MessageUtils.getTopic(pulsarSink.getTopicsProperties(), groupId, streamId);
+ }
+
+ if (event == null) {
+ logger.warn("Event is null!");
+ continue;
+ }
+
+ if (topic == null || topic.equals("")) {
+ pulsarSink.handleMessageSendException(topic, eventStat, new Exception("topic"
+ + " info is null"));
+ processToReTrySend(eventStat);
+ logger.warn("no topic specified, so will retry send!");
+ continue;
+ }
+
+ if (eventStat.isOrderMessage()) {
+ sleep(1000);
+ }
+
+ if (eventStat.getRetryCnt() > maxRetrySendCnt) {
+ logger.warn("Message will be discard! send times reach to max retry cnt."
+ + " topic = {}, max retry cnt = {}", topic, maxRetrySendCnt);
+ continue;
+ }
+
+ String clientSeqId = event.getHeaders().get(ConfigConstants.SEQUENCE_ID);
+
+ boolean hasSend = false;
+ if (pulsarConfig.getClientIdCache() && clientSeqId != null) {
+ hasSend = agentIdCache.asMap().containsKey(clientSeqId);
+ }
+
+ if (pulsarConfig.getClientIdCache() && clientSeqId != null && hasSend) {
+ agentIdCache.put(clientSeqId, System.currentTimeMillis());
+ if (logPrinterA.shouldPrint()) {
+ logger.info("{} agent package {} existed,just discard.",
+ getName(), clientSeqId);
+ }
+ } else {
+ if (pulsarConfig.getClientIdCache() && clientSeqId != null) {
+ agentIdCache.put(clientSeqId, System.currentTimeMillis());
+ }
+ boolean sendResult = pulsarClientService.sendMessage(poolIndex, topic,
+ event, pulsarSink, eventStat);
+ if (!sendResult) {
+ /*
+ * only for order message
+ */
+ processToReTrySend(eventStat);
+ }
+ currentInFlightCount.incrementAndGet();
+ decrementFlag = true;
+ }
+ } catch (InterruptedException e) {
+ logger.error("Thread {} has been interrupted!",
+ Thread.currentThread().getName());
+ return;
+ } catch (Throwable t) {
+ if (t instanceof PulsarClientException) {
+ String message = t.getMessage();
+ if (message != null && (message.contains("No available queue for topic")
+ || message.contains("The brokers of topic are all forbidden"))) {
+ logger.info("IllegalTopicMap.put " + topic);
+ continue;
+ } else {
+ try {
+ /*
+ * The exception of pulsar will cause the sending thread to block
+ * and prevent further pressure on pulsar. Here you should pay
+ * attention to the type of exception to prevent the error of
+ * a topic from affecting the global
+ */
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ //ignore..
+ }
+ }
+ }
+ if (logPrinterA.shouldPrint()) {
+ logger.error("Sink task fail to send the message, decrementFlag="
+ + decrementFlag
+ + ",sink.name="
+ + Thread.currentThread().getName()
+ + ",event.headers="
+ + eventStat.getEvent().getHeaders(), t);
+ }
+ /*
+ * producer.sendMessage is abnormal,
+ * so currentInFlightCount is not added,
+ * so there is no need to subtract
+ */
+ pulsarSink.handleMessageSendException(topic, eventStat, t);
+ processToReTrySend(eventStat);
+ }
+ }
+ }
+
+ public void processToReTrySend(EventStat eventStat) {
+ /*
+ * order message must be retried in local resendQueue
+ */
+ if (eventStat.isOrderMessage()) {
+
+ processReSendEvent(eventStat);
+ }
+ }
+}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
index dc06a9d..ca5070e 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java
@@ -292,7 +292,8 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
}
}
- String value = getTopic(groupId, streamId);
+ String value = MessageUtils.getTopic(configManager.getTopicProperties(), groupId,
+ streamId);
if (StringUtils.isNotEmpty(value)) {
topicInfo.set(value.trim());
}
@@ -328,7 +329,8 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
message.setGroupId(groupId);
message.setStreamId(streamId);
- String value = getTopic(groupId, streamId);
+ String value = MessageUtils.getTopic(configManager.getTopicProperties(),
+ groupId, streamId);
if (StringUtils.isNotEmpty(value)) {
topicInfo.set(value.trim());
}
@@ -351,9 +353,8 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
String streamId = message.getStreamId();
topic = topicInfo.get();
if (StringUtils.isEmpty(topic)) {
- logger.warn("Topic for message is null , inlongGroupId = {}, inlongStreamId",
+ logger.warn("Topic for message is null , inlongGroupId = {}, inlongStreamId = {}",
groupId, streamId);
- return false;
}
// if(groupId==null)groupId="b_test";//default groupId
@@ -485,8 +486,10 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
headers.put(ConfigConstants.PKG_TIME_KEY, pkgTimeStr);
Event event = EventBuilder.withBody(data, headers);
+ String orderType = "non-order";
if (MessageUtils.isSyncSendForOrder(event)) {
event = new OrderEvent(ctx, event);
+ orderType = "order";
}
long dtten = 0;
try {
@@ -508,6 +511,7 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
.append(streamIdEntry.getKey()).append(SEPARATOR)
.append(strRemoteIP).append(SEPARATOR)
.append(NetworkUtils.getLocalIp()).append(SEPARATOR)
+ .append(orderType).append(SEPARATOR)
.append(new SimpleDateFormat("yyyyMMddHHmm")
.format(dtten)).append(SEPARATOR).append(pkgTimeStr);
try {
@@ -739,32 +743,6 @@ public class ServerMessageHandler extends ChannelInboundHandlerAdapter {
}
/**
- * get topic
- */
- private String getTopic(String groupId) {
- return getTopic(groupId, null);
- }
-
- /**
- * get topic
- */
- private String getTopic(String groupId, String streamId) {
- String topic = null;
- if (StringUtils.isNotEmpty(groupId)) {
- if (StringUtils.isNotEmpty(streamId)) {
- topic = configManager.getTopicProperties().get(groupId + "/" + streamId);
- }
- if (StringUtils.isEmpty(topic)) {
- topic = configManager.getTopicProperties().get(groupId);
- }
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Get topic by groupId = {} , streamId = {}", groupId, streamId);
- }
- return topic;
- }
-
- /**
* addMetric
*
* @param result
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
index fe4647b..2aa0401 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java
@@ -18,13 +18,18 @@ package org.apache.inlong.dataproxy.utils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import java.nio.charset.StandardCharsets;
+import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.apache.flume.Event;
import org.apache.inlong.dataproxy.consts.AttributeConstants;
import org.apache.inlong.dataproxy.source.MsgType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class MessageUtils {
+ private static final Logger logger = LoggerFactory.getLogger(MessageUtils.class);
+
/**
* is or not sync send for order message
* @param syncSend syncSend
@@ -82,4 +87,24 @@ public class MessageUtils {
return binBuffer;
}
+ /**
+ * get topic
+ */
+ public static String getTopic(Map<String, String> topicsMap, String groupId, String streamId) {
+ String topic = null;
+ if (topicsMap != null && StringUtils.isNotEmpty(groupId)) {
+ if (StringUtils.isNotEmpty(streamId)) {
+ topic = topicsMap.get(groupId + "/" + streamId);
+ }
+ if (StringUtils.isEmpty(topic)) {
+ topic = topicsMap.get(groupId);
+ }
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("Get topic by groupId = {}, streamId = {}, topic = {}", groupId, streamId,
+ topic);
+ }
+ return topic;
+ }
+
}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/resources/dataproxy-pulsar.conf b/inlong-dataproxy/dataproxy-source/src/main/resources/dataproxy-pulsar.conf
new file mode 100644
index 0000000..0ef2f41
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/resources/dataproxy-pulsar.conf
@@ -0,0 +1,110 @@
+#
+# 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.
+#
+
+#
+# Attention:
+# use this example,use dataproxy-pulsar.conf as file name,
+# In addition, if you want to keep the order of messages,
+# in addition configuring the properties to order when sending,
+# you must configure separate channels and sinks for sequential messages,
+# and config order property in source module in dataproxy.
+#
+
+agent1.sources = tcp-source
+agent1.channels = ch-msg1 ch-msg2 ch-msg3 ch-msg5 ch-msg6
+agent1.sinks = pulsar-sink-msg1 pulsar-sink-msg2 pulsar-sink-msg3 pulsar-sink-msg5 pulsar-sink-msg6
+
+# For example:
+# For order message ,ch-msg3 and pulsar-sink-msg3 must be configured
+# agent1.channels = ch-msg1 ch-msg2 ch-msg3 ch-msg5 ch-msg6
+# agent1.sinks = pulsar-sink-msg1 pulsar-sink-msg2 pulsar-sink-msg3 pulsar-sink-msg5 pulsar-sink-msg6
+
+
+agent1.sources.tcp-source.channels = ch-msg1 ch-msg2 ch-msg3 ch-msg5 ch-msg6
+agent1.sources.tcp-source.type = org.apache.inlong.dataproxy.source.SimpleTcpSource
+agent1.sources.tcp-source.msg-factory-name = org.apache.inlong.dataproxy.source.ServerMessageFactory
+agent1.sources.tcp-source.host = 0.0.0.0
+agent1.sources.tcp-source.port = 46801
+agent1.sources.tcp-source.max-msg-length = 524288
+#agent1.sources.tcp-source.topic = persistent://public/default/dataproxy-default-topic
+agent1.sources.tcp-source.attr = m=9
+agent1.sources.tcp-source.connections = 30000
+agent1.sources.tcp-source.max-threads = 64
+agent1.sources.tcp-source.receiveBufferSize = 8388608
+agent1.sources.tcp-source.sendBufferSize = 8388608
+agent1.sources.tcp-source.custom-cp = true
+agent1.sources.tcp-source.selector.type = org.apache.inlong.dataproxy.channel.FailoverChannelSelector
+agent1.sources.tcp-source.selector.master = ch-msg1 ch-msg2
+# For order message
+agent1.sources.tcp-source.selector.order = ch-msg3
+agent1.sources.tcp-source.metric-recovery-path = /data/DataProxy/file/recovery
+agent1.sources.tcp-source.metric-agent-port = 8003
+agent1.sources.tcp-source.metric-cache-size = 1000000
+agent1.sources.tcp-source.set = 10
+
+agent1.channels.ch-msg1.type = memory
+agent1.channels.ch-msg1.capacity = 50000
+agent1.channels.ch-msg1.keep-alive = 0
+agent1.channels.ch-msg1.transactionCapacity = 200
+
+agent1.channels.ch-msg2.type = memory
+agent1.channels.ch-msg2.capacity = 50000
+agent1.channels.ch-msg2.keep-alive = 0
+agent1.channels.ch-msg2.transactionCapacity = 200
+
+# For order message
+agent1.channels.ch-msg3.type = memory
+agent1.channels.ch-msg3.capacity = 10000
+agent1.channels.ch-msg3.keep-alive = 0
+agent1.channels.ch-msg3.transactionCapacity = 200
+
+agent1.channels.ch-msg5.type = file
+agent1.channels.ch-msg5.capacity = 100000000
+agent1.channels.ch-msg5.maxFileSize = 1073741824
+agent1.channels.ch-msg5.minimumRequiredSpace = 1073741824
+agent1.channels.ch-msg5.checkpointDir = /data/DataProxy/file/ch-msg5/check
+agent1.channels.ch-msg5.dataDirs = DATAPROXY_HOME/file/ch-msg5/data
+agent1.channels.ch-msg5.fsyncPerTransaction = false
+agent1.channels.ch-msg5.fsyncInterval = 10
+
+agent1.channels.ch-msg6.type = file
+agent1.channels.ch-msg6.capacity = 100000000
+agent1.channels.ch-msg6.maxFileSize = 1073741824
+agent1.channels.ch-msg6.minimumRequiredSpace = 1073741824
+agent1.channels.ch-msg6.checkpointDir = /data/DataProxy/file/ch-msg6/check
+agent1.channels.ch-msg6.dataDirs = /data/DataProxy/file/ch-msg6/data
+agent1.channels.ch-msg6.fsyncPerTransaction = false
+agent1.channels.ch-msg6.fsyncInterval = 10
+
+agent1.sinks.pulsar-sink-msg1.channel = ch-msg1
+agent1.sinks.pulsar-sink-msg1.type = org.apache.inlong.dataproxy.sink.PulsarSink
+
+agent1.sinks.pulsar-sink-msg2.channel = ch-msg2
+agent1.sinks.pulsar-sink-msg2.type = org.apache.inlong.dataproxy.sink.PulsarSink
+
+# For order message
+agent1.sinks.pulsar-sink-msg3.channel = ch-msg3
+agent1.sinks.pulsar-sink-msg3.keep-order = true
+agent1.sinks.pulsar-sink-msg3.type = org.apache.inlong.dataproxy.sink.PulsarSink
+
+agent1.sinks.pulsar-sink-msg5.channel = ch-msg5
+agent1.sinks.pulsar-sink-msg5.type = org.apache.inlong.dataproxy.sink.PulsarSink
+
+agent1.sinks.pulsar-sink-msg6.channel = ch-msg6
+agent1.sinks.pulsar-sink-msg6.type = org.apache.inlong.dataproxy.sink.PulsarSink
diff --git a/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf b/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf
deleted file mode 100644
index 2bd68d1..0000000
--- a/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf
+++ /dev/null
@@ -1,162 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-#6-segment protocol TCP source
-agent1.sources.tcp-source.channels = ch-msg1 ch-msg2 ch-msg3 ch-more1 ch-more2 ch-more3 ch-msg5 ch-msg6 ch-msg7 ch-msg8 ch-msg9 ch-msg10 ch-back
-agent1.sources.tcp-source.type = org.apache.inlong.dataproxy.source.SimpleTcpSource
-agent1.sources.tcp-source.msg-factory-name = org.apache.inlong.dataproxy.source.ServerMessageFactory
-agent1.sources.tcp-source.host = 0.0.0.0
-agent1.sources.tcp-source.port = 46801
-agent1.sources.tcp-source.highWaterMark=2621440
-agent1.sources.tcp-source.enableExceptionReturn=true
-agent1.sources.tcp-source.max-msg-length = 524288
-agent1.sources.tcp-source.topic = test_token
-agent1.sources.tcp-source.attr = m=9
-agent1.sources.tcp-source.connections = 5000
-agent1.sources.tcp-source.max-threads = 64
-agent1.sources.tcp-source.receiveBufferSize = 524288
-agent1.sources.tcp-source.sendBufferSize = 524288
-agent1.sources.tcp-source.custom-cp = true
-agent1.sources.tcp-source.selector.type = org.apache.inlong.dataproxy.channel.FailoverChannelSelector
-agent1.sources.tcp-source.selector.master = ch-msg5 ch-msg6 ch-msg7 ch-msg8 ch-msg9
-agent1.sources.tcp-source.selector.transfer = ch-msg5 ch-msg6 ch-msg7 ch-msg8 ch-msg9
-agent1.sources.tcp-source.selector.fileMetric = ch-back
-agent1.sources.tcp-source.selector.slaMetric = ch-msg10
-agent1.sources.tcp-source.metric-recovery-path=FLUME_HOME/metric
-agent1.sources.tcp-source.metric-agent-port=8003
-agent1.sources.tcp-source.metric-cache-size=10000
-agent1.sources.tcp-source.set=10
-agent1.sources.tcp-source.old-metric-on=true
-agent1.sources.tcp-source.new-metric-on=true
-agent1.sources.tcp-source.metric_topic_prefix=manager_tmertic
-
-agent1.channels.ch-back.type = memory
-agent1.channels.ch-back.capacity = 10000000
-agent1.channels.ch-back.keep-alive = 0
-agent1.channels.ch-back.transactionCapacity = 200
-
-agent1.channels.ch-msg1.type = memory
-agent1.channels.ch-msg1.capacity = 200000
-agent1.channels.ch-msg1.keep-alive = 0
-agent1.channels.ch-msg1.transactionCapacity = 200
-
-agent1.channels.ch-msg2.type = memory
-agent1.channels.ch-msg2.capacity = 200000
-agent1.channels.ch-msg2.keep-alive = 0
-agent1.channels.ch-msg2.transactionCapacity = 200
-
-agent1.channels.ch-msg3.type = memory
-agent1.channels.ch-msg3.capacity = 200000
-agent1.channels.ch-msg3.keep-alive = 0
-agent1.channels.ch-msg3.transactionCapacity = 200
-
-agent1.channels.ch-msg5.type = file
-agent1.channels.ch-msg5.capacity = 100000000
-agent1.channels.ch-msg5.maxFileSize = 1073741824
-agent1.channels.ch-msg5.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg5.checkpointDir = FLUME_HOME/file/ch-msg5/check
-agent1.channels.ch-msg5.dataDirs = FLUME_HOME/file/ch-msg5/data
-agent1.channels.ch-msg5.fsyncPerTransaction = false
-agent1.channels.ch-msg5.fsyncInterval = 5
-
-agent1.channels.ch-msg6.type = file
-agent1.channels.ch-msg6.capacity = 100000000
-agent1.channels.ch-msg6.maxFileSize = 1073741824
-agent1.channels.ch-msg6.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg6.checkpointDir = FLUME_HOME/file/ch-msg6/check
-agent1.channels.ch-msg6.dataDirs = FLUME_HOME/file/ch-msg6/data
-agent1.channels.ch-msg6.fsyncPerTransaction = false
-agent1.channels.ch-msg6.fsyncInterval = 5
-
-agent1.channels.ch-msg7.type = file
-agent1.channels.ch-msg7.capacity = 100000000
-agent1.channels.ch-msg7.maxFileSize = 1073741824
-agent1.channels.ch-msg7.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg7.checkpointDir = FLUME_HOME/file/ch-msg7/check
-agent1.channels.ch-msg7.dataDirs = FLUME_HOME/file/ch-msg7/data
-agent1.channels.ch-msg7.fsyncPerTransaction = false
-agent1.channels.ch-msg7.fsyncInterval = 5
-
-agent1.channels.ch-msg8.type = file
-agent1.channels.ch-msg8.capacity = 100000000
-agent1.channels.ch-msg8.maxFileSize = 1073741824
-agent1.channels.ch-msg8.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg8.checkpointDir = FLUME_HOME/file/ch-msg8/check
-agent1.channels.ch-msg8.dataDirs = FLUME_HOME/file/ch-msg8/data
-agent1.channels.ch-msg8.fsyncPerTransaction = false
-agent1.channels.ch-msg8.fsyncInterval = 5
-
-agent1.channels.ch-msg9.type = file
-agent1.channels.ch-msg9.capacity = 100000000
-agent1.channels.ch-msg9.maxFileSize = 1073741824
-agent1.channels.ch-msg9.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg9.checkpointDir = FLUME_HOME/file/ch-msg9/check
-agent1.channels.ch-msg9.dataDirs = FLUME_HOME/file/ch-msg9/data
-agent1.channels.ch-msg9.fsyncPerTransaction = false
-agent1.channels.ch-msg9.fsyncInterval = 5
-
-agent1.channels.ch-msg10.type = file
-agent1.channels.ch-msg10.capacity = 100000000
-agent1.channels.ch-msg10.maxFileSize = 1073741824
-agent1.channels.ch-msg10.minimumRequiredSpace = 1073741824
-agent1.channels.ch-msg10.checkpointDir = FLUME_HOME/file/ch-msg10/check
-agent1.channels.ch-msg10.dataDirs = FLUME_HOME/file/ch-msg10/data
-agent1.channels.ch-msg10.fsyncPerTransaction = false
-agent1.channels.ch-msg10.fsyncInterval = 5
-
-agent1.sinks.meta-sink-more1.channel = ch-msg1
-agent1.sinks.meta-sink-more1.type = org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-more2.channel = ch-more2
-agent1.sinks.meta-sink-more2.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-more3.channel = ch-more3
-agent1.sinks.meta-sink-more3.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-most1.channel = ch-most1
-agent1.sinks.meta-sink-most1.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-most2.channel = ch-most2
-agent1.sinks.meta-sink-most2.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-most3.channel = ch-most3
-agent1.sinks.meta-sink-most3.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-msg1.channel = ch-msg1
-agent1.sinks.meta-sink-msg1.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-msg2.channel = ch-msg2
-agent1.sinks.meta-sink-msg2.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-msg3.channel = ch-msg3
-agent1.sinks.meta-sink-msg3.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-msg8.channel = ch-msg1
-agent1.sinks.meta-sink-msg8.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.meta-sink-msg8.channel = ch-back
-agent1.sinks.meta-sink-msg8.type =org.apache.inlong.dataproxy.sink.TubeSink
-
-agent1.sinks.pulsar-sink-msg1.channel = ch-msg1
-agent1.sinks.pulsar-sink-msg1.type = org.apache.inlong.dataproxy.sink.PulsarSink
-
-agent1.sources = tcp-source
-agent1.channels = ch-back ch-msg1 ch-msg2 ch-msg3 ch-msg5 ch-msg6 ch-msg7 ch-msg8 ch-msg9 ch-msg10
-agent1.sinks = meta-sink-msg1 meta-sink-msg2 meta-sink-msg3 meta-sink-msg5 meta-sink-msg6 meta-sink-msg7 meta-sink-msg8 meta-sink-transfer
-
diff --git a/inlong-dataproxy/dataproxy-source/src/main/resources/log4j2.xml b/inlong-dataproxy/dataproxy-source/src/main/resources/log4j2.xml
new file mode 100644
index 0000000..d4f70d4
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/resources/log4j2.xml
@@ -0,0 +1,132 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<configuration status="WARN" monitorInterval="30">
+ <Properties>
+ <property name="basePath">${sys:dataproxy.log.path}</property>
+ <property name="log_pattern">%d{yyyy-MM-dd HH:mm:ss.SSS} -%5p ${PID:-} [%15.15t] %-30.30C{1.} : %m%n</property>
+ <property name="every_file_size">1G</property>
+ <property name="output_log_level">DEBUG</property>
+ <property name="rolling_max">50</property>
+ <property name="info_fileName">${basePath}/info.log</property>
+ <property name="info_filePattern">${basePath}/info-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="info_max">10</property>
+ <property name="debug_fileName">${basePath}/debug.log</property>
+ <property name="debug_filePattern">${basePath}/debug-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="debug_max">10</property>
+ <property name="warn_fileName">${basePath}/warn.log</property>
+ <property name="warn_filePattern">${basePath}/warn-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="warn_max">10</property>
+ <property name="error_fileName">${basePath}/error.log</property>
+ <property name="error_filePattern">${basePath}/error-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="error_max">10</property>
+ <property name="console_print_level">DEBUG</property>
+ <property name="index_fileName">${basePath}/index.log</property>
+ <property name="index_filePattern">${basePath}/index-%d{yyyy-MM-dd}-%i.log.gz</property>
+ <property name="monitors_fileName">${basePath}/monitors.log</property>
+ <property name="monitors_filePattern">${basePath}/monitors-%d{yyyy-MM-dd}-%i.log.gz</property>
+ </Properties>
+
+ <appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <ThresholdFilter level="${console_print_level}" onMatch="ACCEPT" onMismatch="DENY"/>
+ <PatternLayout pattern="${log_pattern}"/>
+ </Console>
+
+ <RollingFile name="DebugFile" fileName="${debug_fileName}" filePattern="${debug_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${debug_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="DEBUG" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="InfoFile" fileName="${info_fileName}" filePattern="${info_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${info_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="IndexFile" fileName="${index_fileName}" filePattern="${index_filePattern}">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %m%n"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${info_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="MonitorFile" fileName="${monitors_fileName}" filePattern="${monitors_filePattern}">
+ <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %m%n"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${info_max}" />
+ <Filters>
+ <ThresholdFilter level="WARN" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="WarnFile" fileName="${warn_fileName}" filePattern="${warn_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${warn_max}" />
+ <Filters>
+ <ThresholdFilter level="ERROR" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="WARN" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+
+ <RollingFile name="ErrorFile" fileName="${error_fileName}" filePattern="${error_filePattern}">
+ <PatternLayout pattern="${log_pattern}"/>
+ <SizeBasedTriggeringPolicy size="${every_file_size}"/>
+ <DefaultRolloverStrategy max="${error_max}" />
+ <Filters>
+ <ThresholdFilter level="FATAL" onMatch="DENY" onMismatch="NEUTRAL"/>
+ <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY"/>
+ </Filters>
+ </RollingFile>
+ </appenders>
+
+ <loggers>
+ <logger name="org.apache.inlong.common.monitor.MonitorIndexExt" level="info" additivity="false">
+ <appender-ref ref="MonitorFile"/>
+ </logger>
+ <logger name="org.apache.inlong.common.monitor.MonitorIndex" level="info" additivity="false">
+ <appender-ref ref="IndexFile"/>
+ </logger>
+ <logger name="org.apache.pulsar" level="info" additivity="false">
+ <appender-ref ref="InfoFile"/>
+ </logger>
+ <root level="${output_log_level}">
+ <appender-ref ref="Console"/>
+ <appender-ref ref="DebugFile"/>
+ <appender-ref ref="InfoFile"/>
+ <appender-ref ref="WarnFile"/>
+ <appender-ref ref="ErrorFile"/>
+ </root>
+ </loggers>
+</configuration>
\ No newline at end of file