4.1 – Retention Policies

What is Retention?

Retention is Kafka's mechanism for automatically deleting old data from topics to reclaim disk space.

Key Concept: Unlike traditional message queues (which delete messages after consumption), Kafka retains messages for a configured period or until a size limit is reached, regardless of whether they've been consumed.

Why This Matters:

• Replay: Consumers can re-read old messages (time travel)
• New Consumers: New applications can read historical data
• Debugging: Inspect old messages for troubleshooting

Retention vs Deletion

Kafka's Retention Model

Traditional Message Queue:

Producer → Queue → Consumer
                  ↓
            Message deleted immediately after consumption

Kafka:

Producer → Topic (log) → Consumer
                  ↓
            Message retained for N days/bytes
                  ↓
            Deleted based on retention policy (not consumption)

Key Point: Consumption does not trigger deletion. Messages are deleted based on time or size, not whether they've been consumed.

Time-Based Retention

How It Works

Config: retention.ms (milliseconds)

Behavior: Messages older than retention.ms are eligible for deletion.

Example:

retention.ms=604800000  # 7 days (default)

Interpretation: Messages written more than 7 days ago are deleted.

Time-Based Deletion Flow

Step 1: Kafka checks each segment's last modified timestamp (when the segment was closed)

Step 2: If current_time - segment_close_time > retention.ms, the segment is eligible for deletion

Step 3: Kafka's log cleaner thread deletes the segment

Example:

Current Time: Day 10

orders-0/
  00000000000000000000.log (closed on Day 1)  ← 9 days old, deleted
  00000000000000123456.log (closed on Day 5)  ← 5 days old, retained
  00000000000000456789.log (active, Day 10)   ← Active, never deleted

Key Point: Retention is segment-based, not message-based. Kafka deletes entire segments, not individual messages.

Message Timestamp Types: CreateTime vs LogAppendTime

Important: Kafka uses message timestamps to determine when retention time starts. The timestamp type affects how retention is calculated.

Config: message.timestamp.type (broker or topic level)

CreateTime (Default)

Behavior: Kafka uses the timestamp set by the producer when the message was created.

Config:

message.timestamp.type=CreateTime  # Default

Example:

Producer creates message at: 2023-01-01 10:00:00 (CreateTime)
Message arrives at broker at: 2023-01-01 10:00:05 (5 seconds later)

Retention check uses: 2023-01-01 10:00:00 (CreateTime)

Benefit: Reflects event time (when the event actually happened).

Risk: If a producer sends old event-time data (e.g., backfill from 7 days ago), it may be deleted immediately if retention is 7 days.

Example Problem:

Retention: 7 days
Producer sends message with CreateTime: 2023-01-01 (8 days ago)
Current time: 2023-01-09

Result: Message is immediately eligible for deletion (8 days > 7 days)

LogAppendTime

Behavior: Kafka overwrites the producer's timestamp with the broker's timestamp when the message is appended to the log.

Config:

message.timestamp.type=LogAppendTime

Example:

Producer creates message at: 2023-01-01 10:00:00 (CreateTime, ignored)
Message arrives at broker at: 2023-01-09 10:00:05
Broker sets timestamp to: 2023-01-09 10:00:05 (LogAppendTime)

Retention check uses: 2023-01-09 10:00:05 (LogAppendTime)

Benefit: Retention is based on when Kafka stored the message, not when it was created.

Use Cases:

• Backfills: Replaying old data from a database (CreateTime would be old, causing immediate deletion)
• Late-arriving events: Events that arrive hours/days after they occurred
• Untrusted producers: Producers with incorrect clocks or malicious timestamps
• Consistent retention: All messages retained for the same duration from when Kafka received them

Example Solution:

Retention: 7 days
Producer sends message with CreateTime: 2023-01-01 (8 days ago)
Broker sets LogAppendTime: 2023-01-09 (now)

Result: Message is retained for 7 days from 2023-01-09 (not deleted immediately)

Choosing Between CreateTime and LogAppendTime

Configuration Examples

Broker-Level Default (server.properties):

log.message.timestamp.type=CreateTime  # Default for all topics

Topic-Level Override:

# Set LogAppendTime for backfill topic
kafka-topics --create --topic historical-data \
  --partitions 10 \
  --replication-factor 3 \
  --config message.timestamp.type=LogAppendTime \
  --bootstrap-server localhost:9092

Verifying Timestamp Type

Check topic config:

kafka-configs --describe --entity-type topics --entity-name orders \
  --bootstrap-server localhost:9092 | grep message.timestamp.type

Output:

message.timestamp.type=CreateTime

Retention Granularity

Problem: If segments are large (e.g., 10 GB), retention is coarse-grained.

Example (with log.segment.bytes=10GB, retention.ms=7 days):

Day 0: Segment 0 starts
Day 3: Segment 0 closes (10 GB full), Segment 1 starts
Day 6: Segment 1 closes (10 GB full), Segment 2 starts
Day 7: Retention check
  - Segment 0 is 7 days old → Deleted
  - Segment 1 is 4 days old → Retained (even though it contains messages from Day 3-6)

Result: Some messages in Segment 1 are older than 7 days but are retained (because the entire segment isn't old enough).

Solution: Use smaller segments (log.segment.bytes=1GB) for finer-grained retention.

Time Calculation: Message Timestamp vs Segment Close Time

Kafka uses segment close time, not individual message timestamps, for retention checks.

Why?

• Performance: Checking every message's timestamp would be too slow
• Simplicity: Segment-based deletion is fast (just delete files)

Implication: If a segment contains messages with timestamps from Day 1-3, but the segment was closed on Day 3, the entire segment is considered "Day 3" for retention purposes.

Retention Configuration Hierarchy

Retention can be configured at three levels:

Example:

# Broker default (server.properties)
log.retention.ms=604800000  # 7 days

# Topic override (more aggressive retention for logs)
kafka-configs --alter --entity-type topics --entity-name app-logs \
  --add-config retention.ms=86400000  # 1 day

Precedence: Topic-level config > Broker-level default

Size-Based Retention

How It Works

Config: retention.bytes (bytes per partition)

Behavior: If a partition's total size exceeds retention.bytes, oldest segments are deleted until size is under the limit.

Example:

retention.bytes=10737418240  # 10 GB per partition

Interpretation: Each partition can grow up to 10 GB. Once it exceeds 10 GB, oldest segments are deleted.

Size-Based Deletion Flow

Step 1: Kafka calculates total size of all segments in a partition

Step 2: If total_size > retention.bytes, delete oldest segments (starting from lowest offset)

Step 3: Repeat until total_size ≤ retention.bytes

Example:

orders-0/ (total size: 12 GB, limit: 10 GB)
  00000000000000000000.log (2 GB)  ← Oldest, deleted
  00000000000000123456.log (3 GB)  ← Next oldest, deleted
  00000000000000456789.log (7 GB)  ← Retained (now total = 7 GB < 10 GB)

Size-Based Retention: Per-Partition Limit

Important: retention.bytes is per partition, not per topic.

Example:

Topic: orders
Partitions: 10
retention.bytes: 10 GB per partition

Total topic size: 10 partitions × 10 GB = 100 GB

Key Point: If you want to limit total topic size, you must calculate: retention.bytes = total_topic_limit / num_partitions.

Combined Time and Size Retention

Kafka supports both time and size retention simultaneously.

Config:

retention.ms=604800000      # 7 days
retention.bytes=10737418240  # 10 GB per partition

Behavior: A segment is deleted if either condition is met:

• Segment is older than 7 days OR
• Partition size exceeds 10 GB (and segment is the oldest)

Example:

Partition size: 12 GB (exceeds 10 GB)
Oldest segment: 5 days old (not expired by time)

Result: Oldest segment is deleted (size limit triggered)

Key Point: Time and size are OR conditions. Whichever triggers first causes deletion.

Infinite Retention (Special Case)

Disabling Retention

Config:

retention.ms=-1  # Infinite retention (never delete by time)
retention.bytes=-1  # Infinite retention (never delete by size)

Use Case: When you want to keep all data forever (e.g., event sourcing, audit logs).

Trade-off: Disk usage grows indefinitely. You must provision enough storage.

Alternative: Use log compaction (covered in Module 4.2) to retain only the latest value per key.

Retention and Replication

Retention applies to each replica independently.

Example (RF=3):

Partition 0, Replica 1 (Broker 1):
  Retention policy runs → Deletes old segments

Partition 0, Replica 2 (Broker 2):
  Retention policy runs → Deletes same old segments

Partition 0, Replica 3 (Broker 3):
  Retention policy runs → Deletes same old segments

Key Point: All replicas have the same retention policy (configured at the topic level). Retention runs independently on each broker.

Retention Check Frequency

Kafka doesn't check retention continuously. It runs at intervals.

Config: log.retention.check.interval.ms (default: 5 minutes)

Behavior: Every 5 minutes, Kafka scans all partitions and deletes segments that violate retention policy.

Implication: A segment might be "expired" for up to 5 minutes before it's actually deleted.

Example:

Time 12:00 PM: Segment becomes eligible for deletion (8 days old)
Time 12:05 PM: Kafka runs retention check, deletes segment

Key Point: Retention is not instant. There's a delay of up to log.retention.check.interval.ms.

Retention and Consumer Lag

Important: Retention Does NOT Wait for Consumers

Scenario: A consumer is lagging (hasn't consumed messages from 10 days ago). Retention policy is 7 days.

What Happens:

Day 0: Messages written
Day 10: Consumer tries to read messages from Day 0
Result: Messages are already deleted (retention policy expired on Day 7)

Consumer Error:

OffsetOutOfRangeException: Fetch position 12345 is out of range. 
The earliest available offset is 456789.

Key Point: Kafka deletes messages based on retention policy, not consumer lag. If a consumer falls too far behind, it will miss messages.

Handling OffsetOutOfRangeException

Options when consumer lag exceeds retention:

1. Seek to Earliest Offset (skip missing messages):

   consumer.seekToBeginning(partitions);
   

2. Increase Retention (prevent future data loss):

   kafka-configs --alter --entity-type topics --entity-name orders \
     --add-config retention.ms=1209600000  # 14 days
   

3. Monitor Consumer Lag (alert before data loss):

   • Use metrics: records-lag-max (per partition)
   • Alert if lag > 50% of retention period

Retention and Disk Space Planning

Estimating Disk Usage

Formula:

Disk Usage (per partition) = throughput (MB/sec) × retention (seconds)

Example:

Topic: orders
Partitions: 10
Throughput: 100 MB/sec (per topic)
Retention: 7 days = 604,800 seconds

Disk per partition = (100 MB/sec / 10 partitions) × 604,800 sec = 604,800 MB = 604 GB
Total disk (all partitions) = 604 GB × 10 = 6.04 TB

With Replication (RF=3):

Total disk (all replicas) = 6.04 TB × 3 = 18.12 TB

Key Point: Disk requirements scale with throughput × retention × replication factor.

Compression (Reduces Disk Usage)

Kafka supports compression: GZIP, Snappy, LZ4, Zstd

Impact on Disk:

Uncompressed: 100 MB/sec × 7 days = 60 TB
Compressed (2x ratio): 50 MB/sec × 7 days = 30 TB

Key Point: Compression can halve disk usage (covered in producer and broker configs).

Retention Best Practices

1. Match Retention to Business Needs

2. Monitor Disk Usage

Key Metrics:

• kafka.log:type=Log,name=Size,topic=orders,partition=0
• Alert if disk usage > 80%

Runbook: If disk is full, either:

• Reduce retention (delete old data faster)
• Add more brokers (distribute partitions)
• Add more disk capacity

3. Use Smaller Segments for Fine-Grained Retention

Problem: Large segments (10 GB) → coarse-grained retention

Solution: Use 1 GB segments (default):

log.segment.bytes=1073741824  # 1 GB

4. Retention ≥ Max Consumer Lag

Rule: Set retention to at least 2x your maximum acceptable consumer lag.

Example:

• Max acceptable lag: 2 hours
• Recommended retention: 4+ hours (or 1 day for safety)

Why: Gives consumers buffer time to catch up before data is deleted.

5. Tiered Storage (Kafka 2.7+)

Concept: Move old segments to cheaper storage (S3, HDFS) instead of deleting.

Config (Confluent Platform or Apache Kafka 2.7+):

log.retention.ms=604800000  # 7 days on local disk
log.remote.storage.enable=true
log.remote.retention.ms=2592000000  # 30 days on S3

Benefit: Keep more data without expensive local SSD storage.

Key Takeaways

1. Retention = Automatic Deletion: Kafka automatically deletes old segments based on time or size, not consumption.

2. Time-Based Retention (retention.ms): Deletes segments older than N milliseconds (default: 7 days).

3. Size-Based Retention (retention.bytes): Deletes oldest segments when partition size exceeds limit (per partition).

4. Combined Retention: Both time and size can be configured. Whichever triggers first causes deletion.

5. Segment-Based Deletion: Kafka deletes entire segments (not individual messages). Segment size affects retention granularity.

6. Retention Runs Periodically: Kafka checks retention every 5 minutes (configurable via log.retention.check.interval.ms).

7. Consumer Lag vs Retention: Kafka deletes messages based on retention policy, not consumer lag. If a consumer lags too far, it will miss messages.

8. Disk Planning: Disk usage = throughput × retention × replication factor. Always provision enough disk capacity.

What's Next?

Next Module: 4.2 – Log Compaction

What We Covered:

• ✅ Time-based retention (retention.ms)
• ✅ Size-based retention (retention.bytes)
• ✅ Combined retention (time + size)
• ✅ Segment-based deletion
• ✅ Retention and consumer lag
• ✅ Disk space planning

What We Didn't Cover (Explicitly Deferred):

• ❌ Log compaction (key-based deduplication) → Module 4.2
• ❌ Retention vs compaction trade-offs → Module 4.3
• ❌ Performance impact of retention → M6: Operations

Retention Policies: Complete ✅