static void validate(KafkaConsumer<byte[], byte[]> consumer, byte[] message, ConsumerRecords<byte[], byte[]> records) {
        if (records.isEmpty()) {
            consumer.commitSync();
            throw new RuntimeException("poll() timed out before finding a result (timeout:[" + POLL_TIMEOUT_MS + "])");
        }

        //Check result matches the original record
        String sent = new String(message, StandardCharsets.UTF_8);
        String read = new String(records.iterator().next().value(), StandardCharsets.UTF_8);

        if (!read.equals(sent)) {
            consumer.commitSync();
            throw new RuntimeException("The message read [" + read + "] did not match the message sent [" + sent + "]");
        }

        //Check we only got the one message
        if (records.count() != 1) {
            int count = records.count();
            consumer.commitSync();
            throw new RuntimeException("Only one result was expected during this test. We found [" + count + "]");
        }
    }
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