This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 641a5c9c0 [INLONG-7519][Audit] Proxy support Kafka (#7524)
641a5c9c0 is described below
commit 641a5c9c03bb72c32db77e8b95d0968dd86d5f0b
Author: haifxu <[email protected]>
AuthorDate: Tue Mar 7 16:39:06 2023 +0800
[INLONG-7519][Audit] Proxy support Kafka (#7524)
---
inlong-audit/audit-proxy/pom.xml | 4 +
.../audit/sink/{PulsarSink.java => KafkaSink.java} | 485 +++++++++------------
.../org/apache/inlong/audit/sink/PulsarSink.java | 18 +-
.../org/apache/inlong/audit/sink/TubeSink.java | 8 +-
.../apache/inlong/audit/sink/KafkaSinkTest.java | 84 ++++
inlong-audit/conf/audit-proxy-kafka.conf | 75 ++++
6 files changed, 387 insertions(+), 287 deletions(-)
diff --git a/inlong-audit/audit-proxy/pom.xml b/inlong-audit/audit-proxy/pom.xml
index 1df74fff6..8fdeef909 100644
--- a/inlong-audit/audit-proxy/pom.xml
+++ b/inlong-audit/audit-proxy/pom.xml
@@ -73,6 +73,10 @@
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ </dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/KafkaSink.java
similarity index 51%
copy from
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
copy to
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/KafkaSink.java
index c8c120537..1f357bec5 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/KafkaSink.java
@@ -19,223 +19,136 @@ package org.apache.inlong.audit.sink;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.RateLimiter;
-import io.netty.handler.codec.TooLongFrameException;
-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.commons.lang3.StringUtils;
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.audit.base.HighPriorityThreadFactory;
-import org.apache.inlong.audit.sink.pulsar.CreatePulsarClientCallBack;
-import org.apache.inlong.audit.sink.pulsar.PulsarClientService;
-import org.apache.inlong.audit.sink.pulsar.SendMessageCallBack;
import org.apache.inlong.audit.utils.FailoverChannelProcessorHolder;
+import org.apache.inlong.common.util.NetworkUtils;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * pulsar sink
- *
- * send to one pulsar cluster
- */
-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_LOG_EVERY_N_EVENTS = 100000;
-
- /*
- * properties for stat
- */
- private static String LOG_EVERY_N_EVENTS = "log_every_n_events";
-
- private static String DISK_IO_RATE_PER_SEC = "disk_io_rate_per_sec";
-
- private static final String SINK_THREAD_NUM = "thread_num";
-
- /*
- * 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;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+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;
- private static AtomicLong totalPulsarSuccSendCnt = new AtomicLong(0);
+public class KafkaSink extends AbstractSink implements Configurable {
- private static AtomicLong totalPulsarSuccSendSize = new AtomicLong(0);
- /*
- * for control
- */
- private boolean overflow = false;
+ private static final Logger logger =
LoggerFactory.getLogger(KafkaSink.class);
- private LinkedBlockingQueue<EventStat> resendQueue;
+ // for kafka producer
+ private static Properties properties = new Properties();
+ private static final String BOOTSTRAP_SERVER = "bootstrap_servers";
+ private static final String TOPIC = "topic";
+ private static final String RETRIES = "retries";
+ private static final String BATCH_SIZE = "batch_size";
+ private static final String LINGER_MS = "linger_ms";
+ private static final String BUFFER_MEMORY = "buffer_memory";
+ private static final String defaultRetries = "0";
+ private static final String defaultBatchSize = "16384";
+ private static final String defaultLingerMs = "0";
+ private static final String defaultBufferMemory = "33554432";
+ private static final String defaultAcks = "all";
- private long logCounter = 0;
-
- private final AtomicLong currentInFlightCount = new AtomicLong(0);
+ private static final Long PRINT_INTERVAL = 30L;
+ private static final KafkaPerformanceTask kafkaPerformanceTask = new
KafkaPerformanceTask();
+ private static ScheduledExecutorService scheduledExecutorService =
Executors.newScheduledThreadPool(1,
+ new HighPriorityThreadFactory("kafkaPerformance-Printer-thread"));
- /*
- * whether the SendTask thread can send data to pulsar
- */
+ private KafkaProducer<String, byte[]> producer;
+ public Map<String, KafkaProducer<String, byte[]>> producerMap;
+ private SinkCounter sinkCounter;
+ private String topic;
private volatile boolean canSend = false;
-
- /*
- * Control whether the SinkRunner thread can read data from the Channel
- */
private volatile boolean canTake = false;
-
- private static int EVENT_QUEUE_SIZE = 1000;
-
private int threadNum;
-
- /*
- * send thread pool
- */
private Thread[] sinkThreadPool;
- private LinkedBlockingQueue<Event> eventQueue;
- private SinkCounter sinkCounter;
+ private static final int BAD_EVENT_QUEUE_SIZE = 10000;
+ private static final int EVENT_QUEUE_SIZE = 1000;
+ private static final int DEFAULT_LOG_EVERY_N_EVENTS = 100000;
+ private LinkedBlockingQueue<EventStat> resendQueue;
+ private LinkedBlockingQueue<Event> eventQueue;
- private PulsarClientService pulsarClientService;
+ // for log
+ private Integer logEveryNEvents;
+ private long diskIORatePerSec;
+ private RateLimiter diskRateLimiter;
- private static final Long PRINT_INTERVAL = 30L;
+ // properties for stat
+ private static final String LOG_EVERY_N_EVENTS = "log_every_n_events";
+ private static final String DISK_IO_RATE_PER_SEC = "disk_io_rate_per_sec";
+ private static final String SINK_THREAD_NUM = "thread-num";
- private static final PulsarPerformanceTask pulsarPerformanceTask = new
PulsarPerformanceTask();
+ // for stas
+ private AtomicLong currentSuccessSendCnt = new AtomicLong(0);
+ private AtomicLong lastSuccessSendCnt = new AtomicLong(0);
+ private long t1 = System.currentTimeMillis();
+ private long t2 = 0L;
+ private static AtomicLong totalKafkaSuccSendCnt = new AtomicLong(0);
+ private static AtomicLong totalKafkaSuccSendSize = new AtomicLong(0);
- private static ScheduledExecutorService scheduledExecutorService =
Executors
- .newScheduledThreadPool(1, new
HighPriorityThreadFactory("pulsarPerformance-Printer-thread"));
+ private boolean overflow = false;
- private String topic;
+ private String localIp = "127.0.0.1";
static {
- /*
- * stat pulsar performance
- */
- System.out.println("pulsarPerformanceTask!!!!!!");
- scheduledExecutorService.scheduleWithFixedDelay(pulsarPerformanceTask,
0L,
+ // stat kafka performance
+ logger.info("init kafkaPerformanceTask");
+ scheduledExecutorService.scheduleWithFixedDelay(kafkaPerformanceTask,
0L,
PRINT_INTERVAL, TimeUnit.SECONDS);
}
- public PulsarSink() {
+ public KafkaSink() {
super();
- logger.debug("new instance of PulsarSink!");
- }
-
- /**
- * configure
- * @param context
- */
- public void configure(Context context) {
- logger.info("PulsarSink started and context = {}", context.toString());
- /*
- * topic config
- */
- topic = context.getString(TOPIC);
- 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");
-
- 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 initTopic() throws Exception {
- long startTime = System.currentTimeMillis();
- if (topic != null) {
- pulsarClientService.initTopicProducer(topic);
- }
- logger.info(getName() + " initTopic cost: "
- + (System.currentTimeMillis() - startTime) + "ms");
+ logger.debug("new instance of KafkaSink!");
}
@Override
- public void start() {
- logger.info("pulsar sink starting...");
- sinkCounter.start();
- pulsarClientService.initCreateConnection(this);
+ public synchronized void start() {
+ logger.info("kafka sink starting");
+ // create connection
+ sinkCounter.start();
super.start();
this.canSend = true;
this.canTake = true;
- try {
- initTopic();
- } catch (Exception e) {
- logger.info("meta sink start publish topic fail.", e);
- }
+
+ // init topic producer
+ initTopicProducer(topic);
for (int i = 0; i < sinkThreadPool.length; i++) {
- sinkThreadPool[i] = new Thread(new SinkTask(), getName()
- + "_pulsar_sink_sender-"
- + i);
+ sinkThreadPool[i] = new Thread(new SinkTask(), getName() +
"_tube_sink_sender-" + i);
sinkThreadPool[i].start();
}
- logger.debug("meta sink started");
+ logger.debug("kafka sink started");
}
@Override
- public void stop() {
- logger.info("pulsar sink stopping");
- pulsarClientService.close();
+ public synchronized void stop() {
+ logger.info("kafka sink stopping");
+ // stop connection
this.canTake = false;
int waitCount = 0;
while (eventQueue.size() != 0 && waitCount++ < 10) {
try {
- Thread.currentThread().sleep(1000);
+ Thread.sleep(1000);
} catch (InterruptedException e) {
logger.info("Stop thread has been interrupt!");
break;
@@ -256,12 +169,12 @@ public class PulsarSink extends AbstractSink
scheduledExecutorService.shutdown();
}
sinkCounter.stop();
- logger.debug("pulsar sink stopped. Metrics:{}", sinkCounter);
+ logger.debug("kafka sink stopped. Metrics:{}", sinkCounter);
}
@Override
- public Status process() throws EventDeliveryException {
- logger.debug("process......");
+ public Status process() {
+ logger.info("kafka sink processing");
if (!this.canTake) {
return Status.BACKOFF;
}
@@ -277,8 +190,8 @@ public class PulsarSink extends AbstractSink
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 "
+ logger.info("[{}] Channel --> Queue(not enough space,
current code point) "
+ + "--> Kafka, check if Kafka server or network is
ok. (If this situation "
+ "last long time it will cause memoryChannel full
and fileChannel write.)", getName());
tx.rollback();
} else {
@@ -293,7 +206,7 @@ public class PulsarSink extends AbstractSink
try {
tx.rollback();
} catch (Throwable e) {
- logger.error("Pulsar Sink transaction rollback exception =
{}", e);
+ logger.error("Kafka Sink transaction rollback exception = {}",
e);
}
} finally {
tx.close();
@@ -302,69 +215,117 @@ public class PulsarSink extends AbstractSink
}
@Override
- public void handleCreateClientSuccess(String url) {
- logger.info("createConnection success for url = {}", url);
- sinkCounter.incrementConnectionCreatedCount();
+ public void configure(Context context) {
+ logger.info("KafkaSink started and context = {}", context.toString());
+
+ topic = context.getString(TOPIC);
+ Preconditions.checkState(StringUtils.isNotEmpty(topic), "No topic
specified");
+
+ producerMap = new HashMap<>();
+
+ 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");
+
+ resendQueue = new LinkedBlockingQueue<>(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_QUEUE_SIZE);
+
+ diskIORatePerSec = context.getLong(DISK_IO_RATE_PER_SEC, 0L);
+ if (diskIORatePerSec != 0) {
+ diskRateLimiter = RateLimiter.create(diskIORatePerSec);
+ }
+
+ if (sinkCounter == null) {
+ sinkCounter = new SinkCounter(getName());
+ }
+
+ localIp = NetworkUtils.getLocalIp();
+
+ properties = new Properties();
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
context.getString(BOOTSTRAP_SERVER));
+ properties.put(ProducerConfig.ACKS_CONFIG, defaultAcks);
+ properties.put(ProducerConfig.RETRIES_CONFIG,
context.getString(RETRIES, defaultRetries));
+ properties.put(ProducerConfig.BATCH_SIZE_CONFIG,
context.getString(BATCH_SIZE, defaultBatchSize));
+ properties.put(ProducerConfig.LINGER_MS_CONFIG,
context.getString(LINGER_MS, defaultLingerMs));
+ properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,
context.getString(BUFFER_MEMORY, defaultBufferMemory));
}
- @Override
- public void handleCreateClientException(String url) {
- logger.error("createConnection has exception for url = {}", url);
- sinkCounter.incrementConnectionFailedCount();
+ private void initTopicProducer(String topic) {
+ if (StringUtils.isEmpty(topic)) {
+ logger.error("topic is empty");
+ }
+
+ if (producer == null) {
+ producer = new KafkaProducer<>(properties, new StringSerializer(),
new ByteArraySerializer());
+ }
+
+ producerMap.put(topic, producer);
+ logger.info(getName() + " success create producer");
}
- @Override
- public void handleMessageSendSuccess(Object result, EventStat eventStat) {
- /*
- * Statistics pulsar performance
- */
- totalPulsarSuccSendCnt.incrementAndGet();
-
totalPulsarSuccSendSize.addAndGet(eventStat.getEvent().getBody().length);
- /*
- * add to sinkCounter
- */
+ private KafkaProducer<String, byte[]> getProducer(String topic) {
+ if (!producerMap.containsKey(topic)) {
+ synchronized (this) {
+ if (!producerMap.containsKey(topic)) {
+ if (producer == null) {
+ producer = new KafkaProducer<>(properties);
+ }
+ producerMap.put(topic, producer);
+ }
+ }
+ }
+ return producerMap.get(topic);
+ }
+
+ static class KafkaPerformanceTask implements Runnable {
+
+ @Override
+ public void run() {
+ try {
+ if (totalKafkaSuccSendSize.get() != 0) {
+ logger.info("Total kafka performance tps: "
+ + totalKafkaSuccSendCnt.get() / PRINT_INTERVAL
+ + "/s, avg msg size: "
+ + totalKafkaSuccSendSize.get() /
totalKafkaSuccSendCnt.get()
+ + ", print every " + PRINT_INTERVAL + " seconds");
+
+ // totalKafkaSuccSendCnt represents the number of packets
+ totalKafkaSuccSendSize.set(0);
+ totalKafkaSuccSendCnt.set(0);
+ }
+
+ } catch (Exception e) {
+ logger.info("tubePerformanceTask error", e);
+ }
+ }
+ }
+
+ public void handleMessageSendSuccess(EventStat es) {
+ // Statistics tube performance
+ totalKafkaSuccSendCnt.incrementAndGet();
+ totalKafkaSuccSendSize.addAndGet(es.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)
- });
+ logger.info("KafkaSink {}, succ put {} events to kafka, in the
past {} millisecond",
+ getName(), (nowCnt - oldCnt), (t2 - t1));
t1 = t2;
}
}
- @Override
- public void handleMessageSendException(EventStat eventStat, Object e) {
- 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)) {
- logger.error("handle message send exception ,msg will resend
later, e = {}", e);
- }
- 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;
}
@@ -376,92 +337,43 @@ public class PulsarSink extends AbstractSink
}
}
- 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) {
- logger.debug("SinkTask process......");
boolean decrementFlag = false;
Event event = null;
EventStat eventStat = null;
try {
- if (PulsarSink.this.overflow) {
- PulsarSink.this.overflow = false;
- Thread.currentThread().sleep(10);
+ if (KafkaSink.this.overflow) {
+ KafkaSink.this.overflow = false;
+ Thread.sleep(10);
}
if (!resendQueue.isEmpty()) {
- /*
- * Send the data in the retry queue first
- */
+ // 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;
- }
- }
event = eventQueue.take();
eventStat = new EventStat(event);
sinkCounter.incrementEventDrainAttemptCount();
}
- 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");
+ if (event == null || StringUtils.isBlank(topic)) {
+ logger.warn("event is null or no topic specified in
event header, just skip");
continue;
}
final EventStat es = eventStat;
- boolean sendResult =
pulsarClientService.sendMessage(topic, event,
- PulsarSink.this, es);
+ boolean sendResult = sendMessage(event, topic, es);
if (!sendResult) {
continue;
}
- currentInFlightCount.incrementAndGet();
+
decrementFlag = true;
} catch (InterruptedException e) {
logger.info("Thread {} has been interrupted!",
Thread.currentThread().getName());
@@ -485,5 +397,30 @@ public class PulsarSink extends AbstractSink
}
}
}
+
+ private boolean sendMessage(Event event, String topic, EventStat es) {
+ KafkaProducer<String, byte[]> producer = getProducer(topic);
+ if (producer == null) {
+ logger.error("Get producer is null, topic:{}", topic);
+ return false;
+ }
+
+ logger.debug("producer start to send msg...");
+ ProducerRecord<String, byte[]> record = new
ProducerRecord<>(topic, event.getBody());
+ producer.send(record, (recordMetadata, e) -> {
+
+ if (e == null) {
+ handleMessageSendSuccess(es);
+ return;
+ } else {
+ logger.warn("Send message failed, error message: {},
resendQueue size: {}, event:{}",
+ e.getMessage(), resendQueue.size(),
es.getEvent().hashCode());
+ }
+
+ es.incRetryCnt();
+ resendEvent(es, true);
+ });
+ return true;
+ }
}
}
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
index c8c120537..e374b5de2 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/PulsarSink.java
@@ -20,11 +20,6 @@ package org.apache.inlong.audit.sink;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.RateLimiter;
import io.netty.handler.codec.TooLongFrameException;
-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;
@@ -46,6 +41,12 @@ import
org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedExcepti
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+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;
+
/**
* pulsar sink
*
@@ -151,7 +152,7 @@ public class PulsarSink extends AbstractSink
/*
* stat pulsar performance
*/
- System.out.println("pulsarPerformanceTask!!!!!!");
+ logger.info("init pulsarPerformanceTask");
scheduledExecutorService.scheduleWithFixedDelay(pulsarPerformanceTask,
0L,
PRINT_INTERVAL, TimeUnit.SECONDS);
}
@@ -205,7 +206,7 @@ public class PulsarSink extends AbstractSink
@Override
public void start() {
- logger.info("pulsar sink starting...");
+ logger.info("pulsar sink starting");
sinkCounter.start();
pulsarClientService.initCreateConnection(this);
@@ -261,7 +262,7 @@ public class PulsarSink extends AbstractSink
@Override
public Status process() throws EventDeliveryException {
- logger.debug("process......");
+ logger.info("pulsar sink processing");
if (!this.canTake) {
return Status.BACKOFF;
}
@@ -406,7 +407,6 @@ public class PulsarSink extends AbstractSink
public void run() {
logger.info("Sink task {} started.",
Thread.currentThread().getName());
while (canSend) {
- logger.debug("SinkTask process......");
boolean decrementFlag = false;
Event event = null;
EventStat eventStat = null;
diff --git
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java
index 0caf67ef7..b5f22d1e5 100644
---
a/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java
+++
b/inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/sink/TubeSink.java
@@ -100,7 +100,7 @@ public class TubeSink extends AbstractSink implements
Configurable {
/*
* stat tube performance
*/
- logger.info("tubePerformanceTask!!!!!!");
+ logger.info("init tubePerformanceTask");
scheduledExecutorService.scheduleWithFixedDelay(tubePerformanceTask,
0L,
PRINT_INTERVAL, TimeUnit.SECONDS);
}
@@ -144,7 +144,7 @@ public class TubeSink extends AbstractSink implements
Configurable {
@Override
public synchronized void start() {
- logger.info("tube sink starting...");
+ logger.info("tube sink starting");
try {
createConnection();
} catch (FlumeException e) {
@@ -176,7 +176,7 @@ public class TubeSink extends AbstractSink implements
Configurable {
@Override
public synchronized void stop() {
- logger.info("tubesink stopping");
+ logger.info("tube sink stopping");
destroyConnection();
this.canTake = false;
int waitCount = 0;
@@ -209,7 +209,7 @@ public class TubeSink extends AbstractSink implements
Configurable {
@Override
public Status process() throws EventDeliveryException {
- logger.debug("process......");
+ logger.info("tube sink processing");
if (!this.canTake) {
return Status.BACKOFF;
}
diff --git
a/inlong-audit/audit-proxy/src/test/java/org/apache/inlong/audit/sink/KafkaSinkTest.java
b/inlong-audit/audit-proxy/src/test/java/org/apache/inlong/audit/sink/KafkaSinkTest.java
new file mode 100644
index 000000000..eed28a208
--- /dev/null
+++
b/inlong-audit/audit-proxy/src/test/java/org/apache/inlong/audit/sink/KafkaSinkTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.audit.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.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.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.powermock.api.mockito.PowerMockito;
+
+public class KafkaSinkTest {
+
+ private KafkaSink kafkaSink;
+ private Channel channel;
+ private Context context;
+
+ @Before
+ public void setUp() throws Exception {
+ kafkaSink = PowerMockito.mock(KafkaSink.class);
+ PowerMockito.doNothing().when(kafkaSink, "start");
+ PowerMockito.when(kafkaSink.process()).thenReturn(Sink.Status.READY);
+
PowerMockito.when(kafkaSink.getLifecycleState()).thenReturn(LifecycleState.ERROR);
+ channel = new MemoryChannel();
+ context = new Context();
+
+ context.put("topic", "inlong-audit");
+ context.put("master-host-port-list", "127.0.0.1:8080");
+
+ kafkaSink.setChannel(channel);
+ Configurables.configure(kafkaSink, context);
+ Configurables.configure(channel, context);
+
+ }
+
+ @Test
+ public void testProcess() throws InterruptedException {
+ Event event = EventBuilder.withBody("test", Charsets.UTF_8);
+ kafkaSink.start();
+ Assert.assertTrue(LifecycleController.waitForOneOf(kafkaSink,
+ 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 = kafkaSink.process();
+ Assert.assertEquals(Sink.Status.READY, status);
+ }
+
+ kafkaSink.stop();
+ Assert.assertTrue(LifecycleController.waitForOneOf(kafkaSink,
+ LifecycleState.STOP_OR_ERROR, 5000));
+ }
+}
diff --git a/inlong-audit/conf/audit-proxy-kafka.conf
b/inlong-audit/conf/audit-proxy-kafka.conf
new file mode 100644
index 000000000..7263b7427
--- /dev/null
+++ b/inlong-audit/conf/audit-proxy-kafka.conf
@@ -0,0 +1,75 @@
+#
+# 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.
+#
+
+# The configuration file needs to define the sources,
+# the channels and the sinks.
+# Sources, channels and sinks are defined per agent,
+# in this case called 'agent'
+
+agent1.sources = tcp-source
+agent1.channels = ch-msg1
+agent1.sinks = kafka-sink-msg1
+
+agent1.sources.tcp-source.channels = ch-msg1
+agent1.sources.tcp-source.type = org.apache.inlong.audit.source.SimpleTcpSource
+agent1.sources.tcp-source.msg-factory-name =
org.apache.inlong.audit.source.ServerMessageFactory
+agent1.sources.tcp-source.host = 0.0.0.0
+agent1.sources.tcp-source.port = 10081
+agent1.sources.tcp-source.max-msg-length = 524288
+agent1.sources.tcp-source.connections = 30000
+agent1.sources.tcp-source.max-threads = 64
+agent1.sources.tcp-source.receiveBufferSize = 1048576
+agent1.sources.tcp-source.sendBufferSize = 1048576
+agent1.sources.tcp-source.custom-cp = true
+agent1.sources.tcp-source.selector.type =
org.apache.inlong.audit.channel.FailoverChannelSelector
+agent1.sources.tcp-source.selector.master = ch-msg1
+agent1.sources.tcp-source.metric-recovery-path=./data/recovery
+agent1.sources.tcp-source.set=10
+
+agent1.channels.ch-msg1.type = memory
+agent1.channels.ch-msg1.capacity = 10000
+agent1.channels.ch-msg1.keep-alive = 0
+agent1.channels.ch-msg1.transactionCapacity = 200
+
+agent1.channels.ch-msg2.type = file
+agent1.channels.ch-msg2.capacity = 100000000
+agent1.channels.ch-msg2.maxFileSize = 1073741824
+agent1.channels.ch-msg2.minimumRequiredSpace = 1073741824
+agent1.channels.ch-msg2.checkpointDir =./data/file/ch-msg2/check
+agent1.channels.ch-msg2.dataDirs =./data/file/ch-msg2/data
+agent1.channels.ch-msg2.fsyncPerTransaction = false
+agent1.channels.ch-msg2.fsyncInterval = 10
+
+agent1.sinks.kafka-sink-msg1.channel = ch-msg1
+agent1.sinks.kafka-sink-msg1.type = org.apache.inlong.audit.sink.KafkaSink
+agent1.sinks.kafka-sink-msg1.bootstrap_servers = localhost:9092
+agent1.sinks.kafka-sink-msg1.topic = inlong-audit
+agent1.sinks.kafka-sink-msg1.retries = 0
+agent1.sinks.kafka-sink-msg1.batch_size = 16384
+agent1.sinks.kafka-sink-msg1.linger_ms = 0
+agent1.sinks.kafka-sink-msg1.buffer_memory = 33554432
+
+agent1.sinks.kafka-sink-msg2.channel = ch-msg1
+agent1.sinks.kafka-sink-msg2.type = org.apache.inlong.audit.sink.KafkaSink
+agent1.sinks.kafka-sink-msg2.bootstrap_servers = localhost:9092
+agent1.sinks.kafka-sink-msg2.topic = inlong-audit
+agent1.sinks.kafka-sink-msg2.retries = 0
+agent1.sinks.kafka-sink-msg2.batch_size = 16384
+agent1.sinks.kafka-sink-msg2.linger_ms = 0
+agent1.sinks.kafka-sink-msg2.buffer_memory = 33554432
