Time-Based Retention #

One of the most fundamental differences between Apache Kafka and traditional message brokers (like RabbitMQ or ActiveMQ) lies in data deletion behavior. If traditional message brokers delete messages as soon as they’re successfully read by a consumer (destructive read), Kafka separates the reading process from storage. Kafka stores all events permanently in a commit log on disk, allowing data to be read many times by different systems (non-destructive read). However, since physical disk storage isn’t unlimited, we need a reliable data cleanup policy. The default and most commonly used cleanup policy is Time-Based Retention, which ensures data is automatically discarded after passing a certain age limit.

Dissecting the Three Retention Time Parameters #

Kafka provides three configuration parameters for setting log storage duration by time. All three can be configured globally in server.properties or overridden per topic custom.

1. log.retention.ms (Milliseconds) #

This is the highest-precision parameter and is highly recommended for production environments. If we want to apply very strict retention (for example, deleting data exactly after 12 hours), use this parameter.

2. log.retention.minutes (Minutes) #

This parameter offers medium precision. Rarely used directly except for special use cases where retention is measured in minute multiples.

3. log.retention.hours (Hours) #

This is the default parameter most often configured globally. By default, Kafka sets:

$$\text{log.retention.hours} = 168 \text{ hours (7 days)}$$

Meaning, if we don’t specify any configuration on a topic, messages are stored for 7 days before being deleted.

Precedence Rule #

If we define more than one of the parameters above (for example, accidentally setting log.retention.hours and log.retention.ms simultaneously), Kafka prioritizes the smallest time unit (highest precision). The evaluation order is:

$$\text{log.retention.ms} \longrightarrow \text{log.retention.minutes} \longrightarrow \text{log.retention.hours}$$

Setting log.retention.ms always overrides the log.retention.hours value even if the hours value is set larger.

How Does Kafka Determine Message Age? (Timestamp Dynamics) #

Time retention can’t work without a timestamp on every message record. Kafka determines the “age” of data based on the timestamp value embedded in the message metadata. This value is controlled by a topic-level configuration called log.message.timestamp.type.

There are two timestamp types supported by Kafka:

1. CreateTime (Default) #

In this mode, the timestamp is set by the producer (client) when it creates the message record (new ProducerRecord(...)). The producer system’s local clock is written into the message metadata.

The Hidden Danger of Clock Drift #

Relying on CreateTime in production has a high architectural risk if client server clocks aren’t synchronized (clock drift):

  • Future Timestamp Scenario: If a producer has a local clock incorrectly set to the year 2030 (while it’s actually 2026), the messages it sends are recorded with a 2030 timestamp. Kafka considers those messages newly born in 2030, so they’ll never be deleted by time retention for the next 4 years, clogging broker disk space.
  • Past Timestamp Scenario: Conversely, if a client clock is incorrectly set to 2010, new messages are immediately considered 16 years old as soon as they arrive at the broker. Kafka’s time retention system immediately classifies those messages as expired and deletes them instantly on the next scan cycle. Consumers won’t get a chance to read that data.

When we set log.message.timestamp.type=LogAppendTime, the leader broker ignores any timestamp sent by the client. The leader broker overwrites that timestamp with its own local server clock when it successfully writes the message to its physical commit log.

  • Advantage: Time retention policy becomes very consistent, safe, and predictable because it’s fully controlled by the broker cluster’s internal time synchronization (using the NTP — Network Time Protocol).

Dissecting the Partition Directory Files: .log, .index, and .timeindex #

To understand how Kafka evaluates time retention efficiently, we must look at the physical contents of a partition folder in the broker’s Linux filesystem. Every partition (for example my-topic-0/) contains several file types:

1. Log File (.log) #

The main binary file holding actual message records sequentially. This is where our business data is stored.

2. Offset Index File (.index) #

Maps log offsets to physical byte positions inside the .log file. Helps speed up offset-based data lookups.

3. Time Index File (.timeindex) #

This file is the key to time-based retention. It maps message timestamps to the corresponding offset number.

When Kafka evaluates whether a segment is expired, the broker doesn’t read the very large .log file. The broker simply opens the small .timeindex file and reads the last entry. This last entry records the largest (latest) timestamp of all messages in that segment. If this largest timestamp has passed the time retention limit, the entire log segment is declared expired.

Broker Server Time Synchronization Using NTP #

Because time retention reliability and KRaft coordination heavily depend on time accuracy between broker servers, we must configure and monitor time synchronization using the NTP (Network Time Protocol) service across all broker Linux operating systems.

If clock drift happens between brokers, for example Broker 1 is 10 minutes slower than Broker 2, then the expiration evaluation of partition segments moved between leaders can become inconsistent.

Using Chrony for Time Synchronization on Linux #

On modern Linux, the chrony utility is the industry standard for fast, precise NTP synchronization.

Here’s the basic chrony configuration on /etc/chrony.conf on brokers:

# Using the nearest public NTP server pool (e.g., Indonesia)
pool id.pool.ntp.org iburst

# Allowing gradual clock synchronization at startup
initstepslew 10 pool.ntp.org

# Storing local clock drift data
driftfile /var/lib/chrony/drift

To monitor the health status of time synchronization on broker operating systems, we can execute the following commands:

# Checking whether the server time is well-synchronized to external NTP servers
chronyc tracking

# Viewing the list of time source servers used and their accuracy
chronyc sources -v

How File Segmentation and the Cleaner Scheduler Work #

One of the keys to Kafka’s I/O performance speed is the commit log’s immutable and append-only nature. Therefore, Kafka never deletes messages individually from the middle of a file. Deleting messages one by one from a large file would require shuffling file lines, consuming very expensive CPU and disk I/O operations.

As a solution, Kafka divides a partition’s physical log file into small segments called Log Segments.

flowchart TD
    subgraph Partisi_Disk["Physical Partition Folder: my-topic-0/"]
        direction LR
        S1["Segment 1 (Closed) <br> Max Timestamp: 8 Days Ago <br> Status: Expired"]
        S2["Segment 2 (Closed) <br> Max Timestamp: 4 Days Ago <br> Status: Safe"]
        S3["Segment 3 (Active) <br> Receiving New Writes <br> Status: Retention-Immune"]
    end

    S1 -->|"Deleted Whole from Disk"| Trash["OS Trash"]
    style S1 stroke:#c62828,stroke-width:2px
    style S2 stroke:#2e7d32,stroke-width:2px
    style S3 stroke:#0288d1,stroke-width:2px

1. The Active Segment Concept #

The segment where the broker is currently writing new data is called the Active Segment. There’s only one active segment per partition at a time.

  • Immunity Rule: The active segment is never deleted by the time retention system, even if it contains old messages that by age calculation have passed the retention limit. This is because the broker still needs that file to write new data.

2. Segment Rolling #

An active segment is closed (rolled) and changes status to an inactive segment when one of the following conditions is met:

  • The segment file size reaches the maximum limit (default: log.segment.bytes = 1 GB).
  • The segment write duration has passed (default: log.segment.ms = 7 days).

When rolling happens, the old segment is permanently closed as a read-only file, and Kafka creates a new active segment file to hold subsequent data.

3. Log Cleaner Thread #

The broker runs an internal cleaning thread called the Log Cleaner Thread that executes periodic scans every:

$$\text{log.retention.check.interval.ms} = 300000 \text{ ms (5 minutes)}$$

When this thread runs:

  1. It scans all inactive segments for each partition.
  2. For each inactive segment, it checks the time index file (.timeindex) to find the largest timestamp in that segment.
  3. If the difference between the current time and the largest timestamp in that segment is greater than the log.retention.ms value, the entire segment file is declared expired.
  4. The broker immediately deletes the physical log segment file (.log) along with all its supporting index files (.index, .timeindex) wholesale from the OS disk using a fast file deletion system call.

The Impact of Retention Cleanup on Consumers: Handling OffsetOutOfRangeException #

As application developers, we must understand what happens when our consumer application experiences very long downtime (for example, dead for 10 days) while the topic is configured with 7-day time retention.

What Happens When a Consumer Lags #

When the consumer comes back and tries to read data from its last committed offset (for example, offset 5000), the broker realizes the segment file holding offset 5000 has already been deleted from disk by the Log Cleaner Thread because its age passed 7 days.

In this condition:

  1. The broker rejects the consumer’s read request and throws the OffsetOutOfRangeException error.
  2. The consumer’s subsequent recovery behavior is controlled by a consumer property configuration called auto.offset.reset.

Setting the auto.offset.reset Property #

  • earliest: The consumer automatically moves its read pointer to the smallest offset still available on disk (for example, offset 12,000). The consumer immediately reads new data from there. Consequence: we lose historical data from offsets 5000 to 11,999.
  • latest: The consumer moves its read directly to the end of the partition log (the newest incoming messages).
  • none: The consumer does no automatic recovery and immediately throws an error to the client application code, forcing the operations team to do manual intervention.

Time-Based Retention in the Tiered Storage Era #

Starting from Kafka 3.0+, a modern architecture feature called Tiered Storage was introduced. This feature separates fast local storage (Hot Tier) from cheap remote object storage (Cold Tier, like AWS S3, Google Cloud Storage, or Azure Blob).

With Tiered Storage, our time retention policy is divided into two stages:

flowchart TD
    Producer["Producer"] -- "Writes" --> Active["Active Segment (Hot Tier)"]
    Active --> Inactive["Inactive Segment (Rolled)"]
    Inactive -- "After log.local.retention.ms" --> Cold["Cold Tier (Object Storage S3/GCS)"]
    Cold -- "After log.retention.ms" --> Deleted["Permanently Deleted"]

Tiered Storage Configuration Parameters #

  • log.local.retention.ms: The duration data must stay on local disk (Hot Tier). Usually set very short (e.g., 24 hours) to save expensive local NVMe/SSD capacity.
  • log.retention.ms: The total data storage duration (Hot Tier + Cold Tier). For example, set to 1 year. After data passes local retention, it’s asynchronously moved to S3/GCS. That data is only truly permanently deleted after passing the 1-year limit.

Operational CLI Guide #

Here are practical command-line commands for dynamically configuring time retention at the topic level without broker restarts.

1. Changing Topic Retention to 3 Days (In Milliseconds) #

We recommend converting days to milliseconds for accuracy: 3 days = 259,200,000 milliseconds.

# Dynamically changing the topic's time retention property
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name user-activity-logs \
  --alter \
  --add-config log.retention.ms=259200000

2. Changing the Timestamp Type to LogAppendTime #

To secure the time retention policy from client clock drift threats:

# Rejecting client CreateTime and forcing server LogAppendTime
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name user-activity-logs \
  --alter \
  --add-config log.message.timestamp.type=LogAppendTime

3. Lowering the Segment Roll Interval #

If we have a low-throughput topic but need very responsive time retention (for example, data must disappear exactly after 1 hour), we must speed up segment rolling (e.g., every 1 hour / 3,600,000 ms) so segments close quickly and can be immediately deleted by the cleaner:

# Speeding up segment roll so cleanup isn't delayed
kafka-configs.sh --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name ephemeral-data \
  --alter \
  --add-config log.segment.ms=3600000 \
  --add-config log.retention.ms=3600000

Summary #

  • Time-Based Retention: Kafka’s default policy for discarding data segments after passing a certain storage age limit.
  • Parameter Hierarchy: log.retention.ms has the highest priority, followed by log.retention.minutes, then log.retention.hours.
  • LogAppendTime: Use this setting to ensure time retention reliability is free from client system clock errors (clock drift).
  • Time Index File: Kafka uses the .timeindex file to evaluate segment age instantly without scanning the large main log file.
  • NTP Server: Install Linux clock synchronization (e.g., chrony) on all brokers to prevent clock drift mismatch problems.
  • OffsetOutOfRangeException: Occurs if a consumer is offline longer than the topic retention. Set the consumer’s auto.offset.reset parameter wisely.
  • Tiered Storage: A modern architecture enabling asynchronous data movement from local disk (Hot Tier) to Cloud Object Storage (Cold Tier) based on the log.local.retention.ms parameter.
Next: Size-Based →
About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact