Understanding Message Timestamps
1. Understanding Timestamp Types
| Type |
Source |
Detail |
| CreateTime |
Set by producer |
Event time |
| LogAppendTime |
Set by broker on write |
Ingestion time |
| message.timestamp.type |
Topic config selects |
CreateTime default |
Example: Read record timestamp
long ts = record.timestamp();
TimestampType type = record.timestampType(); // CREATE_TIME or LOG_APPEND_TIME
2. Using Create Time
| Aspect |
Description |
Detail |
| CreateTime |
When event occurred |
Producer-supplied |
| Default |
Current time if unset |
System.currentTimeMillis |
| Streams |
Drives event-time windows |
Preferred |
Example: Explicit create time
// topic, partition, timestamp, key, value
producer.send(new ProducerRecord<>("events", null, eventTimeMs, key, value));
3. Using Log Append Time
| Aspect |
Description |
Detail |
| LogAppendTime |
Broker write moment |
Overwrites producer ts |
| Monotonic |
Roughly increasing per partition |
Server clock |
| Use Case |
Retention by arrival |
Deterministic |
Example: Set log append time
kafka-configs.sh --alter --entity-type topics --entity-name events \
--add-config message.timestamp.type=LogAppendTime \
--bootstrap-server localhost:9092
4. Configuring Timestamp Type
| Config |
Values |
Default |
| message.timestamp.type |
CreateTime / LogAppendTime |
CreateTime |
| Scope |
Broker default + topic override |
Per topic |
log.message.
timestamp.type |
Broker-level default |
CreateTime |
Example: Broker default
log.message.timestamp.type=CreateTime
5. Setting Producer Timestamp
| Aspect |
Description |
Detail |
| Timestamp arg |
Explicit in ProducerRecord |
ms epoch |
| null |
Defaults to send time |
Auto |
| Ignored if |
Topic uses LogAppendTime |
Broker overwrites |
Example: Backdated event
long original = parseEventTime(json);
producer.send(new ProducerRecord<>("events", null, original, key, json));
6. Understanding Timestamp Ordering
| Aspect |
Description |
Detail |
| Not Guaranteed |
CreateTime may be unordered |
Out-of-order events |
| Offset Order |
Offsets always increase |
Storage order |
| Streams |
Handles via grace period |
Late events |
Example: Detect out-of-order
if (record.timestamp() < lastSeenTs) outOfOrderCount++;
lastSeenTs = Math.max(lastSeenTs, record.timestamp());
7. Using Timestamps for Retention
| Aspect |
Description |
Detail |
| Retention Basis |
Segment max timestamp vs retention.ms |
Per segment |
| CreateTime risk |
Backdated data deleted early |
Watch clocks |
| LogAppendTime |
Predictable expiry |
Arrival-based |
Warning: With CreateTime, future-dated timestamps can keep segments alive far beyond retention.ms.
Example: Retention with append time
message.timestamp.type=LogAppendTime
retention.ms=259200000
8. Filtering by Timestamp
| Method |
Purpose |
Detail |
| offsetsForTimes |
Find offset at/after time |
Per partition |
| seek |
Jump to that offset |
Then poll |
| In-loop |
Skip records below time |
Post-filter |
Example: Seek by timestamp
Map<TopicPartition, Long> query = Map.of(tp, sinceMs);
var offsets = consumer.offsetsForTimes(query);
consumer.seek(tp, offsets.get(tp).offset());
9. Configuring Timestamp Difference
| Config |
Description |
Default |
message.timestamp.
difference.max.ms |
Max allowed skew vs broker time |
Long.MAX |
| Reject |
Records beyond skew refused |
CreateTime only |
| Use Case |
Block bad clocks |
Data hygiene |
Example: Limit timestamp skew
message.timestamp.difference.max.ms=3600000 # 1 hour
10. Resetting Offsets by Timestamp
| Method |
Description |
Detail |
| --to-datetime |
CLI reset to time |
Group offline |
| offsetsForTimes |
API equivalent |
In-app |
| Use Case |
Replay from incident |
Recovery |
Example: Reset group to datetime
kafka-consumer-groups.sh --reset-offsets --group analytics \
--topic events --to-datetime 2026-06-10T08:00:00.000 \
--execute --bootstrap-server localhost:9092