Understanding Message Timestamps
1. Understanding Timestamp Types
| Type | Description | Usage | Notes |
|---|---|---|---|
| CreateTime | Timestamp assigned by the Kafka broker when the message is received | Default timestamp type | Used for ordering messages within a partition |
| LogAppendTime | Timestamp assigned by the Kafka broker when the message is appended to the log | Configurable via message.timestamp.type |
Ensures consistent ordering when messages are produced in batches |
| Custom | Timestamp provided by the producer when sending the message | Set via ProducerRecord.timestamp |
Allows for precise control over message timing, useful for event sourcing |
Example: Setting Custom Timestamp in Producer
ProducerRecord<String, String> record = new ProducerRecord<>(
"my-topic",
"key1",
"value1"
);
record.headers().add("timestamp", String.valueOf(System.currentTimeMillis()).getBytes());
Note: The
CreateTime is the default timestamp type and is automatically set by Kafka when a message is received. LogAppendTime is useful for scenarios where messages are batched, ensuring that the timestamp reflects when the message was actually written to disk.2. Using Create Time
| Property | Description | Values |
|---|---|---|
| create.time | Timestamp when the message was created by the producer | timestamp |
Example: Setting Create Time in Producer
const kafka = new Kafka({ brokers: ['localhost:9092'] });
const producer = kafka.producer();
await producer.connect();
await producer.send({
topic: 'my-topic',
messages: [
{
value: JSON.stringify({ message: 'Hello, Kafka!' }),
timestamp: Date.now()
}
]
});
Note: Create time is set by the producer and can be overridden if the broker configuration allows it.
3. Using Log Append Time
| Property | Default Value | Description |
|---|---|---|
| log.append.time | false |
When enabled, Kafka uses the timestamp of when a message is appended to the log as its message timestamp. |
Note: When
log.append.time is enabled, the timestamp of each message will reflect the time it was appended to the log rather than when it was originally produced.
4. Configuring Timestamp Type
| Property | Values | Description |
|---|---|---|
| log.message.timestamp.type | CreateTime | LogAppendTime |
Sets the timestamp type for messages. CreateTime uses the timestamp provided by the producer, while LogAppendTime uses the broker's receive time. |
Example: Setting Timestamp Type in Producer Configuration
{
"bootstrap.servers": "localhost:9092",
"message.timestamp.type": "CreateTime"
}
Note: The
CreateTime timestamp is set by the producer, whereas LogAppendTime is set by the broker when the message is appended to the log.
5. Setting Producer Timestamp
| Property | Values | Description |
|---|---|---|
| timestamp.type | CreateTime | LogAppendTime |
Determines how the timestamp is set. CreateTime uses the time when the message is created by the producer, while LogAppendTime uses the time when the message is appended to the log by the broker. |
| timestamp | Long |
Explicit timestamp value in milliseconds since epoch. Only used when timestamp.type is set to CreateTime. |
Example: Setting Producer Timestamp
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("timestamp.type", "CreateTime");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
long timestamp = System.currentTimeMillis();
ProducerRecord<String, String> record = new ProducerRecord<>(
"my-topic",
"key1",
"value1",
timestamp
);
producer.send(record);
producer.close();
Note: When
timestamp.type is set to LogAppendTime, the timestamp is automatically set by the broker and cannot be overridden.
6. Understanding Timestamp Ordering
| Timestamp Type | Description | Usage | Notes |
|---|---|---|---|
| CreateTime | Timestamp when the message was created by the producer | Default timestamp type in Kafka | Used for ordering messages by creation time |
| LogAppendTime | Timestamp when the message was appended to the log by the broker | Set via message.timestamp.type |
Useful for ensuring messages are ordered by broker arrival time |
Example: Setting Timestamp Type in Producer Configuration
{
"message.timestamp.type": "LogAppendTime",
"bootstrap.servers": "localhost:9092"
}Note: Timestamps are not guaranteed to be monotonically increasing across partitions, so ordering by timestamp alone may not reflect the actual message sequence.
7. Using Timestamps for Retention
| Property | Default Value | Description |
|---|---|---|
| log.retention.hours | 168 |
Retention period in hours for logs (default 7 days) |
| log.retention.ms | -1 |
Retention period in milliseconds; overrides log.retention.hours |
| log.retention.bytes | -1 |
Maximum size of the log before deletion (in bytes); -1 means unlimited |
Note: Kafka uses the
CreateTime timestamp by default for retention logic. If you're using LogAppendTime, retention is based on the time when messages are appended to the log.
8. Filtering by Timestamp
| Filter Type | Description | Usage | Notes |
|---|---|---|---|
| Timestamp Range | Filter messages within a specific time range | KafkaConsumer with timestamp parameter |
Supports both minTimestamp and maxTimestamp |
| Exact Timestamp Match | Filter messages matching an exact timestamp | KafkaConsumer with timestamp set to specific value |
Useful for replaying or debugging specific events |
Example: Filtering by Timestamp Range
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));
// Define timestamp range
long startTime = System.currentTimeMillis() - (24 * 60 * 60 * 1000); // 24 hours ago
long endTime = System.currentTimeMillis();
// Create filter for messages within the timestamp range
Filter<String, String> timestampFilter = new Filter<>() {
@Override
public boolean shouldConsume(String key, String value, long timestamp) {
return timestamp >= startTime && timestamp <= endTime;
}
};
// Apply filter to consumer
consumer.poll(Duration.ofSeconds(1)).forEach(record -> {
if (timestampFilter.shouldConsume(record.key(), record.value(), record.timestamp())) {
System.out.println("Consumed: " + record.value());
}
});
Note: Timestamp filtering is supported in Kafka 2.0+ and requires the
log.message.timestamp.type to be set to CreateTime or LogAppendTime.
9. Configuring Timestamp Difference
| Property | Default Value | Description |
|---|---|---|
| log.message.timestamp.diff.max.ms | 900000 |
Maximum difference between the timestamp of a message and the broker's current time. Messages with timestamps older than this will be rejected. |
| log.message.timestamp.type | CreateTime |
Timestamp type to use for messages. Can be CreateTime or LogAppendTime. |
Example: Setting Timestamp Difference Limit
# Set the maximum timestamp difference to 5 minutes
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config log.message.timestamp.diff.max.ms=300000
Note: Setting
log.message.timestamp.diff.max.ms too low may cause issues if producers send messages with timestamps that are slightly in the past due to clock drift.
10. Resetting Offsets by Timestamp
Example: Resetting Offsets Using AdminClient
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
public class OffsetResetter {
public static void main(String[] args) {
AdminClient adminClient = AdminClient.create(
Collections.singletonMap("bootstrap.servers", "localhost:9092")
);
// Define topic and timestamp
String topic = "my-topic";
long timestamp = System.currentTimeMillis() - 3600000; // 1 hour ago
// Create offset reset request
Map<TopicPartition, OffsetSpec> partitionOffsets = new HashMap<>();
partitionOffsets.put(new TopicPartition(topic, 0), OffsetSpec.forTimestamp(timestamp));
// Execute reset
try {
adminClient.listOffsets(partitionOffsets).all().get();
System.out.println("Offset reset completed.");
} catch (Exception e) {
System.err.println("Error resetting offsets: " + e.getMessage());
}
}
}| Parameter | Description | Values |
|---|---|---|
| topic | Name of the Kafka topic | string |
| timestamp | Timestamp to reset offsets to | long (milliseconds since epoch) |
| partition | Partition number to reset | int |
Note: The
listOffsets API returns the earliest offset greater than or equal to the specified timestamp. If no message exists at that timestamp, it will return the next available offset.
Warning: Resetting offsets by timestamp can cause data loss if messages are consumed and acknowledged before the reset point. Ensure that consumers are paused or stopped during the reset process.