static void execute(String... args) throws Exception {
        if (args.length != 5 && args.length != 6) {
            throw new TerseException("USAGE: java " + EndToEndLatency.class.getName()
                    + " broker_list topic num_messages producer_acks message_size_bytes [optional] properties_file");
        }

        String brokers = args[0];
        String topic = args[1];
        int numMessages = Integer.parseInt(args[2]);
        String acks = args[3];
        int messageSizeBytes = Integer.parseInt(args[4]);
        Optional<String> propertiesFile = (args.length > 5 && !Utils.isBlank(args[5])) ? Optional.of(args[5]) : Optional.empty();

        if (!Arrays.asList("1", "all").contains(acks)) {
            throw new IllegalArgumentException("Latency testing requires synchronous acknowledgement. Please use 1 or all");
        }

        try (KafkaConsumer<byte[], byte[]> consumer = createKafkaConsumer(propertiesFile, brokers);
             KafkaProducer<byte[], byte[]> producer = createKafkaProducer(propertiesFile, brokers, acks)) {

            if (!consumer.listTopics().containsKey(topic)) {
                createTopic(propertiesFile, brokers, topic);
            }
            setupConsumer(topic, consumer);
            double totalTime = 0.0;
            long[] latencies = new long[numMessages];
            Random random = new Random(0);

            for (int i = 0; i < numMessages; i++) {
                byte[] message = randomBytesOfLen(random, messageSizeBytes);
                long begin = System.nanoTime();
                //Send message (of random bytes) synchronously then immediately poll for it
                producer.send(new ProducerRecord<>(topic, message)).get();
                ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(POLL_TIMEOUT_MS));
                long elapsed = System.nanoTime() - begin;

                validate(consumer, message, records);

                //Report progress
                if (i % 1000 == 0)
                    System.out.println(i + "\t" + elapsed / 1000.0 / 1000.0);
                totalTime += elapsed;
                latencies[i] = elapsed / 1000 / 1000;
            }

            printResults(numMessages, totalTime, latencies);
            consumer.commitSync();
        }
    }
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