// Path: hadoop-mini-clusters-kafka/src/test/java/com/github/sakserv/minicluster/datatime/GenerateRandomDay.java
// public class GenerateRandomDay {
// 
//     public static String genRandomDay() {
// 
//         GregorianCalendar gc = new GregorianCalendar();
// 
//         int year = randBetween(2013, 2014);
// 
//         gc.set(gc.YEAR, year);
// 
//         int dayOfYear = randBetween(1, gc.getActualMaximum(gc.DAY_OF_YEAR));
// 
//         gc.set(gc.DAY_OF_YEAR, dayOfYear);
// 
//         return String.format("%04d-%02d-%02d", gc.get(gc.YEAR), gc.get(gc.MONTH), gc.get(gc.DAY_OF_MONTH));
// 
//     }
// 
//     public static int randBetween(int start, int end) {
//         return start + (int)Math.round(Math.random() * (end - start));
//     }
// }


import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.sakserv.minicluster.datatime.GenerateRandomDay;
            return this;
        }

        public KafkaSimpleTestProducer build() {
            KafkaSimpleTestProducer kafkaSimpleTestProducer = new KafkaSimpleTestProducer(this);
            return kafkaSimpleTestProducer;
        }

    }

    public Map<String, Object> createConfig() {
        Map<String, Object> config = new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaHostname() + ":" + getKafkaPort());
        config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return config;
    }

    public void produceMessages() {

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(createConfig());

        int count = 0;
        while(count < getMessageCount()) {

            // Create the JSON object
            JSONObject obj = new JSONObject();
            try {
                obj.put("id", String.valueOf(count));
                obj.put("msg", "test-message" + 1);
// Path: hadoop-mini-clusters-kafka/src/test/java/com/github/sakserv/minicluster/datatime/GenerateRandomDay.java
// public class GenerateRandomDay {
// 
//     public static String genRandomDay() {
// 
//         GregorianCalendar gc = new GregorianCalendar();
// 
//         int year = randBetween(2013, 2014);
// 
//         gc.set(gc.YEAR, year);
// 
//         int dayOfYear = randBetween(1, gc.getActualMaximum(gc.DAY_OF_YEAR));
// 
//         gc.set(gc.DAY_OF_YEAR, dayOfYear);
// 
//         return String.format("%04d-%02d-%02d", gc.get(gc.YEAR), gc.get(gc.MONTH), gc.get(gc.DAY_OF_MONTH));
// 
//     }
// 
//     public static int randBetween(int start, int end) {
//         return start + (int)Math.round(Math.random() * (end - start));
//     }
// }


// Path: hadoop-mini-clusters-kafka/src/test/java/com/github/sakserv/minicluster/kafka/producer/KafkaSimpleTestProducer.java
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.sakserv.minicluster.datatime.GenerateRandomDay;

            return this;
        }

        public KafkaSimpleTestProducer build() {
            KafkaSimpleTestProducer kafkaSimpleTestProducer = new KafkaSimpleTestProducer(this);
            return kafkaSimpleTestProducer;
        }

    }

    public Map<String, Object> createConfig() {
        Map<String, Object> config = new HashMap<String, Object>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getKafkaHostname() + ":" + getKafkaPort());
        config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return config;
    }

    public void produceMessages() {

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(createConfig());

        int count = 0;
        while(count < getMessageCount()) {

            // Create the JSON object
            JSONObject obj = new JSONObject();
            try {
                obj.put("id", String.valueOf(count));
                obj.put("msg", "test-message" + 1);
                obj.put("dt", GenerateRandomDay.genRandomDay());