@@ -0,0 +1,1046 @@
+package org.ctrip.ops.sysdev.outputs;%0A%0Aimport java.util.Map;%0Aimport java.util.Properties;%0A%0Aimport kafka.javaapi.producer.Producer;%0Aimport kafka.producer.KeyedMessage;%0Aimport kafka.producer.ProducerConfig;%0A%0Aimport org.apache.log4j.Logger;%0Aimport org.json.simple.JSONValue;%0A%0Apublic class Kafka extends BaseOutput %7B%0A%09private static final Logger logger = Logger%0A%09%09%09.getLogger(Kafka.class.getName());%0A%0A%09private Producer producer;%0A%09private String topic;%0A%0A%09public Kafka(Map config) %7B%0A%09%09super(config);%0A%09%7D%0A%0A%09protected void prepare() %7B%0A%09%09Properties props = new Properties();%0A%09%09props.put(%22metadata.broker.list%22,%0A%09%09%09%09(String) this.config.get(%22broker_list%22));%0A%09%09props.put(%22key.serializer.class%22, %22kafka.serializer.StringEncoder%22);%0A%09%09props.put(%22request.required.acks%22, %220%22);%0A%09%09ProducerConfig pconfig = new ProducerConfig(props);%0A%09%09this.topic = (String) this.config.get(%22topic%22);%0A%09%09producer = new Producer%3C%3E(pconfig);%0A%09%7D;%0A%0A%09@Override%0A%09public void emit(Map event) %7B%0A%09%09producer.send(new KeyedMessage%3C%3E(topic, JSONValue.toJSONString(event)%0A%09%09%09%09.getBytes()));%0A%09%7D%0A%7D%0A

add kafka output