default command line values, javadoc work, cleanup

Don Tregonning · Don Tregonning · commit 5df24a9197ba · 2017-11-27T15:23:18.000-08:00
diff --git a/.gitignore b/.gitignore
@@ -7,5 +7,5 @@
 
 ### JAVA ###
 *.class
-data-gen.log
+*.log
 
diff --git a/src/main/java/generator/CommandLineParams.java b/src/main/java/generator/CommandLineParams.java
@@ -3,46 +3,58 @@
 import org.kohsuke.args4j.Option;
 
 /**
- * Created by dtregonning on 9/13/17.
+ * CommandLineParams is the class implementation of commandline parameters for the kafka-data-gen application.
+ *
+ * @author      Don Tregonning (dtregonning)
+ * @version     1.0
+ * @since       1.0
  */
 public class CommandLineParams {
-    @Option(name="-message-count", usage="Sets amount of messages to create")
+    @Option(name="-message-count", usage="(Required)Sets amount of messages to create and send to kafka topic")
     public String messageCount;
 
-    @Option(name="-message-size", usage="Sets size of messages to create")
+    @Option(name="-message-size", usage="(Required)Sets size of messages to create")
     public String messageSize;
 
-    @Option(name="-eps", usage="Amount of events per second to send to Kafka")
+    @Option(name="-eps", usage="(Required)Amount of events per second to send to Kafka")
     public String eps;
 
-    @Option(name="-worker-thread-count", usage="Kafka buffer amount")
+    @Option(name="-worker-thread-count", usage="Event generating worker threads, default = 4")
     public String workerThreadCount;
 
-    @Option(name="-topic", usage="Kafka Topic tp send messages to")
+    @Option(name="-topic", usage="(Required)Kafka Topic to send messages to")
     public String topic;
 
-    @Option(name="-bootstrap.servers", usage="Kafka Servers to send messages to")
+    @Option(name="-bootstrap.servers", usage="(Required) Kafka Servers to send messages to")
     public String bootStrapServers;
 
-    @Option(name="-acks", usage="Acknowledgement Scheme (all, 1, 0)")
+    @Option(name="-acks", usage="Acknowledgement Scheme (all, 1, 0),  default = all")
     public String acks;
 
-    @Option(name="-kafka-retries", usage="Kafka retries amount")
+    @Option(name="-kafka-retries", usage="Kafka retries amount,  default = 0")
     public String retries;
 
-    @Option(name="-kafka-batch-size", usage="Kafka batch size amount")
+    @Option(name="-kafka-batch-size", usage="Kafka batch size amount,  default = 1000")
     public String kafkaBatchSize;
 
-    @Option(name="-kafka-linger", usage="Kafka linger setting(ms)")
+    @Option(name="-kafka-linger", usage="Kafka linger setting(ms) ,  default = 1ms")
     public String kafkaLingerms;
 
-    @Option(name="-kafka-buffer-memory", usage="Kafka buffer amount")
+    @Option(name="-kafka-buffer-memory", usage="Kafka buffer amount,  default = 16384")
     public String kafkaBufferMemory;
 
-    public void run() {
-        System.out.println("Command Line Parameters");
-        System.out.println("- message-count: " + messageCount);
-        System.out.println("- message-size: " + messageSize);
-        System.out.println("- topic: " + topic);
+    public String toString() {
+        return "[Command Line Parameters]"
+        + "{ message-count: " + messageCount
+        + ", message-size: " + messageSize
+        + ", topic : " + topic
+        + ", eps: " + eps
+        + ", worker-thread-count: " + workerThreadCount
+        + ", bootstrap.servers: " + bootStrapServers
+        + ", acks: " + acks
+        + ", kafka-retries: " + retries
+        + ", kafka-batch-size: " + kafkaBatchSize
+        + ", kafka-linger: " + kafkaLingerms
+        + ", kafka-buffer-memory: " + kafkaBufferMemory + "}";
     }
 }
diff --git a/src/main/java/generator/DataGenerator.java b/src/main/java/generator/DataGenerator.java
@@ -8,11 +8,25 @@
 import org.kohsuke.args4j.CmdLineParser;
 import org.kohsuke.args4j.CmdLineException;
 
+/**
+ * DataGenerator is the main method and entry point for the kafka-data-gen application, The DataGenerator will:
+ * <ul>
+ * <li> Read and Parse in command line values</li>
+ * <li> Set any default values for parameters</li>
+ * <li> Create all threads needed and loop till finished</li>
+ * <li> Output metrics on program complete.</li>
+ * </ul>
+ *
+ * @author      Don Tregonning (dtregonning)
+ * @version     1.0
+ * @since       1.0
+ */
 public class DataGenerator {
-    private static Logger logger = LogManager.getLogger(DataGenerator.class);
+    private static final Logger logger = LogManager.getLogger(DataGenerator.class);
 
     public static void main(String[] args) {
         logger.info("Starting Kafka Data Generator");
+
         CommandLineParams params = new CommandLineParams();
         CmdLineParser parser = new CmdLineParser(params);
 
@@ -22,28 +36,40 @@ public static void main(String[] args) {
         } catch (CmdLineException e) {
             logger.error(e.getMessage());
             parser.printUsage(System.err);
+            System.exit(2);
         }
 
-        Properties props = new Properties();
-        props = parseKafkaArguments(params, props);
+        // Check for required parameters
+        if(params.bootStrapServers == null || params.topic == null){
+            logger.error("Missing required commandline parameter - quiting kafka-data-gen - Exit Status 1");
+            System.exit(1);
+        }
 
-        long startTime = System.currentTimeMillis();
+        //Set defaults for non required params. And log printout of value default or configured
+        if(params.workerThreadCount == null) { params.workerThreadCount = "4";}
+        if(params.eps == null) { params.eps = "0";}
+        if(params.messageSize == null) { params.messageSize = "256";}
+        if(params.messageCount == null) { params.messageSize = "10000";}
+        logger.info(params);
 
-        int workers;
-        if(params.workerThreadCount != null) {
-            workers = Integer.parseInt(params.workerThreadCount);
-        }else {
-            workers = 4;
-        }
-        logger.info("Worker count configured to: " + workers);
+        //Create and configure Kafka Producer variables. Store in Properties object
+        Properties props = new Properties();
+        props = parseKafkaArguments(params, props);
 
         EPSToken epsToken = new EPSToken();
+
+        /* Thread Implementation.
+        * RefreshTokenThread will be used to throttle and control message creation. If a token is not available at a
+        * given time a message cannot be created and sent. By using tokens we can control throughput. This thread will
+        * refresh tokens. Logic implemented in EPSToken().
+        * MetricsCalculatorThread is responsible for periodically reporting metrics to the log file.
+        * WorkThreads - create, bacth and sent events to a Kafka topic.
+        * */
         EPSThread thread_01 = new EPSThread("RefreshTokenThread", epsToken, props, params);
         EPSThread thread_02 = new EPSThread("MetricsCalculatorThread", epsToken, props, params);
+        EPSThread[] epsThreadArray = new EPSThread[Integer.parseInt(params.workerThreadCount)];
 
-
-
-        EPSThread[] epsThreadArray = new EPSThread[workers];
+        long startTime = System.currentTimeMillis();
 
         for(int i = 0; i < epsThreadArray.length; i++) {
             String threadName = "WorkerThread-" + i;
@@ -60,8 +86,11 @@ public static void main(String[] args) {
         }
 
         long elapsedTime = System.currentTimeMillis() - startTime;
+
+        // Program Exit statistics
         logger.info(epsToken.getMessageKey() + " messages created and sent in " + (elapsedTime / 1000) + " seconds");
         logger.info("Events Per Second of " + (long)(epsToken.getMessageKey())/(elapsedTime / 1000));
+        logger.info("Program run and complete without interruption");
     }
 
     public static Properties parseKafkaArguments(CommandLineParams params, Properties props) {
diff --git a/src/main/java/generator/EPSThread.java b/src/main/java/generator/EPSThread.java
@@ -46,19 +46,19 @@ public void run() {
                         Thread.sleep(500);
                     } while(epsTokenObj.getMessageKey() < Integer.parseInt(params.messageCount));
                     epsTokenObj.toggleFinished();
-                    logger.info("Total Message count reached, cleaning up.");
+                    logger.info("Total Message count reached, cleaning up program for exit.");
                 }
 
             } else if(thrd.getName().compareTo("MetricsCalculatorThread") == 0) {
                 do {
                     Thread.sleep(5000);
-                    metricsCalc.getMetrics();
+                    logger.info("Current Record Send Rate is: " + metricsCalc.getKafkaProducerMetrics("record-send-rate", "producer-metrics"));
                 } while (epsTokenObj.complete() == false);
             }
             else {
-
                 Producer<String, String> producer = new KafkaProducer<>(props);
-                metricsCalc.addProducer(producer);
+                if(!metricsCalc.addProducer(producer)) {logger.warn("Error adding producer for metrics Calculator, Metric Calculations may be incorrect" + thrd.getName()); }
+
                 do {
                     if (epsTokenObj.takeToken()) { shipEvent(producer, epsTokenObj, params); }
                 } while (epsTokenObj.complete() == false);
@@ -72,7 +72,11 @@ public void run() {
     }
 
     public static void shipEvent(Producer<String, String> producer,EPSToken epsTokenObj , CommandLineParams params) {
-        int sequenceNumber = epsTokenObj.getKey();
+        int sequenceNumber = epsTokenObj.getMessageKeyAndInc();
+
+        //TODO: Smarter Live Logging, hardcoded 10000 value. 10% of total messages?
+        if(sequenceNumber % 100000 == 0) { logger.info("Current message with sequence number: " + sequenceNumber); }
+
         byte[] event = createEvent(params, sequenceNumber);
         try {
             ProducerRecord<String, String> record = new ProducerRecord<>(params.topic, Integer.toString(sequenceNumber), new String(event));
diff --git a/src/main/java/generator/EPSToken.java b/src/main/java/generator/EPSToken.java
@@ -1,10 +1,6 @@
 package generator;
 
-import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.LogManager;
-
 class EPSToken {
-    private static Logger logger = LogManager.getLogger(EPSToken.class);
     private int tokenCount;
     private int messageKey;
     private boolean finished;
@@ -14,9 +10,7 @@ public EPSToken() {
         messageKey = 0;
         finished = false;
     }
-    public synchronized void increaseTokens(int tokenNumber) {
-        tokenCount += tokenNumber;
-    }
+    public synchronized void increaseTokens(int tokenNumber) { tokenCount += tokenNumber; }
 
     public synchronized boolean takeToken() {
         if(tokenCount != 0) {
@@ -26,22 +20,13 @@ public synchronized boolean takeToken() {
         return false;
     }
 
-    public synchronized void toggleFinished() {
-        finished = !finished;
-    }
-
     public boolean complete() {
         if(getTokenCount() == 0 && getFinished() == true) { return true; }
         else {return false; }
     }
-
+    public synchronized void toggleFinished() { finished = !finished; }
     public synchronized int getTokenCount() { return tokenCount; }
-    public synchronized int getKey() {
-        if(messageKey % 10000 == 0) {
-            logger.info(getMessageKey() + " events created and shipped");
-        }
-
-    return ++messageKey; }
+    public synchronized int getMessageKeyAndInc() { return ++messageKey; }
     public synchronized int getMessageKey() { return messageKey; }
     public synchronized boolean getFinished() { return finished; }
 }
diff --git a/src/main/java/generator/MetricsCalculator.java b/src/main/java/generator/MetricsCalculator.java
@@ -1,4 +1,5 @@
 package generator;
+
 import org.apache.kafka.clients.producer.*;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
@@ -7,35 +8,84 @@
 
 import java.util.Map;
 
+/**
+ * MetricsCalculator is a class responsible for producing consolidated metric counts from multiple threads and
+ * returning a string and metric based on the parameter sent through.
+
+ *
+ * @author      Don Tregonning (dtregonning)
+ * @version     1.0
+ * @since       1.0
+ */
 public class MetricsCalculator {
-    Producer<String, String>[] producerArray;
+    /** Single Log4J Logger for class. */
     private static Logger logger = LogManager.getLogger(EPSThread.class);
 
-    int arrayIndex = 0;
+    /** Array of Kafka Producers to track and return metrics for. */
+    protected static Producer<String, String>[] producerArray;
+
+    /** Current index value for the producerArray */
+    private static int producerArrayIndex = 0;
 
+    /**
+     * Constructor implements the static array producerArray with the correct size based on the configurable
+     * commandline parameter CommandLineParams.workerThreadCount.
+     *
+     * @param workerThreadCount       the amount of worker threads configured within kafka-data-gen
+
+     * @see             Producer
+     * @since           1.0
+     */
     public MetricsCalculator(int workerThreadCount) {
         producerArray = (Producer<String, String>[])new Producer[workerThreadCount];
     }
 
-    public synchronized void addProducer(Producer<String, String> producer) {
-        producerArray[arrayIndex] = producer;
-        logger.info("Im here");
-        arrayIndex++;
+    /**
+     * addProducer will add a Kafka Producer to the class array producerArray. Adding producers to this
+     * array is used as a central lookup of the current producers inside of worker threads.
+     *
+     * @param  producer the amount of worker threads configured within kafka-data-gen
+     * @return          <code>true</code>if the producer is successfully added to array.
+     *                  <code>false</code> otherwise.
+     *
+     * @since           1.0
+     */
+    public synchronized boolean addProducer(Producer<String, String> producer) {
+        try {
+            producerArray[producerArrayIndex] = producer;
+            producerArrayIndex++;
+        } catch (Exception Ex) {
+            logger.error(Ex.toString());
+            return false;
+        }
+          return true;
     }
-
-    public void getMetrics() {
-        double recordSendRate = 0;
-        for(int i = 0; i < arrayIndex; i++){
+    /**
+     * method used for fetching metrics from the following
+     * https://docs.confluent.io/current/kafka/monitoring.html#producer-metrics.
+     *
+     *
+     *
+     * @param   metricName  Name of metric to
+     * @param   metricGroup Group of metric
+     * @return              double value consolidatedMetricValue which will bring together metric value from all
+     *                      worker threads and return a single value.
+     *
+     * @since               1.0
+     */
+    public double getKafkaProducerMetrics(String metricName, String metricGroup) {
+        double consolidatedMetricValue = 0;
+        for(int i = 0; i < producerArrayIndex; i++){
             Producer<String, String> producer = producerArray[i];
             Map<MetricName, ? extends Metric> metrics = producer.metrics();
             for (Map.Entry<MetricName, ? extends Metric> entry : metrics.entrySet()) {
                 String name = entry.getKey().name();
                 String group = entry.getKey().group();
-                if (name.equalsIgnoreCase("record-send-rate") && group.equalsIgnoreCase("producer-metrics")) {
-                    recordSendRate += entry.getValue().value();
+                if (name.equalsIgnoreCase(metricName) && group.equalsIgnoreCase(metricGroup)) {
+                    consolidatedMetricValue += entry.getValue().value();
                 }
             }
         }
-        System.out.println("Current Record Send Rate is: " + recordSendRate);
+        return consolidatedMetricValue;
     }
 }
