package kfk.load;

import com.beust.jcommander.Parameter;
import com.google.common.base.Strings;
import com.google.common.net.HostAndPort;

/**
 */
public class Args {
    @Parameter(names = {"-h", "--help"}, description = "Display arguments usage help message", help = true)
    private boolean help;

    @Parameter(names = {"--threads"}, description = "Number of concurrent load generators")
    private int threads = Runtime.getRuntime().availableProcessors() - 1;

    @Parameter(names = {"--size"}, description = "Size of random message in kilo bytes")
    private int messageSize = 1024 * 2;

    @Parameter(names = {"-t", "--topic"}, description = "Name of topic")
    private String topic = "testTopic";

    @Parameter(names = {"-c", "--cluster"}, description = "Kafka cluster DNS entry", required = true)
    private String cluster;

    @Parameter(names = {"--port"}, description = "Kafka servers listening port")
    private int port = 9092;

    @Parameter(names = {"--message-serializer"}, description = "Kafka message serializer class, must be in classpath")
    private String messageSerializerClass = "org.apache.kafka.common.serialization.StringSerializer";

    public int messageSize() {
        return messageSize;
    }

    public int threads() {
        return threads;
    }

    public String topic() {
        return topic;
    }

    public boolean help() {
        return help;
    }

    public void verify() {
        if (Strings.isNullOrEmpty(cluster)) {
            throw new RuntimeException("Invalid kafka cluster: " + cluster);
        }
    }

    public String messageSerializerClass() {
        return messageSerializerClass;
    }

    public HostAndPort kafkaCluster() {
        return HostAndPort.fromParts(cluster, port);
    }
}

@@ -51,16 +51,61 @@
 ameter;%0A
+import com.google.common.base.Preconditions;%0A
 import c
@@ -505,12 +505,8 @@
 rs()
- - 1
 ;%0A%0A 
@@ -1608,16 +1608,109 @@
        %7D
+%0A%0A        Preconditions.checkArgument(threads %3C 1, %22Invalid number of threads: %25s%22, threads);
 %0A    %7D%0A%0A

Fix threads arg issue