import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
public class FlumeToKafkaSink extends AbstractSink implements Configurable {
	Producer<String, String> producer;
	String topic = "zz";
	@Override
	public Status process() throws EventDeliveryException {
		Status status = null;
		Channel channel = getChannel();
		Transaction transaction = channel.getTransaction();
		transaction.begin();
		try {
			Event event = channel.take();
			if (event==null) {
				transaction.rollback();
				status = Status.BACKOFF;
				return status;
			}
			byte[] body = event.getBody();
			final String msg = new String(body);
			String str[] = msg.split("#XT#");
			if(str[str.length - 1].equals("td")){
				final KeyedMessage<String, String> message = new KeyedMessage<String, String>(topic , msg);
				producer.send(message);
				transaction.commit();
				status = Status.READY;  
			}else{
				transaction.commit();
				status = Status.READY; 
			}
		} catch (Exception e) {
			transaction.rollback();
			status = Status.BACKOFF;
		} finally {
			transaction.close();
		}
		return status;
	}
	@Override
	public void configure(Context arg0) {
		Properties prop = new Properties();
		prop.put("zookeeper.connect", "Master:2181,Slave:2181,Slave2:2181");
		prop.put("metadata.broker.list", "Master:9092,Slave:9092,Slave2:9092");
		prop.put("serializer.class", StringEncoder.class.getName());
		producer = new Producer<String, String>(new ProducerConfig(prop));
	}
}