Tiered storage

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

Tiered storage (KIP-405) lets Kafka offload older log segments to a remote object store while keeping recent segments locally on the broker. Producers and consumers see a single contiguous log; the broker transparently fetches from the remote store when a consumer reads below the local start offset. This makes long retention (weeks, months, years) cheap without inflating the local disks of every broker.

Concept

graph LR
    Prod[Producer] -->|append| L[Leader broker]
    L --> Local[(Local log:<br/>recent segments)]
    L --> RLM[RemoteLogManager]
    RLM -->|copy older segments| RSM[(RemoteStorageManager plugin<br/>S3 / GCS / HDFS / ...)]
    RLM -->|metadata| RLMM[(__remote_log_metadata topic)]
    Cons[Consumer] -->|Fetch old offset| L
    L -->|fetch from remote| RSM
    L --> Cons

Old segments live in the remote store; their existence and offsets are tracked in __remote_log_metadata. Consumers see a single log — the broker handles the routing.

Components

Concern	Where
Public plugin SPI	storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteStorageManager.java
Public metadata SPI	storage/api/.../RemoteLogMetadataManager.java
Default metadata manager (Kafka topic-backed)	storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java
Remote log manager (broker side)	storage/src/main/java/org/apache/kafka/storage/internals/log/remote/RemoteLogManager.java
Remote-segment fetch path	storage/.../log/remote/RemoteFetchInfo.java, UnifiedLog.java (read fallback to remote)
Remote index cache	storage/.../log/RemoteIndexCache.java
Schema for metadata records	storage/src/main/resources/common/message/RemoteLogSegmentMetadataRecord.json (and friends)

Plugin SPI

RemoteStorageManager is intentionally narrow — implementations supply:

void copyLogSegmentData(RemoteLogSegmentMetadata, LogSegmentData);
InputStream fetchLogSegment(RemoteLogSegmentMetadata, int startPosition);
InputStream fetchLogSegment(RemoteLogSegmentMetadata, int startPosition, int endPosition);
InputStream fetchIndex(RemoteLogSegmentMetadata, IndexType);
void deleteLogSegmentData(RemoteLogSegmentMetadata);

RemoteLogMetadataManager is the durable index of "what segments exist remotely":

void addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata);
void updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate);
Optional<RemoteLogSegmentMetadata> remoteLogSegmentMetadata(TopicIdPartition, int leaderEpoch, long offset);
Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(TopicIdPartition);

The default TopicBasedRemoteLogMetadataManager persists this metadata to the internal __remote_log_metadata topic, so users get a working tiered-storage installation by configuring just RemoteStorageManager.

External backends (S3, GCS, HDFS, NFS, ...) ship as third-party JARs and are loaded via remote.log.storage.manager.class.name. Kafka itself does not bundle any; the project provides only the API and the topic-backed metadata manager.

Lifecycle of a segment

1. Active. Producer appends to the latest local segment.
2. Roll. Segment exceeds segment.bytes / segment.ms; a new active segment opens.
3. Eligible for tiering. When the segment's age exceeds local.retention.ms (or local size exceeds local.retention.bytes), RemoteLogManager schedules an async upload.
4. Upload. copyLogSegmentData sends the data file plus offset / time / leader-epoch indexes; on success, a RemoteLogSegmentMetadataRecord is appended to __remote_log_metadata.
5. Local delete. Once the upload is committed, the local copy is removed.
6. Read. A consumer fetch with offset below the local start triggers UnifiedLog.read to fall back through RemoteLogManager.fetchRecords, which streams from the remote store via RemoteIndexCache for the indexes.
7. Retention. When the segment exceeds retention.ms / retention.bytes (the total retention), deleteLogSegmentData is called and the metadata record is updated to DELETE_SEGMENT_FINISHED.

Read path

sequenceDiagram
    participant C as Consumer
    participant Brk as Leader broker
    participant Idx as RemoteIndexCache
    participant RSM as RemoteStorageManager
    C->>Brk: FetchRequest(offset=O)
    alt O above local start
      Brk->>Brk: read from local segment
    else O below local start
      Brk->>Idx: lookup remote segment for O
      Idx->>RSM: fetchIndex (offsetIndex, txnIndex)
      Brk->>RSM: fetchLogSegment(O, ...)
      RSM-->>Brk: bytes
      Brk->>Brk: decode → records
    end
    Brk-->>C: FetchResponse

To avoid hammering the remote store, indexes are cached locally in RemoteIndexCache (with size-bounded LRU eviction). The data itself is streamed; the broker does not pre-fetch entire remote segments.

The default behavior is to serve one partition per FetchRequest when the partition needs a remote read — this prevents one slow remote backend from blocking many partitions on the same fetch.

Metadata topic

__remote_log_metadata is an internal Kafka topic that holds the segment metadata records. Like other coordinator topics, it is partitioned (by topicIdPartition) and replicated. The default TopicBasedRemoteLogMetadataManager produces records here and reads them back to populate its in-memory index on broker startup.

Users running at very large scale can implement a custom RemoteLogMetadataManager (e.g., backed by an external KV store) by setting remote.log.metadata.manager.class.name, but the topic-based default is the supported path.

Configuration

Top-level (cluster):

• remote.log.storage.system.enable=true — turn on tiered storage globally.
• remote.log.storage.manager.class.name — fully-qualified plugin class implementing RemoteStorageManager.
• remote.log.storage.manager.class.path — optional classpath entry for the plugin JAR.
• remote.log.metadata.manager.class.name — defaults to TopicBasedRemoteLogMetadataManager.
• remote.log.manager.thread.pool.size, remote.log.manager.task.interval.ms — thread tuning.
• remote.log.index.file.cache.total.size.bytes — index cache budget.

Per-topic (override on kafka-configs.sh --topic):

• remote.storage.enable=true — opt the topic into tiering.
• local.retention.ms, local.retention.bytes — when to start uploading.
• retention.ms, retention.bytes — total retention; segments older than this are deleted from both local and remote.

Operational model

• Broker disks no longer need to scale with retention — only with active retention.
• Reads of cold data (rarely-accessed offsets) go to the remote store. Latency depends on the backend: S3-class backends typically add 100s of ms per fetch.
• Tiering is broker-local: each leader handles its own copy / delete tasks. Followers replicate from the leader's local segments only — they do not double-upload.
• On leader failover, the new leader picks up tiering for its partitions; segments already uploaded under previous leaders remain accessible (the metadata records are durable).

Failure modes

• Plugin error during upload — the segment stays local and is retried later. Metrics under kafka.server.RemoteLogManager expose failure counts.
• Plugin error during fetch — the consumer sees a KafkaStorageException. Subsequent fetches will retry; clients should be prepared for transient failures during cold reads.
• __remote_log_metadata corruption — extremely rare; recovery requires re-bootstrapping metadata from the remote store via the plugin's listRemoteLogSegments (manually orchestrated).
• Stuck deletes — a segment whose metadata says DELETE_SEGMENT_STARTED but whose plugin call failed. The RemoteLogManager retries on a schedule; persistent failures need operator intervention.

Entry points for modification

• Implement a new backend: a JAR with a RemoteStorageManager impl and (optionally) a RemoteLogMetadataManager. Drop on classpath, set remote.log.storage.manager.class.name. No core changes required.
• Tune scheduling: RemoteLogManager.java — be cautious, the task scheduler is sensitive to backpressure.
• Change index caching: RemoteIndexCache.java. Eviction policy is LRU; growing the cache budget is usually preferable to changing the algorithm.
• Tweak fetch behavior: UnifiedLog.java (the local-vs-remote dispatch) and RemoteLogManager.read*.

Related pages

• Modules: storage — segment / index / UnifiedLog background.
• Modules: core (broker) — ReplicaManager is the read entry point.
• Reference: Configuration.