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 + "]");
        }
    }
@Test
    @SuppressWarnings("unchecked")
    public void shouldPassInValidation() {
        Iterator<ConsumerRecord<byte[], byte[]>> iterator = mock(Iterator.class);
        ConsumerRecord<byte[], byte[]> record = mock(ConsumerRecord.class);
        when(records.isEmpty()).thenReturn(false);
        when(records.iterator()).thenReturn(iterator);
        when(iterator.next()).thenReturn(record);
        when(record.value()).thenReturn("kafkaa".getBytes(StandardCharsets.UTF_8));
        when(records.count()).thenReturn(1);
        assertDoesNotThrow(() -> EndToEndLatency.validate(consumer, "kafkaa".getBytes(StandardCharsets.UTF_8), records));
    }