package com.dtstack.logstash.outputs;

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

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dtstack.logstash.annotation.Required;

/**
 * 
 * Reason: TODO ADD REASON(可选)
 * Date: 2016年8月31日 下午1:35:11
 * Company: www.dtstack.com
 * @author sishu.yss
 *
 */
public class Kafka extends BaseOutput {
	
	private static final Logger logger = LoggerFactory.getLogger(Kafka.class);

	private static ObjectMapper objectMapper = new ObjectMapper();
	
	private Producer producer;
	
	private static String encoding="utf-8";
	
	@Required(required=true)
	private static String topic;
	
	@Required(required=true)
	private static String brokerList;

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

	public void prepare() {
		try{
			Properties props = new Properties();
			props.put("metadata.broker.list",brokerList);
			props.put("key.serializer.class", "kafka.serializer.StringEncoder");
//			props.put("request.required.acks", "0");
			ProducerConfig pconfig = new ProducerConfig(props);
			producer = new Producer<>(pconfig);
		}catch(Exception e){
			logger.error(e.getMessage());
			System.exit(1);
		}
	}

	protected void emit(Map event) {
		try {
			producer.send(new KeyedMessage<>(topic, objectMapper.writeValueAsString(event).getBytes(encoding)));
		} catch (IOException e) {
			// TODO Auto-generated catch block
			logger.error(e.getMessage());
		}
	}
}

@@ -36,36 +36,8 @@
 s;%0A%0A
-import java.io.IOException;%0A
 impo
@@ -83,17 +83,16 @@
 erties;%0A
-%0A
 import k
@@ -197,16 +197,60 @@
 Config;%0A
+import org.apache.commons.lang3.StringUtils;
 %0Aimport 
@@ -702,24 +702,55 @@
 Mapper();%0A%09%0A
+%09@SuppressWarnings(%22rawtypes%22)%0A
 %09private Pro
@@ -933,253 +933,932 @@
 st;%0A
-%0A%09public Kafka(Map config) %7B%0A%09%09super(config);%0A%09%7D%0A%0A%09public void prepare() %7B%0A%09%09try%7B%0A%09%09%09Properties props = new Properties();%0A%09%09%09props.put(%22metadata.broker.list%22,brokerList);%0A%09%09%09props.put(%22key.serializer.class%22, %22kafka.serializer.StringEncoder%22
+%09%0A%09private static String keySerializer=%22kafka.serializer.StringEncoder%22;%0A%09%0A%09private static String valueSerializer=%22kafka.serializer.StringEncoder%22;%09%0A%09%0A%09private static String partitionerClass = %22kafka.producer.DefaultPartitioner%22;%0A%09%0A%09private static String producerType =%22sync%22;//sync async%0A%09%0A%09private static String compressionCodec = %22none%22;//gzip,snappy,snappy,none%0A%09%0A%09private static String clientId =%22%22;%0A%09%0A%09private static Long batchNum = 0l;%0A%09%0A%09private static Integer requestRequiredAcks=1;%0A%0A%09public Kafka(Map config) %7B%0A%09%09super(config);%0A%09%7D%0A%0A%09public void prepare() %7B%0A%09%09try%7B%0A%09%09%09Properties props = new Properties();%0A%09%09%09props.put(%22metadata.broker.list%22,brokerList);%0A%09%09%09props.put(%22key.serializer.class%22, keySerializer);%0A%09%09%09props.put(%22value.serializer.class%22, valueSerializer);%0A%09%09%09props.put(%22partitioner.class%22, partitionerClass);%0A%09%09%09props.put(%22producer.type%22, producerType);%0A%09%09%09props.put(%22compression.codec%22, compressionCodec
 );%0A
-//
 %09%09%09p
@@ -1895,13 +1895,219 @@
 s%22, 
-%220%22);
+String.valueOf(requestRequiredAcks));%0A%09%09%09if(StringUtils.isNotBlank(clientId))%7B%0A%09%09%09%09props.put(%22client.id%22, clientId);%09%09%09%0A%09%09%09%7D%0A%09%09%09if(batchNum%3E0)%7B%0A%09%09%09%09props.put(%22batch.num.messages%22, String.valueOf(batchNum));%0A%09%09%09%7D
 %0A%09%09%09
@@ -2280,16 +2280,47 @@
 %09%09%7D%0A%09%7D%0A%0A
+%09@SuppressWarnings(%22rawtypes%22)%0A
 %09protect
@@ -2472,10 +2472,8 @@
 ch (
-IO
 Exce

add kafka properties