@Override
        public long extract(final ConsumerRecord<Object, Object> record, final long partitionTime) {
            if (record.value().toString().matches(".*@[0-9]+")) {
                return Long.parseLong(record.value().toString().split("@")[1]);
            }

            if (record.timestamp() >= 0L) {
                return record.timestamp();
            }

            return DEFAULT_TIMESTAMP;
        }
A custom timestamp extractor that extracts the timestamp from the record's value if the value is in ".*@[0-9]+"
format. Otherwise, it returns the record's timestamp or the default timestamp if the record's timestamp is negative.
streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java