Introduction

Kafka Streams uses a timestamp for every record, and that timestamp drives event-time features such as windows, joins, and suppression. If the meaningful event time is stored inside the record payload rather than in the Kafka record metadata, you need a custom TimestampExtractor.

Why Embedded Timestamps Matter

A producer record already has a Kafka timestamp, but that value may represent producer time or broker append time. In many streaming systems, the real business event time lives inside the message body, such as an eventTime field serialized in JSON.

If you use the wrong timestamp source, your windows and aggregations can still run, but they will reflect the wrong notion of time.

Implement a Custom TimestampExtractor

Kafka Streams lets you plug in your own extractor by implementing TimestampExtractor. The method receives the ConsumerRecord and the current partition time.

1package com.example.streams;
2
3import com.fasterxml.jackson.databind.JsonNode;
4import com.fasterxml.jackson.databind.ObjectMapper;
5import org.apache.kafka.clients.consumer.ConsumerRecord;
6import org.apache.kafka.streams.processor.TimestampExtractor;
7
8public class JsonEventTimestampExtractor implements TimestampExtractor {
9    private static final ObjectMapper MAPPER = new ObjectMapper();
10
11    @Override
12    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
13        try {
14            String json = record.value().toString();
15            JsonNode root = MAPPER.readTree(json);
16            long timestamp = root.get("eventTime").asLong();
17
18            if (timestamp < 0) {
19                throw new IllegalArgumentException("Negative eventTime");
20            }
21
22            return timestamp;
23        } catch (Exception ex) {
24            throw new RuntimeException("Could not extract event time", ex);
25        }
26    }
27}

The extractor must return milliseconds since the Unix epoch. That is the contract Kafka Streams expects.

Register the Extractor in Streams Config

After implementing the extractor, wire it into the application configuration with StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG.

1package com.example.streams;
2
3import java.util.Properties;
4import org.apache.kafka.streams.StreamsConfig;
5
6public class StreamsConfiguration {
7    public static Properties build() {
8        Properties props = new Properties();
9        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
10        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
11        props.put(
12                StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
13                JsonEventTimestampExtractor.class
14        );
15        return props;
16    }
17}

With this in place, source records will be timestamped from the embedded field instead of from the default record timestamp behavior.

A Minimal Topology

The topology itself can remain simple. The extractor changes the time semantics without changing the rest of the DSL code.

1package com.example.streams;
2
3import org.apache.kafka.common.serialization.Serdes;
4import org.apache.kafka.streams.KafkaStreams;
5import org.apache.kafka.streams.StreamsBuilder;
6import org.apache.kafka.streams.kstream.TimeWindows;
7
8import java.time.Duration;
9
10public class OrdersApp {
11    public static void main(String[] args) {
12        StreamsBuilder builder = new StreamsBuilder();
13
14        builder.stream("orders")
15                .groupByKey()
16                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
17                .count()
18                .toStream()
19                .foreach((key, value) -> System.out.println(key + " -> " + value));
20
21        KafkaStreams streams = new KafkaStreams(builder.build(), StreamsConfiguration.build());
22        streams.start();
23        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
24    }
25}

If your payload timestamps are correct, these five-minute windows now use event time from the message body.

What to Do With Bad Timestamps

This is the most important design decision. Kafka Streams allows several behaviors:

• throw an exception and fail fast
• return a fallback value
• return a negative value and let the record be skipped

Fail-fast behavior is often the safest default because silently skipping bad timestamps can hide data quality problems. If you do estimate a value, use that choice deliberately and document the rule clearly.

Performance Considerations

The extractor runs for every source record, so keep it lightweight. Parsing full JSON for each message may be acceptable at moderate throughput, but if performance matters, use an efficient serializer or store the timestamp in a strongly typed value object that avoids repeated generic parsing.

The extractor should also be stateless. Kafka's contract assumes it can be reused safely without holding record-specific mutable state.

Common Pitfalls

One common mistake is returning seconds instead of milliseconds. Kafka Streams expects epoch milliseconds, and getting that wrong shifts windows dramatically.

Another issue is using a default extractor accidentally because the custom class was not registered in the config or was registered under the wrong property name.

Developers also overlook what happens to downstream topics. Record timestamps can propagate into changelogs and internal topics, so nonsensical extracted values can affect retention and stream-time behavior beyond one operator.

Finally, do not bury parsing failures. If the payload schema changes and the extractor quietly invents timestamps, the topology may keep running while producing misleading results.

Summary

• Use a custom TimestampExtractor when the real event time is embedded in the message payload.
• Return epoch milliseconds, not seconds or formatted strings.
• Register the extractor with the Kafka Streams timestamp extractor config.
• Choose an explicit policy for malformed or missing timestamps.
• Keep the extractor stateless and efficient because it runs for every source record.