Introduction

Kafka Streams uses timestamps to decide how records participate in windowed aggregations. If the event time you care about lives inside the message payload instead of in Kafka record metadata, you need a custom TimestampExtractor so your aggregation windows follow business event time rather than broker time.

Why Timestamp Extraction Matters

Windowed operations such as tumbling, hopping, and session windows all depend on a timestamp. By default, Kafka Streams can use the record timestamp that Kafka stores with the message, but that may not represent when the business event actually happened.

Typical cases where event time is better:

• IoT devices send delayed measurements
• mobile clients buffer events offline
• upstream services publish historical backfills

If you aggregate by ingestion time instead of event time, your windows can become misleading.

A Custom TimestampExtractor

The extractor must return milliseconds since the Unix epoch:

1import org.apache.kafka.clients.consumer.ConsumerRecord;
2import org.apache.kafka.streams.processor.TimestampExtractor;
3
4public class EventTimeExtractor implements TimestampExtractor {
5    @Override
6    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
7        Event event = (Event) record.value();
8
9        if (event == null || event.getEventTimeMillis() <= 0) {
10            return partitionTime;
11        }
12
13        return event.getEventTimeMillis();
14    }
15}

Using partitionTime as a fallback is often safer than throwing when a bad record arrives, though the right policy depends on your data quality requirements.

Wiring It into Kafka Streams

You can configure the extractor globally:

1import java.util.Properties;
2import org.apache.kafka.streams.StreamsBuilder;
3import org.apache.kafka.streams.StreamsConfig;
4
5Properties props = new Properties();
6props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
7props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
8props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, EventTimeExtractor.class.getName());
9
10StreamsBuilder builder = new StreamsBuilder();

With that configuration, timestamp-based operations across the topology will use the extracted event time by default.

You can also attach an extractor at the Consumed level for a specific source when you do not want one global extractor to apply to every input topic.

Example Aggregation

Here is a simple five-minute windowed count:

1import java.time.Duration;
2import org.apache.kafka.common.serialization.Serdes;
3import org.apache.kafka.streams.kstream.Consumed;
4import org.apache.kafka.streams.kstream.Grouped;
5import org.apache.kafka.streams.kstream.KStream;
6import org.apache.kafka.streams.kstream.Materialized;
7import org.apache.kafka.streams.kstream.TimeWindows;
8
9KStream<String, Event> events = builder.stream(
10    "events",
11    Consumed.with(Serdes.String(), eventSerde).withTimestampExtractor(new EventTimeExtractor())
12);
13
14events.groupByKey(Grouped.with(Serdes.String(), eventSerde))
15      .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
16      .count(Materialized.as("event-counts"));

The important point is that the five-minute windows are now based on event.getEventTimeMillis().

Late Records and Grace Periods

Custom extraction changes what "late" means. A record can arrive now but belong to an older window if its event-time field is old.

That is why grace periods matter:

.windowedBy(
    TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1))
)

With a grace period, slightly late events can still update the intended window instead of being dropped as too late.

Validating the Extracted Timestamp

Do not assume every record contains a valid timestamp. Good extractors usually protect against:

• null payloads
• negative timestamps
• missing fields
• timestamps in seconds instead of milliseconds

A defensive version:

1if (event == null) {
2    return partitionTime;
3}
4
5long ts = event.getEventTimeMillis();
6if (ts <= 0) {
7    return partitionTime;
8}
9
10return ts;

The last thing you want is one malformed record breaking a stateful topology unnecessarily.

Common Pitfalls

The biggest mistake is returning seconds instead of milliseconds. Kafka Streams expects epoch milliseconds, so a seconds-based timestamp silently shifts windows by a factor of one thousand.

Another issue is using event time without thinking about late data. Once you switch from broker time to payload time, out-of-order records become normal, not exceptional, and your window grace settings need to reflect that.

Finally, avoid throwing away bad records inside the extractor without observability. If the extractor falls back or drops records, log or monitor that behavior so window results do not become mysterious later.

Summary

• A custom TimestampExtractor lets Kafka Streams aggregate by event time instead of broker timestamp.
• The extractor must return epoch milliseconds.
• Use it when the true business timestamp is stored inside the record payload.
• Combine event-time extraction with appropriate grace periods for late data.
• Validate malformed or missing timestamps so one bad record does not destabilize the topology.