storage

Active contributors: Mickael Maison, Kamal Chandraprakash, Ken Huang, PoAn Yang, TaiJuWu

Purpose

The storage/ module owns the on-disk log: how partitions are written to and read from local files, how segments are rolled and indexed, how producer state is recovered after a crash, and how the tiered-storage path moves cold segments to a remote store. It is split into a public storage-api/ (interfaces an external RemoteStorageManager plugin implements) and storage/ (the broker's internal implementation).

Directory layout

storage/
├── api/                                                ← storage-api submodule (public SPI)
│   └── src/main/java/org/apache/kafka/server/log/remote/storage/
│       ├── RemoteStorageManager.java                  ← plugin interface (KIP-405)
│       ├── RemoteLogMetadataManager.java
│       └── RemoteLogSegmentMetadata.java
└── src/
    └── main/java/org/apache/kafka/storage/internals/
        ├── log/
        │   ├── UnifiedLog.java                        ← per-partition replicated log
        │   ├── LogSegment.java                        ← single segment file pair
        │   ├── LogSegments.java                       ← per-log skiplist of segments
        │   ├── LogConfig.java                         ← per-topic config (segment.bytes, retention, ...)
        │   ├── LocalLog.java                          ← non-replicated parts of UnifiedLog
        │   ├── ProducerStateManager.java              ← idempotency / EOS state per partition
        │   ├── LogCleaner.java                        ← compaction
        │   ├── RemoteIndexCache.java                  ← cached remote-segment indexes
        │   └── ...
        ├── checkpoint/                                 ← recovery-point and log-start-offset checkpoints
        ├── epoch/                                      ← LeaderEpochCheckpoint, LeaderEpochFileCache
        └── utils/

The Scala counterpart in core/src/main/scala/kafka/log/ is now thin — LogManager, the cleaner registry, and a few adapter types — most of the logic lives in storage/'s Java code.

On-disk layout for one partition

<log.dirs>/<topic>-<partition>/
├── 00000000000000000000.log          ← segment data file
├── 00000000000000000000.index        ← offset → file position
├── 00000000000000000000.timeindex    ← timestamp → offset
├── 00000000000000000000.snapshot     ← producer state at segment start (txn / idempotent)
├── 00000000000000000000.txnindex     ← aborted transactions, for read-committed
├── ... (more segments rolled by size/time)
├── leader-epoch-checkpoint           ← (epoch, startOffset) records
├── partition.metadata                ← topic id, version
└── metadata.properties               ← cluster id, node id, version (per log.dir)

A separate special partition __cluster_metadata-0 is the KRaft metadata log; a separate special partition prefix __remote_log_metadata carries RemoteLogMetadataManager records when tiered storage is enabled.

Key abstractions

Type	File	Purpose
UnifiedLog	storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java	The replicated log abstraction; combines local + remote segments.
LocalLog	storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java	Local-only operations (append, read from disk).
LogSegment	storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java	One .log + sibling .index + .timeindex file group.
OffsetIndex, TimeIndex, TransactionIndex	storage/.../log/	Sparse-index lookups into the data file.
ProducerStateManager	storage/.../log/ProducerStateManager.java	Tracks producer ID / epoch / sequence per partition; recovers from .snapshot.
LogCleaner	storage/.../log/LogCleaner.java	Per-key compaction for cleanup.policy=compact topics.
LeaderEpochFileCache	storage/.../epoch/LeaderEpochFileCache.java	(epoch, startOffset) pairs used during follower truncation / leader election.
LogManager (Scala)	core/src/main/scala/kafka/log/LogManager.scala	Owns all UnifiedLogs; handles startup recovery, retention, compaction scheduling.
RemoteLogManager	storage/.../log/remote/RemoteLogManager.java (in storage/src/main/java/org/apache/kafka/storage/internals/log/remote/)	Coordinates copying segments to RemoteStorageManager and serving remote reads.
RemoteStorageManager	storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java	Plugin SPI implemented by external storage backends (S3, GCS, HDFS, etc.).

Append path

1. ReplicaManager.appendRecords validates the request and forwards records to the relevant Partition on the leader.
2. Partition.appendRecordsToLeader calls UnifiedLog.appendAsLeader.
3. UnifiedLog validates each RecordBatch (CRC, magic version, idempotency / EOS sequence number via ProducerStateManager).
4. The active segment writes the bytes; OffsetIndex and TimeIndex get sparse entries every index.interval.bytes.
5. If the segment exceeds segment.bytes or is older than segment.ms, the log rolls a new segment and writes a .snapshot of producer state at the boundary.
6. The append returns a LogAppendInfo (first/last offsets, timestamps, error). The HW is advanced separately when followers ack via the fetcher loop.

Read path

1. ReplicaManager.fetchMessages consults Partition to determine if this broker leads the partition (or is a designated follower in a KIP-392 read-from-follower scenario).
2. The read offset is bounded by the high-watermark (consumers) or the LEO (followers).
3. UnifiedLog.read looks up the segment via LogSegments (which is keyed on segment baseOffset), then LogSegment.read uses OffsetIndex to seek into the .log file and zero-copies bytes via FileChannel.transferTo.
4. If the requested offset is below the local logStartOffset and tiered storage is enabled, the read is satisfied via RemoteLogManager instead.

Producer state and recovery

ProducerStateManager is the broker-side counterpart of the client-side TransactionManager. It maintains, per partition, the most recent (producerId, producerEpoch, sequence) for each PID. Snapshots are taken on segment roll and on shutdown; on broker restart the manager re-replays only the latest snapshot plus subsequent segments, not the whole log. Aborted transactions are tracked separately in the .txnindex so that read_committed consumers can skip them efficiently.

Tiered storage (KIP-405)

When remote.log.storage.system.enable=true for a topic, RemoteLogManager schedules per-partition tasks that:

1. Copy segments older than local.retention.ms to the configured RemoteStorageManager plugin.
2. Record metadata in the RemoteLogMetadataManager (default implementation persists metadata to an internal Kafka topic, __remote_log_metadata).
3. Delete the local copy after the remote upload is acknowledged.
4. On a fetch below the local start offset, transparently fetch from the remote backend.

The plugin SPI is intentionally narrow — copyLogSegmentData, fetchLogSegment, deleteLogSegmentData — so that S3, GCS, HDFS, NFS, and other backends can be wired in without broker changes. See features/tiered-storage.md for the user-facing view.

Log cleaner (compaction)

For topics with cleanup.policy=compact, LogCleaner runs background threads that:

1. Build an in-memory map of key → latest offset for the dirty section of each log.
2. Rewrite older segments, dropping records whose offset is not the latest for their key.
3. Update indexes and atomically swap the rewritten segments in.

Compaction respects tombstones (null value records) and the delete.retention.ms setting.

Configuration

• Per-broker: log.dirs, num.recovery.threads.per.data.dir, log.cleaner.threads, log.cleaner.dedupe.buffer.size, ...
• Per-topic (declared in LogConfig): segment.bytes, segment.ms, retention.ms, retention.bytes, cleanup.policy, min.insync.replicas, compression.type, min.cleanable.dirty.ratio, ... — see reference/configuration.md.
• Tiered storage: remote.log.storage.system.enable, remote.log.metadata.manager.class.name, remote.log.storage.manager.class.name, plus plugin-specific properties.

Entry points for modification

• Change segment / index format: LogSegment.java, OffsetIndex.java, TimeIndex.java. On-disk format changes need a MetadataVersion bump and migration code.
• Change retention / compaction behavior: LogManager.scala schedules retention; the actual deletion logic is in UnifiedLog. Compaction lives in LogCleaner.java.
• Add a remote-storage backend: implement RemoteStorageManager (and optionally RemoteLogMetadataManager) in your own JAR, drop it on the classpath, and configure remote.log.storage.manager.class.name.
• Change the producer state / EOS protocol: ProducerStateManager.java plus clients/.../producer/internals/TransactionManager.java. KIP required.

Related pages

• Modules: core (broker) — ReplicaManager, LogManager callers.
• Modules: clients — RecordBatch / MemoryRecords types written here.
• Features: Tiered storage.
• Features: Replication.
• Features: Exactly-once semantics.