Tiered Storage

Architecture

Key Numbers

Configuration

# Cluster-level: enable the remote storage system
remote.log.storage.system.enable=true

# Storage manager plugin (Confluent S3, AWS S3, custom)
remote.log.storage.manager.class.name=org.apache.kafka.server.log.remote.storage.RemoteLogStorageManager
remote.log.storage.manager.class.path=/opt/kafka/plugins/tiered/*

# Implementation-specific (S3 example)
rsm.config.s3.bucket.name=my-kafka-tier
rsm.config.s3.region=us-east-1
rsm.config.s3.credentials.provider.class=com.amazonaws.auth.DefaultAWSCredentialsProviderChain

# Metadata manager (uses __remote_log_metadata internal topic)
remote.log.metadata.manager.class.name=org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager
remote.log.metadata.topic.replication.factor=3
remote.log.metadata.topic.num.partitions=50

# Per-topic: enable tiered storage and set local retention
$ kafka-topics.sh --bootstrap-server kafka-1:9092 --create \
    --topic events \
    --partitions 100 \
    --replication-factor 3 \
    --config remote.storage.enable=true \
    --config local.retention.ms=604800000 \
    --config retention.ms=7776000000

# local.retention.ms = 7 days  (kept on broker disk)
# retention.ms       = 90 days (total: local + remote)
# local.retention.bytes = -1   (no size cap on local; use ms)

Upload Path

# RemoteLogManager runs on each broker, processes its leader partitions
# Periodic task (every remote.log.manager.task.interval.ms = 30s default):

for partition in leader_partitions:
  segments = list_local_segments(partition)
  closed_old_enough = [s for s in segments
                       if s.is_closed() and (now - s.last_modified) > local.retention.ms]
  for segment in closed_old_enough:
    if not already_in_remote_metadata(segment):
      # Upload .log + .index + .timeindex + .producerSnapshot + .leaderEpochCheckpoint
      remote_id = generate_uuid()
      remote_storage_manager.copyLogSegmentData(remote_id, segment)
      remote_log_metadata_manager.addRemoteLogSegmentMetadata(metadata)
      # Now safe to delete locally
      delete_local_segment(segment)

Only the partition leader uploads. Followers replicate as before from the leader's local copy. When leadership changes, the new leader checks remote metadata and resumes uploading from where the previous leader left off — no redundant uploads.

Fetch Path: Remote Read

# Consumer fetches offset 0.5M (in remote tier)
# Broker handling FetchRequest:

if requested_offset >= local_log_start_offset:
  # Hot path — local disk, sendfile() to socket
  return read_local_segment(requested_offset)
else:
  # Cold path — remote tier
  metadata = remote_log_metadata_manager.find(partition, offset=0.5M)
  if metadata is None:
    return OFFSET_OUT_OF_RANGE
  # Stream from S3 directly to consumer
  remote_input_stream = remote_storage_manager.fetchLogSegment(metadata, position)
  copy_bytes(remote_input_stream, fetch_response_buffer, max_bytes)
  return fetch_response

# The broker does not persist the remote bytes locally.
# It also does not page-cache them — cold reads cost full S3 latency every time.

The broker uses a small in-memory cache for remote-segment indexes (so it does not re-fetch the .index file on every read), but the actual log data is streamed pass-through. This means cold reads are slower but they do not impact the page-cache hit rate of the hot tail.

Cost Economics

A worked example: 100 brokers, 5 GB/s throughput, 30-day retention. Without tiered storage, that is ~13 PB on local SSD across the cluster — about $130K/month at gp3 prices. With tiered storage (1 day local, 29 days remote), it is ~430 TB local + 12.6 PB in S3 = ~$4K + $290K = wait. The savings come from broker count, not just disk: smaller local disks mean fewer brokers (since you no longer need to spread storage across many machines), so the total cost drops.

When It Doesn't Help

Operational Notes

# Key metrics to watch
kafka.log.remote:type=RemoteStorageThreadPool,name=RemoteLogReaderTaskQueueSize
kafka.log.remote:type=RemoteLogManager,name=RemoteCopyBytesPerSec
kafka.log.remote:type=RemoteLogManager,name=RemoteFetchBytesPerSec
kafka.log.remote:type=RemoteLogManager,name=RemoteCopyLagSegments
kafka.log.remote:type=RemoteLogManager,name=RemoteDeleteErrorsPerSec

# Throttling — bound the bandwidth used by tiered uploads
remote.log.manager.copier.task.thread.pool.size=10
remote.log.manager.thread.pool.size=10
remote.log.manager.copy.max.bytes.per.second=104857600  # 100 MB/s per broker

# Disable tiered storage on a topic (topic must be re-created)
$ kafka-configs.sh --alter --entity-type topics --entity-name events \
    --add-config remote.storage.enable=false  # not allowed once enabled, must recreate

Frequently Asked Questions