Adding Kafka-emitter #3860

dkhwangbo · 2017-01-19T01:29:49Z

This PR is about adding new extensions-contrib.

Currently, Druid has various emitter module and also has HttpPostEmitter for general purpose. But, Many user currently using Apache Kafka platform and I assume that If druid has another extension emitting their metrics to kafka directly, many user feel comfort and convenience using various monitoring dashboard UI or another eco.

Please feel free for commenting this PR.

dkhwangbo · 2017-01-19T08:06:03Z

This PR is under construction but I wanna get many people's comment about direction of this PR.

fjy · 2017-01-20T18:06:19Z

Thanks @dkhwangbo. Will review soon

dkhwangbo · 2017-01-31T01:01:27Z

@fjy Hi. Any update is in there?

b-slim · 2017-02-01T21:53:42Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+      TypeReference<Map<String, Object>> typeRef = new TypeReference<Map<String, Object>>() {};
+      HashMap<String, Object> result = jsonMapper.readValue(jsonMapper.writeValueAsString(event), typeRef);
+      result.put("clusterName", config.getClusterName());
+      producer.send(new ProducerRecord<String, String>(config.getTopic(), jsonMapper.writeValueAsString(result)));


at least two major issues here.

producer.send can be blocking and that's is not ok (emit call should never block)

alerts are not handled.

doing retry in case of exception is the norm

@b-slim Hi! Thanks for your review.
Kafka new producer api (since 0.8.2) only works asynchronously, so, as my understanding, using another method or logic is needless. If my comment is wrong, please correct it.

dkhwangbo · 2017-02-02T05:10:07Z

@b-slim I just reflect your another comment. Please feel free for review and giving another comment to me.

pjain1 · 2017-02-03T16:46:18Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+        public void onCompletion(RecordMetadata metadata, Exception exception) {
+          if(exception != null) {
+            log.warn(exception, "Exception is occured! Retry.");
+            emit(event);


This can keep on retrying infinitely if lets says kafka brokers are down.
In my opinion retries in case of emitting events are not necessary but if you want to retry you can set the retires property on KafkaProducer like - props.put("retries", <num retires>); which would let KafkaProducer retry by itself and throw exception if the send fails even after the retires.

@pjain1 Thanks for your review! What a great point! I'll apply your comment.

pjain1 · 2017-02-03T16:59:59Z

also it would be great if you can write some documentation for this extension. For example, here's the documentation for StatsD Emitter - https://github.com/druid-io/druid/blob/master/docs/content/development/extensions-contrib/statsd.md
You can include the extension in list of Community Extensions here - https://github.com/druid-io/druid/blob/master/docs/content/development/extensions.md

pjain1 · 2017-02-04T03:07:18Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+      HashMap<String, Object> result = jsonMapper.readValue(jsonMapper.writeValueAsString(event), typeRef);
+      result.put("clusterName", config.getClusterName());
+      producer.send(new ProducerRecord<String, String>(config.getTopic(), jsonMapper.writeValueAsString(result)),
+                    new Callback() {


Since now the callback function is same for all the events, there is no need to create a new callback object for each and every event. You can create a single class level callback object and use the same object for all the events.

@pjain1 Thanks for your kindness! I reflect your comment.

dkhwangbo · 2017-02-04T03:56:15Z

@pjain1 Thanks for your review! I just reflect your kind comment and also write documentation. Please review it.

pjain1 · 2017-02-04T04:42:29Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+  @Override
+  public void emit(final Event event) {
+    if(event instanceof ServiceMetricEvent) {
+      if(event == null) {


This condition does not make sense because if event is null then check at line 87 event instanceof ServiceMetricEvent will always be false. So you can put a simple check if(event == null) { return; } and remove the instanceof check unless you want to do different thing with different types of events (ServiceMetricEvent and AlertEvent)

Oh! thanks for correct my mistake. I'll fix it.

pjain1 · 2017-02-04T04:45:54Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+      @Override
+      public void onCompletion(RecordMetadata recordMetadata, Exception e) {
+        if(e != null) {
+          log.warn(e, "Exception is occured! Retry.");


Exception is occured! Retry. - > Exception occurred!

Thanks! I'll rewrite this message.

pjain1 · 2017-02-04T04:54:37Z

@dkhwangbo Also please file the CLA if you have not done so far - http://druid.io/community/cla.html

pjain1 · 2017-02-04T05:03:22Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitter.java

+    try {
+      TypeReference<Map<String, Object>> typeRef = new TypeReference<Map<String, Object>>() {};
+      HashMap<String, Object> result = jsonMapper.readValue(jsonMapper.writeValueAsString(event), typeRef);
+      result.put("clusterName", config.getClusterName());


since you just want to put "clusterName" property in the event map, line 93 - 95 can be replaced with -

Map<String, Object> result = event.toMap(); result.put("clusterName", config.getClusterName());

This will prevent the costly serialization and deserialization at line 94.

Got it. I'll change this line.

Hi. I try with your line but UnsupportedOperationException of guava is occured so I try another way similiar with your comment.

pjain1 · 2017-02-04T05:06:09Z

@dkhwangbo have you tested this emitter on your druid cluster ?

dkhwangbo · 2017-02-04T05:33:06Z

@pjain1 I reflect your comment! This extension is running in my druid cluster since 3 weeks ago and All functionality doing well. Also, I already done signing CLA during writing this PR

pjain1 · 2017-02-04T15:10:03Z

Closing/Reopening PR for travis

pjain1 · 2017-02-04T15:12:30Z

Changes look good to me so 👍 .. however I have one last question - Why are you not handling AlertEvent ?

dkhwangbo · 2017-02-04T15:45:56Z

@pjain1 To make this PR, I refer graphite-emitter and StatsD-emitter. Both has no handling AlertEvent. (Just graphite-emitter write error log when AlertEvent emitted). I dig a little up but druid has no documentation about AlertEvent, furthermore, AlertEvent class only used in ambari-metrics-emitter and graphite-emitter in whole druid project. So, I cannot get a sense how to handle AlertEvent appropriately.

What's the purpose of AlertEvent? When AlertEvent is emitted? What's the class AlertEvent called? If you give me the answer of my question, I really appreciate your help and support.

pjain1 · 2017-02-05T01:08:46Z

This is the AlertEvent class. It is outside of Druid repo. It is used whenever an Alert is made using log.makeAlert(). This is used for alerting critical problematic situations like Segment could not be loaded or task failed to run etc.

I am not sure why graphite-emitter is not handling alerts but for Kafka emitter Ideally alerts should go to a different topic than the metrics (ServiceMetricEvent) so that users immediately know that something is wrong in the cluster and they can take a look.

I am OK with the PR as it is, if someone wants to have alerts handled then it can be done in a follow up PR. If you want you can write the code for handling alerts in this PR as well or in a follow up PR or you can choose not to do it.

dkhwangbo · 2017-02-05T02:07:25Z

@pjain1 Oh, I got it. Thanks for your nice explanation. I'll add the code line to handle AlertEvent in another kafka topic. I'll ping you when I ready for review.

tutunci · 2017-02-05T15:12:24Z

Hi Dongkyu, I'm interested in your implementation as I've tricked current http metric emitter. What I've currently done is an http server in Go lang that receives http emitter. It parses messages and put events into a Kafka topic. Main issue is that some of the events has multiple informational like JVM data or System ones, so I've set generic fields for Druid datasource as user1 user2 an so on. A suggestion is about having a configuration file which set events that can be set into Kafka and which not. This is what I've set in my http server, avoiding to have unwanted metrics. About alerts what I've done is generating an email and a Slack notification with the error details. Mainly I use the standard event metrics to monitor Druid through Grafana charts and alerts as immediate notifications (this is why I didn't put into Kafka). Hope I helps in your implementation Thanks Maurizio Il 05 feb 2017 3:07 AM, "Dongkyu Hwangbo" <[email protected]> ha scritto:

<https://github.com/pjain1> @pjain1 <https://github.com/pjain1> Oh, I got it. Thanks for your nice

explanation. I'll add the code line to handle AlertEvent in another kafka topic. I'll ping you when I ready for review.

— You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub

<#3860 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAYbMSDRWAPh4d9vr2Rne85JTXdqQMXMks5rZS7fgaJpZM4LnmVX> .

dkhwangbo · 2017-02-06T00:13:05Z

@tutunci Hi Maurizio. Thanks for your interesting suggestion. I understand your point. In fact, I use this extension with another my own Spark streaming job. It can consume Kafka topic, metric-value-based anomaly detect, and send the metric to my timeseries DB. I think your idea also reasonable, so I'll add something like whitelist based logic.

dkhwangbo · 2017-02-06T02:01:50Z

@tutunci Hi Maurizio. Your suggestion is under construction by me, but I think origin purpose of this PR is already done. So, I think It is better finishing this PR in here and handle your suggestion in another PR. Is it Okay?
@pjain1 It's ready for your review.

tutunci · 2017-02-06T14:22:44Z

Hi Dongkyu, It's absolutely OK, of course. Just to give you some additional details, here my configuration file: enabled=sys/disk/write/count||all enabled=jvm/mem/max|memKind|all enabled=jvm/mem/committed|memKind|all enabled=jvm/mem/used|memKind|all enabled=jvm/mem/init|memKind|all enabled=jvm/pool/max|poolKind,poolName|all enabled=jvm/pool/committed|poolKind,poolName|all enabled=jvm/pool/used|poolKind,poolName|all enabled=jvm/pool/init|poolKind,poolName|all enabled=segment/loadQueue/size|server|druid:aws:coordinator enabled=segment/loadQueue/failed|server|druid:aws:coordinator enabled=segment/loadQueue/count|server|druid:aws:coordinator enabled=query/cache/total/hitRate||druid:aws:broker,druid:aws:historical enabled=query/cache/total/averageBytes||druid:aws:broker,druid:aws:historical enabled=query/cache/total/timeouts||druid:aws:broker,druid:aws:historical enabled=query/cache/total/errors||druid:aws:broker,druid:aws:historical Here what's setup is metric | additional params to set as user1,user2... | services involved So for example when I receive these metrics: {"feed":"metrics","host":"10.80.4.154:8083","metric":"*jvm/pool/init*"," *poolKind*":"heap","*poolName*":"PS Survivor Space","service":" *druid:aws:realtime* ","timestamp":"2017-02-05T23:59:54.891-05:00","value":268435456} {"feed":"metrics","host":"10.80.4.154:8083","metric":"jvm/pool/max","poolKind":"heap","poolName":"PS Old Gen","service":"druid:aws:realtime","timestamp":"2017-02-05T23:59:54.891-05:00","value":10737418240} It matches with this condition enabled=jvm/pool/init|poolKind,poolName|all because metric matches, services are all and additional user-defined fields are poolKind=>user1 and poolName=>user2 (it's positional based on the insert order). Hope it can give you some kind of help. Let me know if you create another PR, so I can follow it. Thanks Maurizio 2017-02-06 3:01 GMT+01:00 Dongkyu Hwangbo <[email protected]>:

…

@tutunci <https://github.com/tutunci> Hi Maurizio. Your suggestion is under construction by me, but I think origin purpose of this PR is already done. So, I think It is better finishing this PR in here and handle your suggestion in another PR. Is it Okay? @pjain1 <https://github.com/pjain1> It's ready for your review. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3860 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAYbMbAFNwJmyWDVrTOJEtjEFGdJ3nouks5rZn8QgaJpZM4LnmVX> .

pjain1 · 2017-02-06T18:31:11Z

extensions-contrib/kafka-emitter/src/main/java/io/druid/emitter/kafka/KafkaEmitterConfig.java

+import com.google.common.base.Preconditions;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+public class KafkaEmitterConfig {


nit: can you implement toString() method. It is useful because when the actual KafkaEmitterConfig object is created from the runtime properties, the toString() method will be used to print the config in the logs.

@pjain1 Hi! Thanks for your suggestion. I apply your comment. Thanks for your support.

… config value of kafka producer

dkhwangbo · 2017-03-27T04:17:26Z

@gianm Hi! Thanks for your reply. I addressed your comment, also I apply code formatting rule. Furthermore, I use lambda a little bit to the right place.

dkhwangbo · 2017-03-30T07:36:35Z

@fjy @b-slim @gianm Anything else?

b-slim · 2017-03-30T13:33:41Z

@dkhwangbo thanks for contribution ! this looks good to me ! good job !

erikdubbelboer · 2017-04-19T01:52:42Z

Can we maybe get some example JSON in the documentation so people know what to expect? Also maybe and example kafka consumer task that consumes this JSON back into Druid?

dkhwangbo · 2017-04-24T02:02:34Z

@erikdubbelboer Thanks for your suggestion! I'll revise document your suggestion included soon.

jon-wei · 2017-06-08T01:33:38Z

@pjain1 Can you describe what the 0.10.1 release notes should call out regarding this patch?

ambition119 · 2017-07-21T07:27:06Z

hi dkhwangbo :
In kafka-emitter code kafka version is 0.10.2.0 , I use kafka-0.8.2.1 config kafka-emitter status is ok.But use kafka-0.10.2.0 status is exception :
io.druid.emitter.kafka.KafkaEmitter - Failed to take record from queue!
java.lang.InterruptedException

dkhwangbo · 2017-07-21T07:40:23Z

@ambition119
Can I get your config file both with 0.8.2.1 and 0.10.2.0?

ambition119 · 2017-07-21T09:12:09Z

@dkhwangbo :
Use kafka-0.8.2.1 and kafka-0.10.2.0 common.runtime.properties kafka-emitter config is the same, info :
druid.extensions.loadList=["druid-kafka-eight","druid-hdfs-storage", "druid-histogram", "druid-datasketches", "druid-lookups-cached-global", "mysql-metadata-storage", "kafka-emitter"]
druid.monitoring.monitors=["com.metamx.metrics.JvmMonitor"]
druid.emitter=composing
druid.emitter.composing.emitters=["logging","kafka"]
druid.emitter.logging.logLevel=info
#druid.emitter=kafka
druid.emitter.kafka.bootstrap.servers=hdp:9092
druid.emitter.kafka.metric.topic=druid-metric
druid.emitter.kafka.alert.topic=druid-alert
druid.emitter.kafka.producer.config={"key.serializer":"org.apache.kafka.common.serialization.StringSerializer","value.serializer":"org.apache.kafka.common.serialization.StringSerializer","max.block.ms":10}

extensions/kafka-emitter directory Are not the same, kafka-emitter（0.8.2.1）info:
kafka-emitter/
kafka-emitter-0.11.0-SNAPSHOT.jar
kafka_2.10-0.8.2.1.jar
kafka-clients-0.8.2.1.jar
metrics-core-2.2.0.jar
scala-library-2.10.4.jar
snappy-java-1.1.1.6.jar
zkclient-0.3.jar
jopt-simple-3.2.jar

kafka-emitter（0.10.2.0）info:
kafka-emitter/
jopt-simple-5.0.3.jar
kafka_2.10-0.10.2.0.jar
kafka-clients-0.10.2.0.jar
kafka-emitter-0.11.0-SNAPSHOT.jar
metrics-core-2.2.0.jar
scala-library-2.10.6.jar
snappy-java-1.1.2.6.jar
zkclient-0.10.jar

thanks!

Dadard · 2017-09-28T13:39:54Z

hi @dkhwangbo

Do you still plan to work on a whitelisting system?

thanks

dkhwangbo force-pushed the kafka-emitter branch from 602b9d8 to 03a7b90 Compare January 20, 2017 02:07

dkhwangbo force-pushed the kafka-emitter branch from 1b1393a to c0c5459 Compare January 21, 2017 03:13

b-slim reviewed Feb 1, 2017

View reviewed changes

dkhwangbo force-pushed the kafka-emitter branch from e7e7b02 to 02e9ffb Compare February 2, 2017 05:01

pjain1 reviewed Feb 3, 2017

View reviewed changes

pjain1 reviewed Feb 4, 2017

View reviewed changes

pjain1 closed this Feb 4, 2017

pjain1 reopened this Feb 4, 2017

pjain1 reviewed Feb 6, 2017

View reviewed changes

dkhwangbo added 14 commits March 27, 2017 11:34

bump up druid version

01800f2

add producer.config option which make user can apply another optional…

ea12d92

… config value of kafka producer

remove potential blocking in emit()

8d1e80c

using MemoryBoundLinkedBlockingQueue

2d9178c

Fixing coding convention

e4fb6eb

Remove logging every exception and just increment counting

4d316ae

refactoring

c7e8fce

trivial modification

e1706f7

logging when callback has exception

6b383ea

Replace kafka-clients 0.10.1.1 with 0.10.2.0

1408124

Resolve the problem related of classloader

d2058d5

adopt try statement

ecb5c69

code reformatting

3a4ca9f

make variables final

8be4c4c

dkhwangbo force-pushed the kafka-emitter branch from c113f10 to 8be4c4c Compare March 27, 2017 04:11

rewrite toString

f02218a

b-slim approved these changes Mar 30, 2017

View reviewed changes

fjy merged commit 0d2e91e into apache:master Apr 4, 2017

jon-wei mentioned this pull request Jun 8, 2017

Druid 0.10.1 release notes #4384

Closed

justinborromeo mentioned this pull request Feb 15, 2019

[PROPOSAL] Add a Configurable Event Throttling Policy to Asynchronous Emitters #7075

Open

Adding Kafka-emitter #3860

Adding Kafka-emitter #3860

Conversation

dkhwangbo commented Jan 19, 2017

dkhwangbo commented Jan 19, 2017

fjy commented Jan 20, 2017

dkhwangbo commented Jan 31, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dkhwangbo commented Feb 2, 2017

pjain1 Feb 3, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pjain1 commented Feb 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dkhwangbo commented Feb 4, 2017

pjain1 Feb 4, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pjain1 commented Feb 4, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pjain1 commented Feb 4, 2017

dkhwangbo commented Feb 4, 2017

pjain1 commented Feb 4, 2017

pjain1 commented Feb 4, 2017

dkhwangbo commented Feb 4, 2017

pjain1 commented Feb 5, 2017

dkhwangbo commented Feb 5, 2017

tutunci commented Feb 5, 2017 via email

dkhwangbo commented Feb 6, 2017

dkhwangbo commented Feb 6, 2017

tutunci commented Feb 6, 2017 via email

pjain1 Feb 6, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dkhwangbo commented Mar 27, 2017

dkhwangbo commented Mar 30, 2017 • edited Loading

b-slim commented Mar 30, 2017

erikdubbelboer commented Apr 19, 2017

dkhwangbo commented Apr 24, 2017

jon-wei commented Jun 8, 2017

ambition119 commented Jul 21, 2017 • edited Loading

dkhwangbo commented Jul 21, 2017

ambition119 commented Jul 21, 2017 • edited Loading

Dadard commented Sep 28, 2017

pjain1 Feb 3, 2017 •

edited

Loading

pjain1 Feb 4, 2017 •

edited

Loading

pjain1 Feb 6, 2017 •

edited

Loading

dkhwangbo commented Mar 30, 2017 •

edited

Loading

ambition119 commented Jul 21, 2017 •

edited

Loading

ambition119 commented Jul 21, 2017 •

edited

Loading