@Override
    public synchronized void processRecords(ConsumerRecords<byte[], byte[]> consumerRecords) {
        // Save the current time to prevent skew by processing time.
        long curTime = Time.SYSTEM.milliseconds();
        for (ConsumerRecord<byte[], byte[]> record : consumerRecords) {
            try {
                buffer.clear();
                buffer.put(record.value(), 0, Long.BYTES);
                buffer.rewind();
                putHistogram(curTime - buffer.getLong());
            } catch (RuntimeException e) {
                log.error("Error in processRecords:", e);
            }
        }
    }
This class will process records containing timestamps and generate a histogram based on the data.  It will then be
present in the status from the `ConsumeBenchWorker` class.  This must be used with a timestamped PayloadGenerator
implementation.

Example spec:
{
   "type": "timestamp",
   "histogramMaxMs": 10000,
   "histogramMinMs": 0,
   "histogramStepMs": 1
}

This will track total E2E latency up to 10 seconds, using 1ms resolution and a timestamp size of 8 bytes.
trogdor/src/main/java/org/apache/kafka/trogdor/workload/TimestampRecordProcessor.java