Hey there! Let's unpack why your test topic's messages are sticking around longer than expected—this is a super common mix-up with Kafka's retention configuration, so you're not alone.

First, Fix the Configuration Mix-Up #

The key issue here is a misused setting: you set delete.retention.ms=60000, but this config only applies to tombstone messages (used with log compaction, cleanup.policy=compact). It has no effect on regular message retention when you're using cleanup.policy=delete.

For regular message expiration, the critical config is retention.ms—this defines how long Kafka keeps your messages before marking them for deletion. Since you didn't set this in your topic creation command, Kafka fell back to the cluster's default value (usually 7 days, or 604800000 ms), which is why your messages stayed visible for hours.

How Kafka's Delete Mechanism Actually Works #

Kafka doesn't delete individual messages—it operates on segment files. Here's the step-by-step flow with your intended settings:

1. Segment Rolling: Your segment.ms=60000 means Kafka will create a new segment file for the topic after 1 minute. Once a segment is rolled (no longer accepting new messages), it becomes eligible for retention checks.
2. Retention Check: A background log cleaner thread runs periodically (default every 10 seconds via log.cleaner.backoff.ms) to check if segments are older than retention.ms. If a segment's last modified time exceeds this value, it's marked for deletion.
3. Deletion Execution: The cleaner will delete marked segments, but this isn't instantaneous—there's a small delay between a segment being eligible and actually being removed.

How to Fix Your Test #

Let's adjust your setup to get the behavior you expect:

1. Update the Existing Topic Config:

   ./kafka-configs.sh --bootstrap-server broker:29092 --alter --topic test --add-config "retention.ms=60000"
   

2. Or Recreate the Topic with Correct Settings:

   ./kafka-topics --create --zookeeper localhost:2181 --topic test --replication-factor 2 --partitions 1 --config "retention.ms=60000" --config "segment.ms=60000"
   

   Note: cleanup.policy=delete is Kafka's default, so you can omit it if you want.

How to Verify It's Working #

To check when segments are marked for deletion, use the kafka-log-dirs tool to inspect your topic's log segments:

./kafka-log-dirs.sh --bootstrap-server broker:29092 --describe --topic-list test

Look for the retentionMs value and the lastModifiedTime of each segment. Once a segment's age exceeds retentionMs, it'll be deleted in the next cleaner run.

Quick Recap of Key Configs #

• retention.ms: Max time to keep regular messages (critical for delete policy)
• segment.ms: Time after which Kafka rolls to a new segment file
• cleanup.policy=delete: Tells Kafka to delete expired segments (default)
• delete.retention.ms: Only for tombstone messages in log compaction (ignore this for delete policy)

内容的提问来源于stack exchange，提问作者flamengo