/*
 * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.wso2.carbon.sample.kafka.performance;

import com.google.gson.JsonObject;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.log4j.Logger;

import java.text.DecimalFormat;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;


public class KafkaClient {
    private static Logger log = Logger.getLogger(KafkaClient.class);

    private static String url;
    private static String topic;
    private static long noOfEvents;
    private AtomicLong count;
    private static int elapsedCount;
    private static int noOfPublishers;

    public KafkaClient() {
        count = new AtomicLong(0);
    }

    public static void main(String args[]) {
        log.info("Starting Kafka Client");
        url = args[0];
        topic = args[1];
        try {
            noOfEvents = Long.parseLong(args[2]);
            elapsedCount = Integer.parseInt(args[3]);
            noOfPublishers = Integer.parseInt(args[4]);

            KafkaClient kafkaClient = new KafkaClient();
            kafkaClient.start();
        } catch (NumberFormatException e) {
            log.error("Entered value for no of events is invalid. Please enter an integer", e);
        }
    }

    private void start() {
        ExecutorService executor = Executors.newFixedThreadPool(noOfPublishers);
        for (int i = 0; i < noOfPublishers; i++) {
            executor.execute(new KafkaProducer());
        }
    }

    public class KafkaProducer implements Runnable {

        @Override
        public void run() {
            try {
                Properties props = new Properties();
                props.put("metadata.broker.list", url);
                props.put("serializer.class", "kafka.serializer.StringEncoder");

                ProducerConfig config = new ProducerConfig(props);
                Producer<String, Object> producer = new Producer<String, Object>(config);

                log.info("Sending messages..");
                long lastTime = System.currentTimeMillis();
                DecimalFormat decimalFormat = new DecimalFormat("#");
                while (count.getAndIncrement() < noOfEvents) {
                    String message = "{\"event\": " + getRandomEvent(count.get()).toString() + "}";

                    KeyedMessage<String, Object> data = new KeyedMessage<String, Object>(topic, message);
                    producer.send(data);

                    if (count.get() % elapsedCount == 0) {
                        long currentTime = System.currentTimeMillis();
                        long elapsedTime = currentTime - lastTime;
                        double throughputPerSecond = (((double) elapsedCount) / elapsedTime) * 1000;
                        lastTime = currentTime;
                        log.info("Sent " + elapsedCount + " sensor events in " + elapsedTime +
                                " milliseconds with total throughput of " + decimalFormat.format(throughputPerSecond) +
                                " events per second.");
                    }
                }
            } catch (Throwable t) {
                log.error("Error when sending the messages", t);
            }
        }

        private JsonObject getRandomEvent(long count) {

            JsonObject event = new JsonObject();

            JsonObject metaData = new JsonObject();
            JsonObject correlationData = new JsonObject();
            JsonObject payLoadData = new JsonObject();

            metaData.addProperty("timestamp", System.currentTimeMillis());
            metaData.addProperty("isPowerSaverEnabled", false);
            metaData.addProperty("sensorId", count);
            metaData.addProperty("sensorName", "temperature");

            correlationData.addProperty("longitude", 2332.424);
            correlationData.addProperty("latitude", 2323.23232);

            payLoadData.addProperty("humidity", 2.3f);
            payLoadData.addProperty("sensorValue", 23423.234);

            event.add("metaData", metaData);
            event.add("correlationData", correlationData);
            event.add("payloadData", payLoadData);

            return event;
        }
    }


}


@@ -1009,16 +1009,66 @@
 cutors;%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0A
 import j
@@ -1309,38 +1309,8 @@
 ts;%0A
-    private AtomicLong count;%0A
     
@@ -1390,63 +1390,284 @@
     
-public KafkaClient() %7B%0A        count = new AtomicLong(0
+//Atomic Variables%0A    private AtomicLong count;%0A    private AtomicLong lastIndex;%0A    private AtomicBoolean calcInProgress;%0A%0A%0A    public KafkaClient() %7B%0A        count = new AtomicLong(0);%0A        lastIndex = new AtomicLong(0);%0A        calcInProgress = new AtomicBoolean(false
 );%0A 
@@ -3438,12 +3438,21 @@
     
-if (
+long index = 
 coun
@@ -3463,9 +3463,9 @@
 t() 
-%25
+/
  ela
@@ -3477,17 +3477,190 @@
 ount
- == 0) %7B%0A
+;%0A%0A                    if (lastIndex.get() != index) %7B%0A                        if (calcInProgress.compareAndSet(false, true)) %7B%0A                            lastIndex.set(index);%0A    
     
@@ -3718,32 +3718,36 @@
 ntTimeMillis();%0A
+    
                 
@@ -3821,16 +3821,20 @@
         
+    
 double t
@@ -3922,24 +3922,28 @@
             
+    
 lastTime = c
@@ -3970,32 +3970,36 @@
                 
+ 
+   
 log.info(%22Sent %22
@@ -4053,16 +4053,20 @@
 dTime +%0A
+    
         
@@ -4209,16 +4209,20 @@
         
+    
 %22 events
@@ -4233,24 +4233,105 @@
  second.%22);%0A
+                            calcInProgress.set(false);%0A                        %7D%0A
             

Make kafka performance client thread safe