package com.airbnb.plog.kafka;

import com.airbnb.plog.Message;
import com.airbnb.plog.handlers.Handler;
import com.eclipsesource.json.JsonArray;
import com.eclipsesource.json.JsonObject;
import com.yammer.metrics.core.Meter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import kafka.common.FailedToSendMessageException;
import kafka.javaapi.producer.Producer;
import kafka.producer.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.atomic.AtomicLong;

@RequiredArgsConstructor
@Slf4j
public class KafkaHandler extends SimpleChannelInboundHandler<Message> implements Handler {
    private final String defaultTopic;
    private final Producer<byte[], byte[]> producer;
    private final AtomicLong failedToSendMessageExceptions = new AtomicLong();
    private final ProducerStats producerStats;
    private final ProducerTopicMetrics producerAllTopicsStats;

    protected KafkaHandler(final String clientId, final String defaultTopic, final Producer<byte[], byte[]> producer) {
        super();
        this.producerStats = ProducerStatsRegistry.getProducerStats(clientId);
        this.producerAllTopicsStats = ProducerTopicStatsRegistry.getProducerTopicStats(clientId).getProducerAllTopicsStats();
        this.defaultTopic = defaultTopic;
        this.producer = producer;
    }

    private static JsonObject meterToJsonObject(Meter meter) {
        return new JsonObject()
                .add("count", meter.count())
                .add("rate", new JsonArray()
                        .add(meter.oneMinuteRate())
                        .add(meter.fiveMinuteRate())
                        .add(meter.fifteenMinuteRate()));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
        try {
            producer.send(new KeyedMessage<byte[], byte[]>(defaultTopic, msg.asBytes()));
        } catch (FailedToSendMessageException e) {
            log.warn("Failed to send message", e);
            failedToSendMessageExceptions.incrementAndGet();
        }
    }

    @Override
    public JsonObject getStats() {
        return new JsonObject()
                .add("default_topic", defaultTopic)
                .add("failed_to_send", failedToSendMessageExceptions.get())
                .add("failed_send_rate", meterToJsonObject(producerStats.failedSendRate()))
                .add("resend_rate", meterToJsonObject(producerStats.resendRate()))
                .add("serialization_error_rate", meterToJsonObject(producerStats.serializationErrorRate()))
                .add("message_rate", meterToJsonObject(producerAllTopicsStats.messageRate()))
                .add("dropped_message_rate", meterToJsonObject(producerAllTopicsStats.droppedMessageRate()))
                .add("byte_rate", meterToJsonObject(producerAllTopicsStats.byteRate()));
    }

    @Override
    public final String getName() {
        return "kafka";
    }
}

@@ -832,16 +832,49 @@
 ptions =
+ new AtomicLong(), seenMessages =
  new Ato
@@ -1892,14 +1892,580 @@
     
-try %7B%0A
+seenMessages.incrementAndGet();%0A        final byte%5B%5D payload = msg.asBytes();%0A%0A        boolean sawKtTag = false;%0A%0A        for (String tag : msg.getTags())%0A            if (tag.startsWith(%22kt:%22)) %7B%0A                sawKtTag = true;%0A                sendOrReportFailure(tag.substring(3), payload);%0A            %7D%0A%0A        if (!sawKtTag)%0A            sendOrReportFailure(defaultTopic, payload);%0A    %7D%0A%0A    private boolean sendOrReportFailure(String topic, byte%5B%5D msg) %7B%0A        final boolean nonNullTopic = !(%22null%22.equals(topic));%0A        if (nonNullTopic)%0A            try %7B%0A    
     
@@ -2519,24 +2519,17 @@
 byte%5B%5D%3E(
-defaultT
+t
 opic, ms
@@ -2533,22 +2533,16 @@
  msg
-.asBytes()
 ));%0A
+    
     
@@ -2600,16 +2600,20 @@
         
+    
 log.warn
@@ -2633,22 +2633,37 @@
 end 
-message%22, e);%0A
+to topic %7B%7D%22, topic, e);%0A    
     
@@ -2727,17 +2727,50 @@
         
+   
-%7D
+ %7D%0A        return nonNullTopic;
 %0A    %7D%0A%0A
@@ -2898,24 +2898,82 @@
 faultTopic)%0A
+                .add(%22seen_messages%22, seenMessages.get())%0A
             

Support for kt: tags in plog-kafka