private static KafkaConsumer<byte[], byte[]> createKafkaConsumer(Optional<String> propsFile, String brokers) throws IOException {
        Properties consumerProps = loadPropsWithBootstrapServers(propsFile, brokers);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-" + System.currentTimeMillis());
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        consumerProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "0"); //ensure we have no temporal batching
        return new KafkaConsumer<>(consumerProps);
    }
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