package JNode;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

class Producer
{
    private final kafka.javaapi.producer.Producer kafka;
    public Producer()
    {
        Properties props = new Properties();
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        String brokerList = System.getenv("BROKER_LIST");
        if(brokerList == null) {
            brokerList = "localhost:9092";
        }
        props.put("metadata.broker.list", brokerList);
        this.kafka = new kafka.javaapi.producer.Producer(new ProducerConfig(props));
    }

    public void send(String topic, String messageStr) {
        KeyedMessage msg = new KeyedMessage(topic, messageStr);
        kafka.send(msg);
    }
}

@@ -81,16 +81,48 @@
 rConfig;
+%0Aimport org.apache.log4j.Logger;
 %0A%0Aimport
@@ -211,22 +211,104 @@
 Producer
- kafka
+%3CString,String%3E kafka;%0A    final static Logger logger = Logger.getLogger(Producer.class)
 ;%0A    pu
@@ -702,16 +702,31 @@
 Producer
+%3CString,String%3E
 (new Pro
@@ -829,16 +829,44 @@
 dMessage
+%3CString,String%3E msg;%0A       
  msg = n
@@ -880,16 +880,31 @@
 dMessage
+%3CString,String%3E
 (topic, 
@@ -928,24 +928,166 @@
     
-kafka.send(msg);
+try %7B%0A            kafka.send(msg);%0A        %7Dcatch( Exception e) %7B%0A            logger.error(%22cannot send msg %22 + msg, e);%0A            System.exit(1);%0A        %7D
 %0A   

exit on kafka exception