A TimestampExtractor determines which timestamp Kafka Streams assigns to an input record. That choice controls whether processing follows event time, ingestion time, or processing time. Event time usually comes from the producer or the event itself; ingestion time comes from broker log append time; processing time comes from pipeline execution. section 9.4
Choose the timestamp source that matches the event-streaming scenario:
- Use the ProducerRecord metadata when producer-created event time is sufficient. section 9.4 section 9.5
- Use broker metadata when LogAppendTime should represent ingestion time. section 9.4 section 9.5
- Use a custom extractor when the value contains a more accurate or calculated event timestamp. section 9.5.2
- Select invalid-timestamp behavior with FailOnInvalidTimestamp, LogAndSkipOnInvalidTimestamp, or UsePartitionTimeOnInvalidTimestamp. section 9.5
Extracting purchase time from an event
A transaction extractor reads the deserialized PurchaseTransaction from ConsumerRecord.value() and returns purchaseTransaction.transactionTime(). This preserves the actual purchase time when delivery is delayed or records arrive out of order, supporting time-sensitive joins. section 9.5.2
Listing 9.27 Custom TimestampExtractor
public class TransactionTimestampExtractor implements TimestampExtractor {
@Override
public long extract(ConsumerRecord<Object, Object>
consumerRecord,
long partitionTime) {
PurchaseTransaction purchaseTransaction =
(PurchaseTransaction) consumerRecord.value(); #1
return purchaseTransaction.transactionTime(); #2
}
}
The listing shows the minimal custom implementation: cast the record value, then return its transaction timestamp. section 9.5.2
Configure extraction globally with DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, or attach an extractor to one input source through Consumed.with(...).withTimestampExtractor(...). Per-source configuration keeps different input topics independent. The Processor API accepts the extractor directly in Topology.addSource(). section 9.5.3 section 10.1.1
