class MetricDataTimestampExtractor implements TimestampExtractor {

    @Override
    public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
        val metricData = (MetricData) record.value();
        if (metricData == null) {
            // -1 skips the record. Don't log as it can fill up the logs.
            return -1L;
        }
        return metricData.getTimestamp() * 1000L;
    }

}