package com.ctrip.ops.sysdev.outputs;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import com.ctrip.ops.sysdev.baseplugin.BaseOutput;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.ctrip.ops.sysdev.render.TemplateRender;

import lombok.extern.log4j.Log4j2;
import org.json.simple.JSONValue;

@Log4j2
public class Kafka extends BaseOutput {

    private Producer producer;
    private String topic;
    private Properties props;
    private TemplateRender topicRender;

    public Kafka(Map config) {
        super(config);
    }

    protected void prepare() {
        if (!this.config.containsKey("topic")) {
            log.error("topic must be included in config");
            System.exit(1);
        }
        this.topic = (String) this.config.get("topic");
        try {
            this.topicRender = TemplateRender.getRender(topic);
        } catch (IOException e) {
            log.fatal("could not build template from" + topic);
            System.exit(1);
        }
        props = new Properties();

        HashMap<String, String> producerSettings = (HashMap<String, String>) this.config.get("producer_settings");

        if (producerSettings != null) {
            producerSettings.entrySet().stream().forEach(entry -> {
                String k = entry.getKey();
                String v = entry.getValue();
                props.put(k, v);
            });
        } else {
            log.error("producer_settings must be included in config");
            System.exit(1);
        }

        if (props.get("bootstrap.servers") == null) {
            log.error("bootstrap.servers must be included in producer_settings");
            System.exit(1);
        }

        if (props.get("key.serializer") == null) {
            props.put("key.serializer",
                    "org.apache.kafka.common.serialization.StringSerializer");
        }
        if (props.get("value.serializer") == null) {
            props.put("value.serializer",
                    "org.apache.kafka.common.serialization.StringSerializer");
        }
        producer = new KafkaProducer<>(props);
    }

    protected void emit(Map event) {
        String _topic = topicRender.render(event).toString();
        producer.send(new ProducerRecord<String, String>(_topic, JSONValue.toJSONString(event)));
    }

    public void shutdown() {
        log.info("close producer and then shutdown");
        producer.close();
    }
}

@@ -581,16 +581,42 @@
  topic;%0A
+    private String field;%0A
     priv
@@ -1178,32 +1178,150 @@
 t(1);%0A        %7D%0A
+        if (this.config.containsKey(%22field%22)) %7B%0A            this.field = (String) this.config.get(%22field%22);%0A        %7D%0A
         props = 
@@ -2515,24 +2515,62 @@
 toString();%0A
+        if (this.field == null) %7B%0A    
         prod
@@ -2655,16 +2655,213 @@
 ent)));%0A
+        %7D else %7B%0A            if (event.containsKey(this.field)) %7B%0A                producer.send(new ProducerRecord%3CString, String%3E(_topic, (String) event.get(this.field)));%0A            %7D%0A        %7D%0A
     %7D%0A%0A 

Update output kafka