You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by al...@apache.org on 2021/10/22 04:43:27 UTC
[incubator-inlong] branch master updated: [INLONG-1330][DataProxy]
Support pulsar sink (#1705)
This is an automated email from the ASF dual-hosted git repository.
aloyszhang 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 219370f [INLONG-1330][DataProxy] Support pulsar sink (#1705)
219370f is described below
commit 219370fd031107f0ab08b31cff7ac55e8f43b58f
Author: baomingyu <ba...@163.com>
AuthorDate: Fri Oct 22 12:43:19 2021 +0800
[INLONG-1330][DataProxy] Support pulsar sink (#1705)
* add pulsar sink feature
* support mult pulsar cluster
* modify check style
* modify check style
* modify check style
---
inlong-dataproxy/dataproxy-source/pom.xml | 6 +
.../dataproxy/base/HighPriorityThreadFactory.java | 53 ++
.../apache/inlong/dataproxy/sink/PulsarSink.java | 778 +++++++++++++++++++++
.../sink/pulsar/CreatePulsarClientCallBack.java | 25 +
.../dataproxy/sink/pulsar/PulsarClientService.java | 347 +++++++++
.../dataproxy/sink/pulsar/SendMessageCallBack.java | 27 +
.../inlong/dataproxy/source/SimpleTcpSource.java | 2 +-
.../utils/FailoverChannelProcessorHolder.java | 32 +
.../apache/inlong/dataproxy/utils/LogCounter.java | 50 ++
.../inlong/dataproxy/utils/MonitorIndex.java | 127 ++++
.../inlong/dataproxy/utils/MonitorIndexExt.java | 125 ++++
.../dataproxy-source/src/main/resources/flume.conf | 10 +
.../inlong/dataproxy/sink/TestPulsarSink.java | 96 +++
inlong-dataproxy/pom.xml | 13 +
14 files changed, 1690 insertions(+), 1 deletion(-)
diff --git a/inlong-dataproxy/dataproxy-source/pom.xml b/inlong-dataproxy/dataproxy-source/pom.xml
index 301a4fb..b973a20 100644
--- a/inlong-dataproxy/dataproxy-source/pom.xml
+++ b/inlong-dataproxy/dataproxy-source/pom.xml
@@ -33,6 +33,7 @@
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
+ <guava.version>19.0</guava.version>
</properties>
<dependencies>
@@ -42,6 +43,11 @@
<version>${project.version}</version>
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ <version>${guava.version}</version>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/HighPriorityThreadFactory.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/HighPriorityThreadFactory.java
new file mode 100644
index 0000000..07557fc
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/base/HighPriorityThreadFactory.java
@@ -0,0 +1,53 @@
+/*
+ * 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.base;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class HighPriorityThreadFactory implements ThreadFactory {
+ static final AtomicInteger poolNumber = new AtomicInteger(1);
+ final AtomicInteger threadNumber;
+ final ThreadGroup group;
+ final String namePrefix;
+ final boolean isDaemon;
+
+ public HighPriorityThreadFactory() {
+ this("pool");
+ }
+
+ public HighPriorityThreadFactory(String name) {
+ this(name, false);
+ }
+
+ public HighPriorityThreadFactory(String prefix, boolean daemon) {
+ this.threadNumber = new AtomicInteger(1);
+ SecurityManager s = System.getSecurityManager();
+ this.group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
+ this.namePrefix = prefix + "-thread-" + poolNumber.getAndIncrement();
+ this.isDaemon = daemon;
+ }
+
+ public Thread newThread(Runnable r) {
+ Thread t = new Thread(this.group, r, this.namePrefix + this.threadNumber.getAndIncrement(), 0L);
+ t.setDaemon(this.isDaemon);
+ t.setPriority(10);
+ return t;
+ }
+}
+
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
new file mode 100644
index 0000000..2a180e8
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/PulsarSink.java
@@ -0,0 +1,778 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.RateLimiter;
+import org.apache.inlong.dataproxy.base.HighPriorityThreadFactory;
+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;
+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.instrumentation.SinkCounter;
+import org.apache.flume.sink.AbstractSink;
+import org.apache.inlong.dataproxy.consts.AttributeConstants;
+import org.apache.inlong.dataproxy.consts.ConfigConstants;
+import org.apache.inlong.dataproxy.config.ConfigManager;
+import org.apache.inlong.dataproxy.config.holder.ConfigUpdateCallback;
+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.utils.FailoverChannelProcessorHolder;
+import org.apache.inlong.dataproxy.utils.LogCounter;
+import org.apache.inlong.dataproxy.utils.MonitorIndex;
+import org.apache.inlong.dataproxy.utils.MonitorIndexExt;
+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.NotConnectedException;
+import org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError;
+import org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException;
+import org.jboss.netty.handler.codec.frame.TooLongFrameException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * use pulsarSink need adding such config, if these ara not config in flume.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'
+ * pulsar_server_url_list (*): value is pulsar broker url , like this 'pulsar://127.0.0.1:6650'
+ * send_timeout_MILL: send message timeout, unit is millisecond, default value is 30000 (mean 30s)
+ * stat_interval_sec: stat info will be made period time , unit is second, default value is 60s
+ * thread-num: sink thread num. default value is 8
+ * client-id-cache: whether use cache in client, default value is true
+ * max_pending_messages: default value is 10000
+ * max_batching_messages: default value is 1000
+ * enable_batch: default is true
+ * block_if_queue_full: default is true
+ */
+public class PulsarSink extends AbstractSink implements Configurable,
+ SendMessageCallBack, CreatePulsarClientCallBack {
+
+ private static final Logger logger = LoggerFactory.getLogger(PulsarSink.class);
+
+ /*
+ * properties for header info
+ */
+ private static String TOPIC = "topic";
+
+ /*
+ * default value
+ */
+ private static int BAD_EVENT_QUEUE_SIZE = 10000;
+ private static int BATCH_SIZE = 10000;
+ 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;
+
+ /*
+ * properties for stat
+ */
+ private static String LOG_EVERY_N_EVENTS = "log-every-n-events";
+
+ private static String CLIENT_ID_CACHE = "client_id_cache";
+
+ private static String RETRY_CNT = "retry_currentSuccSendedCnt";
+
+ private static String STAT_INTERVAL_SEC = "stat_interval_sec";
+ /*
+ * for log
+ */
+ private Integer logEveryNEvents;
+ private long diskIORatePerSec;
+ private RateLimiter diskRateLimiter;
+
+ /*
+ * for stat
+ */
+ 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 static int retryCnt = DEFAULT_RETRY_CNT;
+ private boolean overflow = false;
+
+ private LinkedBlockingQueue<EventStat> resendQueue;
+
+ private int maxMonitorCnt = 300000;
+
+ private int statIntervalSec = DEFAULT_STAT_INTERVAL_SEC;
+ private long logCounter = 0;
+
+ private final AtomicLong currentInFlightCount = new AtomicLong(0);
+
+ private boolean clientIdCache = false;
+
+ 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
+ */
+ private volatile boolean canTake = false;
+
+ /*
+ * 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 static final String SINK_THREAD_NUM = "thread-num";
+ private static int EVENT_QUEUE_SIZE = 1000;
+ private int threadNum;
+
+
+ /*
+ * send thread pool
+ */
+ private Thread[] sinkThreadPool;
+ private LinkedBlockingQueue<Event> eventQueue;
+
+ private static final String SEPARATOR = "#";
+ private boolean isNewMetricOn = true;
+ private MonitorIndex monitorIndex;
+ private MonitorIndexExt monitorIndexExt;
+ private SinkCounter sinkCounter;
+ private ConfigManager configManager;
+ private Map<String, String> topicProperties;
+
+ private PulsarClientService pulsarClientService;
+
+ private static final Long PRINT_INTERVAL = 30L;
+
+ private static final PulsarPerformanceTask pulsarPerformanceTask = new PulsarPerformanceTask();
+
+ private static ScheduledExecutorService scheduledExecutorService = Executors
+ .newScheduledThreadPool(1, new HighPriorityThreadFactory("pulsarPerformance-Printer-thread"));
+
+ private static final LoadingCache<String, Long> agentIdCache = CacheBuilder.newBuilder()
+ .concurrencyLevel(4 * 8).initialCapacity(500).expireAfterAccess(30, TimeUnit.SECONDS)
+ .build(new CacheLoader<String, Long>() {
+ //默认的数据加载实现，当调用get取值时，如果key没有对应的值，则调用此方法进行加载。
+ @Override
+ public Long load(String key) {
+ return System.currentTimeMillis();
+ }
+ });
+
+ static {
+ /*
+ * stat pulsar performance
+ */
+ System.out.println("pulsarPerformanceTask!!!!!!");
+ scheduledExecutorService.scheduleWithFixedDelay(pulsarPerformanceTask, 0L,
+ PRINT_INTERVAL, TimeUnit.SECONDS);
+ }
+
+ public PulsarSink() {
+ super();
+ logger.debug("new instance of PulsarSink!");
+ }
+
+ public void configure(Context context) {
+ logger.info("PulsarSink started and context = {}", context.toString());
+ isNewMetricOn = context.getBoolean("new-metric-on", true);
+ maxMonitorCnt = context.getInteger("max-monitor-cnt",300000);
+
+ configManager = ConfigManager.getInstance();
+ topicProperties = configManager.getTopicProperties();
+ configManager.getTopicConfig().addUpdateCallback(new ConfigUpdateCallback() {
+ @Override
+ public void update() {
+ diffSetPublish(new HashSet<String>(topicProperties.values()),
+ new HashSet<String>(configManager.getTopicProperties().values()));
+ }
+ });
+
+ retryCnt = context.getInteger(RETRY_CNT, DEFAULT_RETRY_CNT);
+ logger.debug(this.getName() + " " + RETRY_CNT + " " + retryCnt);
+
+ statIntervalSec = context.getInteger(STAT_INTERVAL_SEC, DEFAULT_STAT_INTERVAL_SEC);
+ logger.debug(this.getName() + " " + STAT_INTERVAL_SEC + " " + statIntervalSec);
+
+ logEveryNEvents = context.getInteger(LOG_EVERY_N_EVENTS, DEFAULT_LOG_EVERY_N_EVENTS);
+ logger.debug(this.getName() + " " + LOG_EVERY_N_EVENTS + " " + logEveryNEvents);
+ Preconditions.checkArgument(logEveryNEvents > 0, "logEveryNEvents must be > 0");
+ Preconditions.checkArgument(statIntervalSec >= 0, "statIntervalSec must be >= 0");
+
+ clientIdCache = context.getBoolean(CLIENT_ID_CACHE, clientIdCache);
+
+ resendQueue = new LinkedBlockingQueue<EventStat>(BAD_EVENT_QUEUE_SIZE);
+
+ String sinkThreadNum = context.getString(SINK_THREAD_NUM, "4");
+ threadNum = Integer.parseInt(sinkThreadNum);
+ Preconditions.checkArgument(threadNum > 0, "threadNum must be > 0");
+ sinkThreadPool = new Thread[threadNum];
+ eventQueue = new LinkedBlockingQueue<Event>(EVENT_QUEUE_SIZE);
+
+ diskIORatePerSec = context.getLong("disk-io-rate-per-sec",0L);
+ if (diskIORatePerSec != 0) {
+ diskRateLimiter = RateLimiter.create(diskIORatePerSec);
+ }
+ pulsarClientService = new PulsarClientService(context);
+
+ if (sinkCounter == null) {
+ sinkCounter = new SinkCounter(getName());
+ }
+ }
+
+ private void initTopicSet(Set<String> topicSet) throws Exception {
+ long startTime = System.currentTimeMillis();
+ if (topicSet != null) {
+ for (String topic : topicSet) {
+ pulsarClientService.initTopicProducer(topic);
+ }
+ }
+ logger.info(getName() + " initTopicSet cost: "
+ + (System.currentTimeMillis() - startTime) + "ms");
+ logger.info(getName() + " producer is ready for topics : "
+ + pulsarClientService.getProducerInfoMap().keySet());
+ }
+
+ /**
+ * When topic.properties is re-enabled, the producer update is triggered
+ * @param originalSet
+ * @param endSet
+ */
+ public void diffSetPublish(Set<String> originalSet, Set<String> endSet) {
+
+ boolean changed = false;
+ for (String s : endSet) {
+ if (!originalSet.contains(s)) {
+ changed = true;
+ try {
+ pulsarClientService.initTopicProducer(s);
+ } catch (Exception e) {
+ logger.error("Get producer failed!", e);
+ }
+ }
+ }
+ if (changed) {
+ logger.info("topics.properties has changed, trigger diff publish for {}", getName());
+ topicProperties = configManager.getTopicProperties();
+ }
+ }
+
+ @Override
+ public void start() {
+ logger.info("pulsar sink starting...");
+ sinkCounter.start();
+ pulsarClientService.initCreateConnection(this);
+ if (statIntervalSec > 0) {
+ /*
+ * switch for lots of metrics
+ */
+ if (isNewMetricOn) {
+ monitorIndex = new MonitorIndex("Sink", statIntervalSec, maxMonitorCnt);
+ }
+ monitorIndexExt = new MonitorIndexExt("TDBus_monitors#" + this.getName(),
+ statIntervalSec, maxMonitorCnt);
+ }
+
+ super.start();
+ this.canSend = true;
+ this.canTake = true;
+
+ try {
+ initTopicSet(new HashSet<String>(topicProperties.values()));
+ } catch (Exception e) {
+ logger.info("meta sink start publish topic fail.",e);
+ }
+
+ for (int i = 0; i < sinkThreadPool.length; i++) {
+ sinkThreadPool[i] = new Thread(new SinkTask(), getName()
+ + "_pulsar_sink_sender-"
+ + i);
+ sinkThreadPool[i].start();
+ }
+ logger.debug("meta sink started");
+ }
+
+ @Override
+ public void stop() {
+ logger.info("pulsar sink stopping");
+ pulsarClientService.close();
+ this.canTake = false;
+ int waitCount = 0;
+ while (eventQueue.size() != 0 && waitCount++ < 10) {
+ try {
+ Thread.currentThread().sleep(1000);
+ } catch (InterruptedException e) {
+ logger.info("Stop thread has been interrupt!");
+ break;
+ }
+ }
+ this.canSend = false;
+ if (statIntervalSec > 0) {
+ try {
+ monitorIndex.shutDown();
+ } catch (Exception e) {
+ logger.warn("statrunner interrupted");
+ }
+ }
+ if (sinkThreadPool != null) {
+ for (Thread thread : sinkThreadPool) {
+ if (thread != null) {
+ thread.interrupt();
+ }
+ }
+ sinkThreadPool = null;
+ }
+ super.stop();
+ if (!scheduledExecutorService.isShutdown()) {
+ scheduledExecutorService.shutdown();
+ }
+ sinkCounter.stop();
+ logger.debug("pulsar sink stopped. Metrics:{}", sinkCounter);
+ }
+
+ @Override
+ public Status process() throws EventDeliveryException {
+ if (!this.canTake) {
+ return Status.BACKOFF;
+ }
+
+ Status status = Status.READY;
+ Channel channel = getChannel();
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ try {
+ Event event = channel.take();
+ if (event != null) {
+ if (diskRateLimiter != null) {
+ diskRateLimiter.acquire(event.getBody().length);
+ }
+ if (!eventQueue.offer(event, 3 * 1000, TimeUnit.MILLISECONDS)) {
+ 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();
+ } else {
+ tx.commit();
+ }
+ } else {
+ status = Status.BACKOFF;
+ tx.commit();
+ }
+ } catch (Throwable t) {
+ logger.error("Process event failed!" + this.getName(), t);
+ try {
+ tx.rollback();
+ } catch (Throwable e) {
+ logger.error("metasink transaction rollback exception",e);
+
+ }
+ } finally {
+ tx.close();
+ }
+ return status;
+ }
+
+ private void editStatistic(final Event event, String keyPostfix, String msgId) {
+ String topic = "";
+ String tid = "";
+ String nodeIp = null;
+ if (event != null) {
+ if (event.getHeaders().containsKey(TOPIC)) {
+ topic = event.getHeaders().get(TOPIC);
+ }
+ if (event.getHeaders().containsKey(AttributeConstants.INTERFACE_ID)) {
+ tid = event.getHeaders().get(AttributeConstants.INTERFACE_ID);
+ } else if (event.getHeaders().containsKey(AttributeConstants.INAME)) {
+ tid = event.getHeaders().get(AttributeConstants.INAME);
+ }
+
+ /*
+ * Compatible agent
+ */
+ if (event.getHeaders().containsKey("ip")) {
+ event.getHeaders().put(ConfigConstants.REMOTE_IP_KEY, event.getHeaders().get("ip"));
+ event.getHeaders().remove("ip");
+ }
+
+ /*
+ * Compatible agent
+ */
+ if (event.getHeaders().containsKey("time")) {
+ event.getHeaders().put(AttributeConstants.DATA_TIME,event.getHeaders().get("time"));
+ event.getHeaders().remove("time");
+ }
+
+ if (event.getHeaders().containsKey(ConfigConstants.REMOTE_IP_KEY)) {
+ nodeIp = event.getHeaders().get(ConfigConstants.REMOTE_IP_KEY);
+ if (event.getHeaders().containsKey(ConfigConstants.REMOTE_IDC_KEY)) {
+
+ if (nodeIp != null) {
+ nodeIp = nodeIp.split(":")[0];
+ }
+
+ long tMsgCounterL = 1L;
+ /*
+ * msg counter
+ */
+ if (event.getHeaders().containsKey(ConfigConstants.MSG_COUNTER_KEY)) {
+ tMsgCounterL = Integer.parseInt(event.getHeaders()
+ .get(ConfigConstants.MSG_COUNTER_KEY));
+ }
+
+ /*
+ * SINK_INTF#metasink1#topic#tid#clientIp#busIP#pkgTime#successCnt#packcnt
+ * #packsize#failCnt
+ */
+ StringBuilder newbase = new StringBuilder();
+ newbase.append(this.getName()).append(SEPARATOR).append(topic).append(SEPARATOR)
+ .append(tid).append(SEPARATOR).append(nodeIp)
+ .append(SEPARATOR).append(NetworkUtils.getLocalIp())
+ .append(SEPARATOR).append(msgId).append(SEPARATOR)
+ .append(event.getHeaders().get(ConfigConstants.PKG_TIME_KEY));
+
+ long messageSize = event.getBody().length;
+
+ if (event.getHeaders().get(ConfigConstants.TOTAL_LEN) != null) {
+ messageSize = Long.parseLong(event.getHeaders().get(ConfigConstants.TOTAL_LEN));
+ }
+
+ if (keyPostfix != null && !keyPostfix.equals("")) {
+ monitorIndex.addAndGet(new String(newbase), 0, 0, 0, (int) tMsgCounterL);
+ if (logPrinterB.shouldPrint()) {
+ logger.warn("error cannot send event, {} event size is {}", topic, messageSize);
+ }
+ } else {
+ monitorIndex.addAndGet(new String(newbase), (int) tMsgCounterL, 1, messageSize, 0);
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void handleCreateClientSuccess(String url) {
+ logger.info("createConnection success for url = {}", url);
+ sinkCounter.incrementConnectionCreatedCount();
+ }
+
+ @Override
+ public void handleCreateClientException(String url) {
+ logger.info("createConnection has exception for url = {}", url);
+ sinkCounter.incrementConnectionFailedCount();
+ }
+
+ @Override
+ public void handleMessageSendSuccess(Object result, EventStat eventStat) {
+ /*
+ * Statistics pulsar performance
+ */
+ totalPulsarSuccSendCnt.incrementAndGet();
+ totalPulsarSuccSendSize.addAndGet(eventStat.getEvent().getBody().length);
+ /*
+ *add to sinkCounter
+ */
+ sinkCounter.incrementEventDrainSuccessCount();
+ currentInFlightCount.decrementAndGet();
+ currentSuccessSendCnt.incrementAndGet();
+ long nowCnt = currentSuccessSendCnt.get();
+ long oldCnt = lastSuccessSendCnt.get();
+ if (nowCnt % logEveryNEvents == 0 && nowCnt != lastSuccessSendCnt.get()) {
+ lastSuccessSendCnt.set(nowCnt);
+ t2 = System.currentTimeMillis();
+ logger.info("metasink {}, succ put {} events to pulsar,"
+ + " in the past {} millsec", new Object[] {
+ getName(), (nowCnt - oldCnt), (t2 - t1)
+ });
+ t1 = t2;
+ }
+ monitorIndexExt.incrementAndGet("METASINK_SUCCESS");
+ editStatistic(eventStat.getEvent(), null, result.toString());
+ }
+
+ @Override
+ public void handleMessageSendException(EventStat eventStat, Object e) {
+ monitorIndexExt.incrementAndGet("METASINK_EXP");
+ if (e instanceof TooLongFrameException) {
+ PulsarSink.this.overflow = true;
+ } else if (e instanceof ProducerQueueIsFullError) {
+ PulsarSink.this.overflow = true;
+ } else if (!(e instanceof AlreadyClosedException
+ || e instanceof NotConnectedException
+ || e instanceof TopicTerminatedException)) {
+ if (logPrinterB.shouldPrint()) {
+ logger.error("Send failed!{}{}", getName(), e);
+ }
+ if (eventStat.getRetryCnt() == 0) {
+ editStatistic(eventStat.getEvent(), "failure", "");
+ }
+ }
+ eventStat.incRetryCnt();
+ resendEvent(eventStat, true);
+ }
+
+ /**
+ * Resend the data and store the data in the memory cache.
+ * @param es
+ * @param isDecrement
+ */
+ private void resendEvent(EventStat es, boolean isDecrement) {
+ try {
+ if (isDecrement) {
+ currentInFlightCount.decrementAndGet();
+ }
+ if (es == null || es.getEvent() == null) {
+ return;
+ }
+ /*
+ * If the failure requires retransmission to pulsar,
+ * the sid needs to be removed before retransmission.
+ */
+ if (clientIdCache) {
+ String clientId = es.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());
+ if (logPrinterC.shouldPrint()) {
+ logger.error(getName() + " Channel --> pulsar --> ResendQueue(full) "
+ + "-->FailOverChannelProcessor(current code point), "
+ + "Resend queue is full,Check if pulsar server or network is ok.");
+ }
+ }
+ } catch (Throwable throwable) {
+ 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);
+ }
+ }
+ }
+
+ static class PulsarPerformanceTask implements Runnable {
+ @Override
+ public void run() {
+ try {
+ if (totalPulsarSuccSendSize.get() != 0) {
+ logger.info("Total pulsar performance tps :"
+ + totalPulsarSuccSendCnt.get() / PRINT_INTERVAL
+ + "/s, avg msg size:"
+ + totalPulsarSuccSendSize.get() / totalPulsarSuccSendCnt.get()
+ + ",print every " + PRINT_INTERVAL + " seconds");
+ /*
+ * totalpulsarSuccSendCnt represents the number of packets
+ */
+ totalPulsarSuccSendCnt.set(0);
+ totalPulsarSuccSendSize.set(0);
+ }
+
+ } catch (Exception e) {
+ logger.info("pulsarPerformanceTask error", e);
+ }
+ }
+ }
+
+ class SinkTask implements Runnable {
+ @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);
+ }
+ }
+ 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 (clientIdCache && clientId != null) {
+ hasKey = agentIdCache.asMap().containsKey(clientId);
+ }
+
+ if (clientIdCache && clientId != null && hasKey) {
+ agentIdCache.put(clientId, System.currentTimeMillis());
+ if (logPrinterA.shouldPrint()) {
+ logger.info("{} agent package {} existed,just discard.",
+ getName(), clientId);
+ }
+ } else {
+ if (clientId != null) {
+ agentIdCache.put(clientId, System.currentTimeMillis());
+ }
+ boolean sendResult = pulsarClientService.sendMessage(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/CreatePulsarClientCallBack.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/CreatePulsarClientCallBack.java
new file mode 100644
index 0000000..d2c6c67
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/CreatePulsarClientCallBack.java
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+public interface CreatePulsarClientCallBack {
+
+ void handleCreateClientSuccess(String url);
+
+ void handleCreateClientException(String url);
+}
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
new file mode 100644
index 0000000..7aae1b6
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/PulsarClientService.java
@@ -0,0 +1,347 @@
+/*
+ * 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.base.Preconditions;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.FlumeException;
+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.utils.LogCounter;
+import org.apache.inlong.dataproxy.utils.NetworkUtils;
+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.impl.MessageIdImpl;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PulsarClientService {
+
+ private static final Logger logger = LoggerFactory.getLogger(PulsarClientService.class);
+
+ private static final LogCounter logPrinterA = new LogCounter(10, 100000, 60 * 1000);
+
+ /*
+ * properties key for pulsar client
+ */
+ private static String PULSAR_SERVER_URL_LIST = "pulsar_server_url_list";
+
+ /*
+ * properties key pulsar producer
+ */
+ private static String SEND_TIMEOUT = "send_timeout_mill";
+ private static String CLIENT_TIMEOUT = "client_timeout_second";
+ private static String ENABLE_BATCH = "enable_batch";
+ private static String BLOCK_IF_QUEUE_FULL = "block_if_queue_full";
+ private static String MAX_PENDING_MESSAGES = "max_pending_messages";
+ private static String MAX_BATCHING_MESSAGES = "max_batching_messages";
+ private static String RETRY_INTERVAL_WHEN_SEND_ERROR_MILL = "retry_interval_when_send_error_ms";
+
+ private static int DEFAULT_SEND_TIMEOUT_MILL = 30 * 1000;
+ private static int DEFAULT_CLIENT_TIMEOUT_SECOND = 30;
+ private static long DEFAULT_RETRY_INTERVAL_WHEN_SEND_ERROR_MILL = 30 * 1000L;
+ private static boolean DEFAULT_ENABLE_BATCH = true;
+ private static boolean DEFAULT_BLOCK_IF_QUEUE_FULL = true;
+ private static int DEFAULT_MAX_PENDING_MESSAGES = 10000;
+ private static int DEFAULT_MAX_BATCHING_MESSAGES = 1000;
+
+ /*
+ * for pulsar client
+ */
+ private String[] pulsarServerUrls;
+
+ /*
+ * for producer
+ */
+ private Integer sendTimeout; // in millsec
+ private Integer clientTimeout;
+ private boolean enableBatch = true;
+ private boolean blockIfQueueFull = true;
+ private int maxPendingMessages = 10000;
+ private int maxBatchingMessages = 1000;
+ private long retryIntervalWhenSendMsgError = 30 * 1000L;
+ public Map<String, List<TopicProducerInfo>> producerInfoMap;
+ public Map<String, AtomicLong> topicSendIndexMap;
+ public List<PulsarClient> pulsarClients;
+
+ private String localIp = "127.0.0.1";
+
+ public PulsarClientService(Context context) {
+
+ String pulsarServerUrlList = context.getString(PULSAR_SERVER_URL_LIST);
+ Preconditions.checkState(pulsarServerUrlList != null, "No pulsar server url specified");
+ pulsarServerUrls = pulsarServerUrlList.split("\\|");
+ Preconditions.checkState(pulsarServerUrls != null && pulsarServerUrls.length > 0, "No "
+ + "pulsar server url config");
+ sendTimeout = context.getInteger(SEND_TIMEOUT, DEFAULT_SEND_TIMEOUT_MILL);
+ retryIntervalWhenSendMsgError = context.getLong(RETRY_INTERVAL_WHEN_SEND_ERROR_MILL,
+ DEFAULT_RETRY_INTERVAL_WHEN_SEND_ERROR_MILL);
+ clientTimeout = context.getInteger(CLIENT_TIMEOUT, DEFAULT_CLIENT_TIMEOUT_SECOND);
+ logger.debug("PulsarClientService " + SEND_TIMEOUT + " " + sendTimeout);
+ Preconditions.checkArgument(sendTimeout > 0, "sendTimeout must be > 0");
+
+ enableBatch = context.getBoolean(ENABLE_BATCH, DEFAULT_ENABLE_BATCH);
+ blockIfQueueFull = context.getBoolean(BLOCK_IF_QUEUE_FULL, DEFAULT_BLOCK_IF_QUEUE_FULL);
+ maxPendingMessages = context.getInteger(MAX_PENDING_MESSAGES, DEFAULT_MAX_PENDING_MESSAGES);
+ maxBatchingMessages = context.getInteger(MAX_BATCHING_MESSAGES, DEFAULT_MAX_BATCHING_MESSAGES);
+ producerInfoMap = new ConcurrentHashMap<String, List<TopicProducerInfo>>();
+ topicSendIndexMap = new ConcurrentHashMap<String, AtomicLong>();
+ localIp = NetworkUtils.getLocalIp();
+
+ }
+
+ public void initCreateConnection(CreatePulsarClientCallBack callBack) {
+ try {
+ createConnection(callBack);
+ } catch (FlumeException e) {
+ logger.error("Unable to create pulsar client" + ". Exception follows.", e);
+ close();
+ }
+ }
+
+ public boolean sendMessage(String topic, Event event,
+ SendMessageCallBack sendMessageCallBack, EventStat es) {
+ TopicProducerInfo producer = null;
+ try {
+ producer = getProducer(topic);
+ } catch (Exception e) {
+ 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);
+ }
+ }
+ /*
+ * If the producer is a null value,\ it means that the topic is not yet
+ * ready, and it needs to be played back into the file channel
+ */
+ if (producer == null) {
+ /*
+ * Data within 30s is placed in the exception channel to
+ * prevent frequent checks
+ * After 30s, reopen the topic check, if it is still a null value,
+ * put it back into the illegal map
+ */
+ return false;
+ }
+
+ Map<String, String> proMap = new HashMap<>();
+ proMap.put("tdbusip", localIp);
+ String tid = "";
+ if (event.getHeaders().containsKey(AttributeConstants.INTERFACE_ID)) {
+ tid = event.getHeaders().get(AttributeConstants.INTERFACE_ID);
+ } else if (event.getHeaders().containsKey(AttributeConstants.INAME)) {
+ tid = event.getHeaders().get(AttributeConstants.INAME);
+ }
+ proMap.put(tid, event.getHeaders().get(ConfigConstants.PKG_TIME_KEY));
+ logger.debug("producer send msg!");
+ TopicProducerInfo forCallBackP = producer;
+ forCallBackP.getProducer().newMessage().properties(proMap).value(event.getBody())
+ .sendAsync().thenAccept((msgId) -> {
+ forCallBackP.setCanUseSend(true);
+ sendMessageCallBack.handleMessageSendSuccess((MessageIdImpl)msgId, es);
+
+ }).exceptionally((e) -> {
+ forCallBackP.setCanUseSend(false);
+ sendMessageCallBack.handleMessageSendException(es, e);
+ return null;
+ });
+ return true;
+ }
+
+ /**
+ * If this function is called successively without calling {@see #destroyConnection()}, only the
+ * first call has any effect.
+ *
+ * @throws FlumeException if an RPC client connection could not be opened
+ */
+ private void createConnection(CreatePulsarClientCallBack callBack) throws FlumeException {
+ if (pulsarClients != null) {
+ return;
+ }
+ pulsarClients = new ArrayList<PulsarClient>();
+ for (int i = 0; i < pulsarServerUrls.length; i++) {
+ try {
+ logger.debug("index = {}, url = {}", i, pulsarServerUrls[i]);
+ PulsarClient client = initPulsarClient(pulsarServerUrls[i]);
+ pulsarClients.add(client);
+ callBack.handleCreateClientSuccess(pulsarServerUrls[i]);
+ } catch (PulsarClientException e) {
+ callBack.handleCreateClientException(pulsarServerUrls[i]);
+ logger.error("create connnection error in metasink, "
+ + "maybe pulsar master set error, please re-check.url{}, ex1 {}",
+ pulsarServerUrls[i],
+ e.getMessage());
+ } catch (Throwable e) {
+ callBack.handleCreateClientException(pulsarServerUrls[i]);
+ logger.error("create connnection error in metasink, "
+ + "maybe pulsar master set error/shutdown in progress, please "
+ + "re-check. url{}, ex2 {}",
+ pulsarServerUrls[i],
+ e.getMessage());
+ }
+ }
+ if (pulsarClients.size() == 0) {
+ throw new FlumeException("connect to pulsar error1, "
+ + "maybe zkstr/zkroot set error, please re-check");
+ }
+ }
+
+ private PulsarClient initPulsarClient(String pulsarUrl) throws Exception {
+ return PulsarClient.builder()
+ .serviceUrl(pulsarUrl)
+ .connectionTimeout(clientTimeout, TimeUnit.SECONDS)
+ .build();
+ }
+
+ public List<TopicProducerInfo> initTopicProducer(String topic) {
+ logger.info("initTopicProducer topic = {}", topic);
+ List<TopicProducerInfo> producerInfoList = producerInfoMap.computeIfAbsent(topic, (k) -> {
+ List<TopicProducerInfo> newList = new ArrayList<>();
+ if (pulsarClients != null) {
+ for (PulsarClient pulsarClient : pulsarClients) {
+ TopicProducerInfo info = new TopicProducerInfo(pulsarClient, topic);
+ info.initProducer();
+ newList.add(info);
+ }
+ }
+ return newList;
+ });
+ return producerInfoList;
+ }
+
+ private TopicProducerInfo getProducer(String topic) {
+ List<TopicProducerInfo> producerList = initTopicProducer(topic);
+ AtomicLong topicIndex = topicSendIndexMap.computeIfAbsent(topic,(k) -> {
+ return new AtomicLong(0);
+ });
+ int maxTryToGetProducer = producerList.size();
+ if (maxTryToGetProducer == 0) {
+ return null;
+ }
+ int retryTime = 0;
+ TopicProducerInfo p = null;
+ do {
+ int index = (int)(topicIndex.getAndIncrement() % maxTryToGetProducer);
+ p = producerList.get(index);
+ if (p.isCanUseToSendMessage() && p.getProducer().isConnected()) {
+ break;
+ }
+ retryTime++;
+ } while (retryTime < maxTryToGetProducer);
+ return p;
+ }
+
+ public Map<String, List<TopicProducerInfo>> getProducerInfoMap() {
+ return producerInfoMap;
+ }
+
+ private void destroyConnection() {
+ producerInfoMap.clear();
+ if (pulsarClients != null) {
+ for (PulsarClient pulsarClient : pulsarClients) {
+ try {
+ pulsarClient.shutdown();
+ } catch (PulsarClientException e) {
+ logger.error("destroy pulsarClient error in PulsarSink, PulsarClientException {}",
+ e.getMessage());
+ } catch (Exception e) {
+ logger.error("destroy pulsarClient error in PulsarSink, ex {}", e.getMessage());
+ }
+ }
+ }
+ pulsarClients = null;
+ logger.debug("closed meta producer");
+ }
+
+ public void close() {
+ destroyConnection();
+ }
+
+ class TopicProducerInfo {
+ private long lastSendMsgErrorTime;
+
+ private Producer producer;
+
+ private PulsarClient pulsarClient;
+
+ private String topic;
+
+ private volatile Boolean isCanUseSend = true;
+
+ private volatile Boolean isFinishInit = false;
+
+ public TopicProducerInfo(PulsarClient pulsarClient,
+ String topic) {
+ this.pulsarClient = pulsarClient;
+ this.topic = topic;
+ }
+
+ public void initProducer() {
+ try {
+ producer = pulsarClient.newProducer().sendTimeout(sendTimeout,
+ TimeUnit.MILLISECONDS)
+ .topic(topic)
+ .enableBatching(enableBatch)
+ .blockIfQueueFull(blockIfQueueFull)
+ .maxPendingMessages(maxPendingMessages)
+ .batchingMaxMessages(maxBatchingMessages)
+ .create();
+ isFinishInit = true;
+ } catch (PulsarClientException e) {
+ logger.error("create pulsar client has error e = {}", e);
+ isFinishInit = false;
+ }
+ }
+
+ public void setCanUseSend(Boolean isCanUseSend) {
+ this.isCanUseSend = isCanUseSend;
+ if (!isCanUseSend) {
+ lastSendMsgErrorTime = System.currentTimeMillis();
+ }
+ }
+
+ public boolean isCanUseToSendMessage() {
+ if (isCanUseSend && isFinishInit) {
+ return true;
+ } else if (isFinishInit && (System.currentTimeMillis() - lastSendMsgErrorTime)
+ > retryIntervalWhenSendMsgError) {
+ lastSendMsgErrorTime = System.currentTimeMillis();
+ return true;
+ }
+ return false;
+ }
+
+ public Producer getProducer() {
+ return producer;
+ }
+ }
+}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SendMessageCallBack.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SendMessageCallBack.java
new file mode 100644
index 0000000..66323e7
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/pulsar/SendMessageCallBack.java
@@ -0,0 +1,27 @@
+/*
+ * 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 org.apache.inlong.dataproxy.sink.EventStat;
+
+public interface SendMessageCallBack {
+
+ void handleMessageSendSuccess(Object msgId, EventStat es);
+
+ void handleMessageSendException(EventStat es, Object exception);
+}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/SimpleTcpSource.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/SimpleTcpSource.java
index 94a3b26..59df0a6 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/SimpleTcpSource.java
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/SimpleTcpSource.java
@@ -356,7 +356,7 @@ public class SimpleTcpSource extends AbstractSource implements Configurable, Eve
messageHandlerName =
context.getString(ConfigConstants.MESSAGE_HANDLER_NAME,
- "org.apache.flume.source.ServerMessageHandler");
+ "org.apache.inlong.dataproxy.source.ServerMessageHandler");
messageHandlerName = messageHandlerName.trim();
Preconditions.checkArgument(StringUtils.isNotBlank(messageHandlerName),
"messageHandlerName is empty");
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/FailoverChannelProcessorHolder.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/FailoverChannelProcessorHolder.java
new file mode 100644
index 0000000..4757bde
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/FailoverChannelProcessorHolder.java
@@ -0,0 +1,32 @@
+/*
+ * 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.utils;
+
+import org.apache.flume.channel.ChannelProcessor;
+
+public class FailoverChannelProcessorHolder {
+ private static ChannelProcessor channelProcessor;
+
+ public static ChannelProcessor getChannelProcessor() {
+ return channelProcessor;
+ }
+
+ public static void setChannelProcessor(ChannelProcessor cp) {
+ channelProcessor = cp;
+ }
+}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/LogCounter.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/LogCounter.java
new file mode 100644
index 0000000..9e45827
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/LogCounter.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.utils;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class LogCounter {
+
+ private AtomicInteger counter = new AtomicInteger(0);
+
+ private int start = 10;
+ private int control = 1000;
+ private int reset = 60 * 1000;
+
+ private long lastLogTime = System.currentTimeMillis();
+
+ public LogCounter(int start, int control, int reset) {
+ this.start = start;
+ this.control = control;
+ this.reset = reset;
+ }
+
+ public boolean shouldPrint() {
+ if (System.currentTimeMillis() - lastLogTime > reset) {
+ counter.set(0);
+ this.lastLogTime = System.currentTimeMillis();
+ }
+
+ if (counter.incrementAndGet() > start && counter.get() % control != 0) {
+ return false;
+ }
+
+ return true;
+ }
+}
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndex.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndex.java
new file mode 100644
index 0000000..b140677
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndex.java
@@ -0,0 +1,127 @@
+/*
+ * 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.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MonitorIndex {
+ private static final Logger logger = LoggerFactory.getLogger(MonitorIndex.class);
+ private static final LogCounter logPrinter = new LogCounter(10, 100000, 60 * 1000);
+
+ private IndexCollectThread indexCol;
+ private String name;
+ private ConcurrentHashMap<String, String> counterMap = new ConcurrentHashMap<String, String>();
+ private int intervalSec;
+ private int maxCnt;
+
+ public MonitorIndex(String name, int intervalSec, int maxCnt) {
+ /*
+ * Main indicators, placed in flume_index.log
+ */
+ this.intervalSec = intervalSec;
+ /*
+ * "TDBus","TDBus_intf"...tmetric pattern name
+ */
+ this.name = name;
+ this.maxCnt = maxCnt;
+ indexCol = new IndexCollectThread();
+ indexCol.setDaemon(true);
+ indexCol.setName("IndexCollectThread_MonitorIndex");
+ indexCol.start();
+ }
+
+ public void addAndGet(String key, int cnt, int packcnt, long packsize, int failcnt) {
+ try {
+ if (counterMap.size() < maxCnt) {
+ counterMap.compute(key, (key1, value) -> {
+ if (value != null) {
+ String[] va = value.split("#");
+ value = (Integer.parseInt(va[0]) + cnt) + "#"
+ + (Integer.parseInt(va[1]) + packcnt) + "#"
+ + (Long.parseLong(va[2]) + packsize) + "#"
+ + (Integer.parseInt(va[3]) + failcnt);
+ } else {
+ StringBuilder stringBuilder = new StringBuilder();
+ value = stringBuilder.append(cnt).append("#").append(packcnt).append("#")
+ .append(packsize).append("#").append(failcnt).toString();
+ }
+ return value;
+ });
+ } else if (logPrinter.shouldPrint()) {
+ logger.error(this.name + "exceed monitor's max size");
+ }
+ } catch (Exception e) {
+ if (logPrinter.shouldPrint()) {
+ logger.error("monitor exception",e);
+ }
+ }
+ }
+
+ public void shutDown() {
+ indexCol.shutDown();
+ }
+
+ public int getMaxCnt() {
+ return maxCnt;
+ }
+
+ public void setMaxCnt(int maxCnt) {
+ this.maxCnt = maxCnt;
+ }
+
+ private class IndexCollectThread extends Thread {
+ private boolean bShutDown = false;
+
+ public IndexCollectThread() {
+ bShutDown = false;
+ }
+
+ public void shutDown() {
+ bShutDown = true;
+ }
+
+ @Override
+ public void run() {
+ Map<String, String> counterExt = new HashMap<String, String>();
+ while (!bShutDown) {
+ try {
+ Thread.sleep(intervalSec * 1000L);
+ for (String str : counterMap.keySet()) {
+ counterMap.computeIfPresent(str, (s, s2) -> {
+ counterExt.put(s, s2);
+ return null;
+ });
+ }
+ for (Map.Entry<String, String> entrys : counterExt.entrySet()) {
+ logger.info(
+ "{}#{}#{}", new Object[] { name, entrys.getKey(), entrys.getValue() });
+ }
+ counterExt.clear();
+ } catch (Exception e) {
+ logger.warn("monitor interrupted");
+ }
+ }
+
+ }
+ }
+}
+
diff --git a/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndexExt.java b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndexExt.java
new file mode 100644
index 0000000..959680e
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MonitorIndexExt.java
@@ -0,0 +1,125 @@
+/*
+ * 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.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MonitorIndexExt {
+ private static final Logger logger = LoggerFactory.getLogger(MonitorIndexExt.class);
+ private static final LogCounter logPrinter = new LogCounter(10, 100000, 60 * 1000);
+
+ private IndexCollectThread indexCol;
+ private String name;
+ private ConcurrentHashMap<String, AtomicLong> counterMap = new ConcurrentHashMap<String, AtomicLong>();
+ private int intervalSec;
+ private int maxCnt;
+
+ public MonitorIndexExt(String name, int intervalSec, int maxCnt) {
+ /*
+ * key
+ * Almost unchanging. Component indicators, flume_monitors.log
+ */
+ this.intervalSec = intervalSec;
+ /*
+ * TDBus_monitors", used to monitor...tmetric pattern name
+ */
+ this.name = name;
+ this.maxCnt = maxCnt;
+ indexCol = new IndexCollectThread();
+ indexCol.setDaemon(true);
+ indexCol.setName("IndexCollectThread_MonitorIndexExt");
+ indexCol.start();
+ }
+
+ public void incrementAndGet(String key) {
+ try {
+ if (counterMap.size() < maxCnt) {
+ counterMap.compute(key, (s, atomicLong) -> {
+ if (atomicLong != null) {
+ atomicLong.incrementAndGet();
+ } else {
+ atomicLong = new AtomicLong(1);
+ }
+ return atomicLong;
+ });
+ } else if (logPrinter.shouldPrint()) {
+ logger.error(this.name + "exceed monitorExt's max size");
+ }
+ } catch (Exception e) {
+ if (logPrinter.shouldPrint()) {
+ logger.error("monitorExt exception",e);
+ }
+ }
+ }
+
+ public void shutDown() {
+ indexCol.shutDown();
+ }
+
+ public int getMaxCnt() {
+ return maxCnt;
+ }
+
+ public void setMaxCnt(int maxCnt) {
+ this.maxCnt = maxCnt;
+ }
+
+ private class IndexCollectThread extends Thread {
+ private boolean bShutDown = false;
+
+ public IndexCollectThread() {
+ bShutDown = false;
+ }
+
+ public void shutDown() {
+ bShutDown = true;
+ }
+
+ @Override
+ public void run() {
+ Map<String, Long> counterExt = new HashMap<String, Long>();
+ while (!bShutDown) {
+ try {
+ Thread.sleep(intervalSec * 1000L);
+ for (String str : counterMap.keySet()) {
+ counterMap.computeIfPresent(str, (s, atomicLong) -> {
+ long cnt = atomicLong.get();
+ counterExt.put(str, cnt);
+ atomicLong.set(0L);
+ return atomicLong;
+ });
+ }
+
+ for (Map.Entry<String, Long> entrys : counterExt.entrySet()) {
+ logger.info("{}#{}#{}", new Object[] { name, entrys.getKey(), entrys.getValue() });
+ }
+ counterExt.clear();
+
+ } catch (Exception e) {
+ logger.warn("moniorExt interrupted");
+ }
+ }
+ }
+ }
+}
+
diff --git a/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf b/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf
index d0aa15e..c369630 100644
--- a/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf
+++ b/inlong-dataproxy/dataproxy-source/src/main/resources/flume.conf
@@ -266,6 +266,16 @@ agent1.sinks.meta-sink-msg8.old-metric-on=true
agent1.sinks.meta-sink-msg8.set=10
agent1.sinks.meta-sink-msg8.disk-io-rate-per-sec=20000000
+
+agent1.sinks.pulsar-sink-msg1.channel = ch-msg1
+agent1.sinks.pulsar-sink-msg1.type = org.apache.inlong.dataproxy.sink.PulsarSink
+agent1.sinks.pulsar-sink-msg1.pulsar_server_url_list = pulsar://127.0.0.1:6650|pulsar://127.0.0.2:6650
+agent1.sinks.pulsar-sink-msg1.send_timeout_MILL = 30000
+agent1.sinks.pulsar-sink-msg1.stat_interval_sec = 60
+agent1.sinks.pulsar-sink-msg1.thread-num = 8
+agent1.sinks.pulsar-sink-msg1.client-id-cache = true
+agent1.sinks.pulsar-sink-msg1.disk-io-rate-per-sec=20000000
+
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/test/java/org/apache/inlong/dataproxy/sink/TestPulsarSink.java b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/sink/TestPulsarSink.java
new file mode 100644
index 0000000..a292840
--- /dev/null
+++ b/inlong-dataproxy/dataproxy-source/src/test/java/org/apache/inlong/dataproxy/sink/TestPulsarSink.java
@@ -0,0 +1,96 @@
+/*
+ * 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;
+
+import com.google.common.base.Charsets;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Sink;
+import org.apache.flume.Transaction;
+import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.EventBuilder;
+import org.apache.flume.lifecycle.LifecycleController;
+import org.apache.flume.lifecycle.LifecycleState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import org.testng.Assert;
+
+public class TestPulsarSink {
+ private static final Logger logger = LoggerFactory
+ .getLogger(TestPulsarSink.class);
+ private static final String hostname = "127.0.0.1";
+ private static final Integer port = 1234;
+ private String zkStr = "127.0.0.1:2181";
+ private String zkRoot = "/meta";
+ private int batchSize = 1;
+
+ private PulsarSink sink;
+ private Channel channel;
+
+ @BeforeClass
+ public void setUp() {
+ sink = new PulsarSink();
+ channel = new MemoryChannel();
+
+ Context context = new Context();
+
+ context.put("type", "org.apache.inlong.dataproxy.sink.PulsarSink");
+ context.put("pulsar_server_url_list", "pulsar://127.0.0.1:6650");
+
+ sink.setChannel(channel);
+
+ Configurables.configure(sink, context);
+ Configurables.configure(channel, context);
+ }
+
+ @Test
+ public void testProcess() throws InterruptedException, EventDeliveryException,
+ InstantiationException, IllegalAccessException {
+ setUp();
+ Event event = EventBuilder.withBody("test event 1", Charsets.UTF_8);
+ sink.start();
+ Assert.assertTrue(LifecycleController.waitForOneOf(sink,
+ LifecycleState.START_OR_ERROR, 5000));
+
+ Transaction transaction = channel.getTransaction();
+
+ transaction.begin();
+ for (int i = 0; i < 10; i++) {
+ channel.put(event);
+ }
+ transaction.commit();
+ transaction.close();
+
+ for (int i = 0; i < 5; i++) {
+ Sink.Status status = sink.process();
+ Assert.assertEquals(Sink.Status.READY, status);
+ }
+
+ sink.stop();
+ Assert.assertTrue(LifecycleController.waitForOneOf(sink,
+ LifecycleState.STOP_OR_ERROR, 5000));
+ }
+
+}
diff --git a/inlong-dataproxy/pom.xml b/inlong-dataproxy/pom.xml
index d73a77b..cbf59ab 100644
--- a/inlong-dataproxy/pom.xml
+++ b/inlong-dataproxy/pom.xml
@@ -46,6 +46,8 @@
<codec.version>1.15</codec.version>
<servlet.version>2.5-20110124</servlet.version>
<inlong-common.version>1.3.4</inlong-common.version>
+ <pulsar.version>2.8.1</pulsar.version>
+ <testng.version>6.14.3</testng.version>
</properties>
<dependencies>
@@ -94,6 +96,17 @@
<artifactId>inlong-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client</artifactId>
+ <version>${pulsar.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testng</groupId>
+ <artifactId>testng</artifactId>
+ <version>${testng.version}</version>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file