How is a partition stored on disk (segment files, offset index), and how does retention/compaction interact with segments?

# List files in a partition directory
ls -la /data/kafka/settlements-0/

# Output:
# 00000000000000000000.log        (1 GB -- first segment)
# 00000000000000000000.index
# 00000000000000000000.timeindex
# 00000000000001048576.log        (1 GB -- second segment)
# 00000000000001048576.index
# 00000000000001048576.timeindex
# 00000000000002097152.log        (400 MB -- active segment, still growing)
# 00000000000002097152.index
# 00000000000002097152.timeindex

package com.bank.kafka;

/**
 * Conceptual: how the offset index lookup works.
 */
public class OffsetIndexLookupDemo {
    public static void main(String[] args) {
        // Sparse index entries (simplified)
        // Offset -> Byte position in .log file
        // 0      -> 0
        // 100    -> 32768
        // 200    -> 65536
        // 300    -> 98304

        int targetOffset = 250;
        // Binary search: nearest entry at or below 250 is offset 200 at position 65536
        int seekPosition = 65536;
        System.out.println("Seeking to byte " + seekPosition + " for offset " + targetOffset);
        System.out.println("Then scanning forward ~50 records to find offset 250");
        // Much faster than scanning from byte 0!
    }
}

Seeking to byte 65536 for offset 250
Then scanning forward ~50 records to find offset 250

package com.bank.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.*;

public class TimeIndexSeekDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "replay-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            TopicPartition tp = new TopicPartition("settlements", 0);
            consumer.assign(List.of(tp));

            // Seek to 1 hour ago using the time index
            long oneHourAgo = System.currentTimeMillis() - 3_600_000;
            Map<TopicPartition, Long> timestamps = Map.of(tp, oneHourAgo);
            Map<TopicPartition, OffsetAndTimestamp> result = consumer.offsetsForTimes(timestamps);

            if (result.get(tp) != null) {
                consumer.seek(tp, result.get(tp).offset());
                System.out.println("Seeked to offset " + result.get(tp).offset()
                    + " (1 hour ago)");
            }
        }
    }
}

Seeked to offset 1048200 (1 hour ago)

# Configure segment rolling for the settlements topic
kafka-configs.sh --alter --topic settlements \
  --bootstrap-server kafka-1:9092 \
  --add-config segment.bytes=268435456,segment.ms=3600000

# segment.bytes=256 MB: roll when segment reaches 256 MB
# segment.ms=1 hour: roll after 1 hour even if not full

# Delete retention: remove segments older than 3 days
kafka-configs.sh --alter --topic settlements \
  --bootstrap-server kafka-1:9092 \
  --add-config cleanup.policy=delete,retention.ms=259200000

# Result: segment with newest timestamp older than 3 days -> deleted
# Segment with newest timestamp = 2 days ago -> kept

# Compaction: keep only latest value per account
kafka-configs.sh --alter --topic account.state \
  --bootstrap-server kafka-1:9092 \
  --add-config cleanup.policy=compact,min.cleanable.dirty.ratio=0.3

# Before compaction:
#   offset 0: SAV2001={balance:100000}
#   offset 1: SAV3001={balance:200000}
#   offset 2: SAV2001={balance:90000}   <- supersedes offset 0

# After compaction:
#   offset 1: SAV3001={balance:200000}
#   offset 2: SAV2001={balance:90000}
#   (offset 0 removed, gap is normal)

# Problem: low-traffic topic retains data beyond retention.ms
# The active segment has not rolled because segment.bytes is 1 GB
# and the partition only received 10 MB in the last 30 days

# Fix: set segment.ms so the active segment rolls on a schedule
kafka-configs.sh --alter --topic low-traffic-alerts \
  --bootstrap-server kafka-1:9092 \
  --add-config segment.ms=86400000  # roll daily regardless of size

# Too small: excessive file handles
# 1 TB partition / 16 MB segments = 65,536 segments = 196,608 files
# ulimit -n may not be enough!

# Too large: coarse retention
# 1 TB partition / 1 GB segments = 1,024 segments
# Cannot delete a 1 GB segment until its last record ages out

# Good balance: 256 MB segments
# 1 TB partition / 256 MB = 4,096 segments = 12,288 files
# Fine-grained retention, manageable file count

# Rebuild indexes on startup (automatic if corrupted)
# Log message: "Found a corrupted index, rebuilding..."

# Monitor segment and index health
kafka-log-dirs.sh --describe --bootstrap-server kafka-1:9092 \
  --topic-list settlements

# Key JMX metric
# kafka.log:type=Log,name=NumLogSegments,topic=settlements,partition=0