 package com.core;

import org.apache.log4j.Logger;

import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.BufferedReader;
import java.io.IOException;
import java.util.Properties;
import java.nio.charset.Charset;

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

public class KafkaProducer implements Runnable {
	private static final Logger LOGGER = Logger.getLogger(KafkaProducer.class);
	private final Charset ENC = Charset.forName("UTF-8");
	private final String topic;

	private InputStream inputStream = null;

	public KafkaProducer(String topic, InputStream stream) {
		this.topic = topic;
		this.inputStream = stream;
	}

	public void run() {
		Properties props = new Properties();
		props.put(KafkaParams.BROKER_LIST, KafkaParams.BROKER_LIST_VALUE);
		props.put(KafkaParams.SERIALIZER, KafkaParams.SERIALIZER_VALUE);
		props.put("producer.type", "async");
		ProducerConfig conf = new ProducerConfig(props);
		Producer<Integer, String> producer = null;
		BufferedReader rd = null;
		try {
			try {
				rd = new BufferedReader(new InputStreamReader(this.inputStream,
						ENC));
				String line = null;
				LOGGER.debug("Producing messages");
				producer = new Producer<Integer, String>(conf);
				while ((line = rd.readLine()) != null) {
					producer.send(new KeyedMessage<Integer, String>(this.topic,
							line));
				}
				LOGGER.debug("Done sending messages");
			} catch (IOException ex) {
				LOGGER.fatal("IO Error while producing messages", ex);
				LOGGER.trace(null, ex);
			}
		} catch (Exception ex) {
			LOGGER.fatal("Error while producing messages", ex);
			LOGGER.trace(null, ex);
		} finally {
			try {
				if (rd != null)
					rd.close();
				if (producer != null)
					producer.close();
			} catch (IOException ex) {
				LOGGER.fatal("IO error while cleaning up", ex);
				LOGGER.trace(null, ex);
			}
		}
	}
}

@@ -620,28 +620,34 @@
 roducer(
-String topic
+KafkaParams params
 , InputS
@@ -673,21 +673,23 @@
 his.
-topic = topic
+params = params
 ;%0A%09%09
@@ -715,16 +715,52 @@
 stream;%0A
+%09%09params.KAFKA_TOPIC_VALUE = topic;%0A
 %09%7D%0A%0A%09pub
@@ -823,30 +823,25 @@
 %09%09props.put(
-KafkaP
+p
 arams.BROKER
@@ -847,22 +847,17 @@
 R_LIST, 
-KafkaP
+p
 arams.BR
@@ -886,22 +886,17 @@
 ops.put(
-KafkaP
+p
 arams.SE
@@ -905,22 +905,17 @@
 ALIZER, 
-KafkaP
+p
 arams.SE

Update KafkaProducer.java