private static void createTopic(Optional<String> propertiesFile, String brokers, String topic) throws IOException {
        System.out.printf("Topic \"%s\" does not exist. "
                        + "Will create topic with %d partition(s) and replication factor = %d%n",
                topic, DEFAULT_NUM_PARTITIONS, DEFAULT_REPLICATION_FACTOR);

        Properties adminProps = loadPropsWithBootstrapServers(propertiesFile, brokers);
        Admin adminClient = Admin.create(adminProps);
        NewTopic newTopic = new NewTopic(topic, DEFAULT_NUM_PARTITIONS, DEFAULT_REPLICATION_FACTOR);
        try {
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();
        } catch (ExecutionException | InterruptedException e) {
            System.out.printf("Creation of topic %s failed%n", topic);
            throw new RuntimeException(e);
        } finally {
            Utils.closeQuietly(adminClient, "AdminClient");
        }
    }
This class records the average end to end latency for a single message to travel through Kafka.
Following are the required arguments
<p> broker_list = location of the bootstrap broker for both the producer and the consumer </p>
<p> topic = topic name used by both the producer and the consumer to send/receive messages </p>
<p> num_messages = # messages to send </p>
<p> producer_acks = See ProducerConfig.ACKS_DOC </p>
<p> message_size_bytes = size of each message in bytes </p>

<p> e.g. [localhost:9092 test 10000 1 20] </p>
tools/src/main/java/org/apache/kafka/tools/EndToEndLatency.java