Wrong delay calculation

[sisinduku]

Currently Ziggurat is using WallclockTimestampExtractor stream processor. If you see WallclockTimestampExtractor definition

public class WallclockTimestampExtractor implements TimestampExtractor {

    /**
     * Return the current wall clock time as timestamp.
     *
     * @param record a data record
     * @param previousTimestamp the latest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)
     * @return the current wall clock time, expressed in milliseconds since midnight, January 1, 1970 UTC
     */
    @Override
    public long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
        return System.currentTimeMillis();
    }

In that case, the timestamp will be overridden by current timestamp. Hence when you call delay calculation in kafka_delay.clj:
(calculate-and-report-kafka-delay metric-namespace (.timestamp processor-context))
It will use current timestamp when it is invoked. It will make delay calculation incorrect