05 · Tiered Storage (Remote Log)

Source: Apache Kafka 4.4.0-SNAPSHOT (git 04bfe7d, 2026-06-15), KRaft mode. Derived from source code, not copied from official documentation.

Tiered storage (KIP-405) lets a Kafka topic keep a small, hot tail of log segments on local disk while offloading older, immutable segments to an external object store. The broker-side orchestrator is RemoteLogManager; it drives two pluggable SPIs, RemoteStorageManager (RSM) for byte movement of segment data and indexes, and RemoteLogMetadataManager (RLMM) for tracking RemoteLogSegmentMetadata with strongly-consistent semantics. Leaders run a copy task that uploads eligible segments past the local-retention boundary, an expiration task that enforces remote retention, and a follower task that tracks the highest-copied offset; consumers that fetch below the local log start are served by streaming the segment back from remote storage through a thread-pooled read path, with local index files cached by RemoteIndexCache. The default RLMM, TopicBasedRemoteLogMetadataManager, persists metadata as records in the internal __remote_log_metadata topic.

Role & responsibilities

Without tiered storage, a topic's retention is bounded by the aggregate local disk of the brokers that host its replicas. Tiered storage decouples retention from local disk: the broker copies sealed (non-active) segments to a remote object store, then deletes the corresponding local copies once they fall outside the (smaller) local retention. Reads for offsets that exist only in remote storage are transparently served by fetching from the object store. The RemoteLogManager Javadoc enumerates its duties at storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java:144: initialize the RSM and RLMM instances; react to leader/follower/stop-partition events; expose APIs to fetch indexes and metadata; copy log segments to remote; and clean up segments expired by remote retention.

Tiered storage

System feature enabled cluster-wide by remote.log.storage.system.enable (default false) and per-topic by remote.storage.enable.

Local retention

How long/large the locally-kept tail may be (log.local.retention.ms / .bytes). Always ≤ the topic's total retention.

Remote retention

Effectively the topic's total retention.ms / retention.bytes, applied to remote segments by the expiration task.

Copied offset

Highest offset known to be durably in remote storage for the partition; gates local deletion.

RSM / RLMM

The two plugin SPIs: RemoteStorageManager and RemoteLogMetadataManager.

Where it lives in the code

Core concepts & terminology

The two SPIs

RemoteStorageManager (RemoteStorageManager.java:52) is Configurable + Closeable and exposes exactly five operations: copyLogSegmentData(metadata, LogSegmentData) returning an optional CustomMetadata; two overloads of fetchLogSegment(metadata, startPosition[, endPosition]) returning an InputStream; fetchIndex(metadata, IndexType) for one of the five IndexType values OFFSET, TIMESTAMP, PRODUCER_SNAPSHOT, TRANSACTION, LEADER_EPOCH (RemoteStorageManager.java:57); and deleteLogSegmentData(metadata). Copy and delete must be idempotent (re-copy overwrites; delete of a missing object must not throw). Plugins signal transient failures with RetriableRemoteStorageException so the RLM can retry without cancelling the task (RemoteStorageManager.java:47).

RemoteLogMetadataManager (RemoteLogMetadataManager.java:55) provides async, future-returning mutators, addRemoteLogSegmentMetadata (must be in state COPY_SEGMENT_STARTED), updateRemoteLogSegmentMetadata (state transition), putRemotePartitionDeleteMetadata, plus synchronous lookups: remoteLogSegmentMetadata(tpId, epoch, offset), highestOffsetForEpoch, listRemoteLogSegments(tpId[, epoch]) (epoch variant sorted by start offset), remoteLogSize(tpId, epoch), and the default-method nextSegmentWithTxnIndex. Lifecycle callbacks onPartitionLeadershipChanges and onStopPartitions tell the RLMM which partitions this broker now serves, and isReady(tpId) gates task execution until the partition's metadata is loaded.

Segment state machine

Every remote segment moves through four states defined in RemoteLogSegmentState.java:51, each with a 1-byte id used on the wire:

isValidTransition (RemoteLogSegmentState.java:90) enforces the lattice: from null only COPY_SEGMENT_STARTED; a self-transition is always valid (idempotency under retry/failover); COPY_SEGMENT_STARTED → {FINISHED, DELETE_STARTED}; COPY_SEGMENT_FINISHED → DELETE_STARTED; DELETE_STARTED → DELETE_FINISHED. The cache keeps every segment that has not reached DELETE_SEGMENT_FINISHED; only COPY_SEGMENT_FINISHED segments are visible to reads (RemoteLogMetadataCache.java:78 table).

Data structures

RemoteLogSegmentId & RemoteLogSegmentMetadata

RemoteLogSegmentId is (TopicIdPartition topicIdPartition, Uuid id) with a fresh random Uuid generated for each copy attempt via generateNew (RemoteLogSegmentId.java:40). Re-uploading the same segment after a failure produces a new id, which is how copy stays idempotent without coordinating on object names.

RemoteLogSegmentMetadata (RemoteLogSegmentMetadata.java:37) carries the field set below. Note the constructor invariants: startOffset ≥ 0, endOffset ≥ startOffset, and segmentLeaderEpochs must be non-empty (a single-epoch segment still maps that epoch → its start offset). It extends RemoteLogMetadata, which adds brokerId and eventTimestampMs.

createWithUpdates(RemoteLogSegmentMetadataUpdate) (RemoteLogSegmentMetadata.java:310) produces a new immutable instance with the update's brokerId, eventTimestampMs, customMetadata, and state applied, offsets, epochs, and size are carried over unchanged. CustomMetadata (RemoteLogSegmentMetadata.java:384) wraps a byte[] whose accepted maximum is remote.log.metadata.custom.metadata.max.bytes (default 128; RemoteLogManagerConfig.java:88).

LogSegmentData, the five artefacts

copyLogSegmentData receives a LogSegmentData bundling the segment and its auxiliary indexes (LogSegmentData.java:29): logSegment (the .log file), offsetIndex (.index), timeIndex (.timeindex), transactionIndex (Optional<Path>, may be absent), producerSnapshotIndex (the producer-state snapshot at the next segment's base offset), and leaderEpochIndex, a ByteBuffer rather than a path. The RLM builds that buffer by serializing the leader-epoch checkpoint entries up to the next base offset via epochEntriesAsByteBuffer using the LeaderEpochCheckpointFile.FORMATTER (RemoteLogManager.java:2256).

On-wire metadata record & on-disk index files

The default RLMM serializes each RemoteLogMetadata event to bytes with RemoteLogMetadataSerde and produces it as the value of a record on __remote_log_metadata (key is null; partition is chosen explicitly, see partitioner below) (ProducerManager.java:83). On the read path, the local index files cached by RemoteIndexCache are named {startOffset}_{uuid}{suffix}, e.g. 45_<uuid>.index / .timeindex / .txnindex, under $logdir/remote-log-index-cache (RemoteIndexCache.java:713, :83). A remote segment's data stream is parsed batch-by-batch by RemoteLogInputStream, which reads the fixed log header up to the magic byte, derives the batch size from the SIZE_OFFSET field, allocates LOG_OVERHEAD + size bytes, and constructs a DefaultRecordBatch (magic > v1) over that buffer (clients/.../record/internal/RemoteLogInputStream.java:41).

Architecture & control/data flow

Task model on leadership change

onLeadershipChange(becomeLeader, becomeFollower, topicIds) (RemoteLogManager.java:469) filters to partitions whose UnifiedLog.remoteLogEnabled() is true, caches topic-id ↔ partition mappings, notifies the RLMM via onPartitionLeadershipChanges, then installs tasks. Leaders get a RLMCopyTask (only if remote.log.copy.disable=false) and a RLMExpirationTask; followers get a RLMFollowerTask (doHandleLeaderPartition at :2150, doHandleFollowerPartition at :2176). Each task is scheduled with scheduleWithFixedDelay(task, 0, remote.log.manager.task.interval.ms, MS) on its pool. Converting roles cancels the opposing tasks first via RLMTaskWithFuture.cancel(), which sets a volatile cancelled flag and calls future.cancel(true) (:2207).

Detailed mechanics

The copy task

RLMCopyTask.execute (RemoteLogManager.java:880) first resets per-task state if the log directory changed (an alter-logdir guard, KAFKA-16711), then calls copyLogSegmentsToRemote(log). The algorithm (:996):

1. Establish offsets. On first run after becoming leader, maybeUpdateLogStartOffsetOnBecomingLeader computes the remote log-start via findLogStartOffset (start offset of the earliest segment found by walking leader epochs from the earliest entry forward; falls back to localLogStartOffset) and pushes it through updateRemoteLogStartOffset (:892, :2121). maybeUpdateCopiedOffset computes the highest already-copied offset via findHighestRemoteOffset, walking leader epochs from the latest entry backward, asking the RLMM highestOffsetForEpoch, and reconciling against the epoch's end offset (:902, :2091). It seeds log.updateHighestOffsetInRemoteStorage(...).
2. Select candidates. With copiedOffset and the partition's last-stable-offset (LSO), candidateLogSegments(log, fromOffset, lso) returns all sealed segments (never the active segment) whose base offset ≤ LSO, in order, stopping at the first segment that is not yet eligible by copy-lag (:925). fromOffset = max(copiedOffset+1, logStartOffset).
3. Copy-lag gate. isEligibleForUpload (:949) returns true immediately if either remote.copy.lag.ms or remote.copy.lag.bytes is 0 (the short-circuit at :957 is an ||); otherwise a segment is eligible if its age (now − largestTimestamp) ≥ copy-lag-ms, or the bytes of log after the segment ≥ copy-lag-bytes.
4. Quota wait. Before each segment, under copyQuotaManagerLock, the task polls rlmCopyQuotaManager.getThrottleTimeMs() and, while > 0, records the throttle and awaits on copyQuotaManagerLockCondition for up to quotaTimeout() = 1 s, re-checking each cycle; on clearance it records the segment's byte size against the quota and signals other waiters (:1028).
5. Per-segment copy. The caller copyLogSegmentsToRemote generates a fresh RemoteLogSegmentId via generateNew (:1049) and registers it in segmentIdsBeingCopied (:1050), then passes it to copyLogSegment (:1079), which builds the COPY_SEGMENT_STARTED metadata (offsets, largestTimestamp, size, segment leader epochs, isTxnIdxEmpty), and addRemoteLogSegmentMetadata(...).get(), blocking until the RLMM persists it. It then assembles LogSegmentData and calls RSM.copyLogSegmentData. On success it writes COPY_SEGMENT_FINISHED via updateRemoteLogSegmentMetadata(...).get(), advances the in-memory copiedOffsetOption to (endOffset, lastEpochInSegment), and calls log.updateHighestOffsetInRemoteStorage(endOffset).

Custom-metadata overflow

If the RSM returns CustomMetadata larger than the configured cap, the task does not store the finished update; instead it reconstructs the finished metadata only to delete the just-copied data, throws CustomMetadataSizeLimitExceededException, and cancel()s the copy task for that partition (RemoteLogManager.java:1128, caught at :1061). This stops further copying for the partition until intervention, surfacing as failedRemoteCopyRequestRate.

The follower task

Followers do not copy; RLMFollowerTask.execute simply calls findHighestRemoteOffset and log.updateHighestOffsetInRemoteStorage(...) (RemoteLogManager.java:1662). This keeps a follower's local deletion guard in sync so that when it later becomes leader (or while it serves follower fetches) it will not delete local segments that the cluster believes are not yet remote.

The expiration task, remote retention

RLMExpirationTask.cleanupExpiredRemoteLogSegments (RemoteLogManager.java:1410) enforces remote retention and removes dangling/unreferenced segments. Step by step:

1. Tally. calculateMetadataAndSize iterates all segments to compute metadata count, total tracked size, and the size of only COPY_SEGMENT_FINISHED segments; these feed gauges RemoteLogMetadataCount and RemoteLogSizeBytes (:1383).
2. Build retention contexts. buildRetentionTimeData(retention.ms) yields a cleanupUntilMs = now − retentionMs threshold (only if retentionMs > -1 and the threshold is ≥ 0) (:1575). buildRetentionSizeData(...) computes how many remote bytes breach retention.bytes, validating each finished segment against the current leader-epoch lineage (:1594). The first time, it iterates and sums valid segments; once it proves all segments are valid (isAllSegmentsValid) it short-circuits to the cheaper cached size.
3. Iterate epochs in order. For each leader epoch, for each segment (sorted by start offset): skip segments currently being copied (sets canProcess=false, :1465); re-publish a delete for any segment stuck in DELETE_SEGMENT_STARTED (dangling-delete retry, :1473); decide deletion via, in priority order, log-start-offset breach (isSegmentBreachByLogStartOffset), then, if the segment is within the leader-epoch lineage, retention-time breach, then retention-size breach (:1488).
4. Advance remote log-start. Each deletion reason records the candidate's endOffset + 1; the maximum is pushed via handleLogStartOffsetUpdate before the actual deletes, so followers can pick up the new log-start even if this leader later fails the deletes (:1509; the code's comment at :1511 reasons through leader-change races).
5. Delete. Each candidate goes through deleteRemoteLogSegment (:1677): publish DELETE_SEGMENT_STARTED → RSM.deleteLogSegmentData → publish DELETE_SEGMENT_FINISHED, all blocking, with remote-delete request/failure rates marked.
6. Unreferenced cleanup. Finally it deletes segments whose leader epochs are all below the leader's earliest known epoch, unreferenced after unclean leader election (deleteLogSegmentsDueToLeaderEpochCacheTruncation, :1340, driven from :1539).

The remote read path

When a consumer fetch targets an offset below the local log start, ReplicaManager.handleOffsetOutOfRangeError detects logStartOffset ≤ offset < localLogStartOffset and routes to remote (core/.../server/ReplicaManager.scala:1904). It first consults the fetch quota: if remoteLogManager.getFetchThrottleTimeMs > 0, it returns an empty FetchDataInfo with UNKNOWN_OFFSET_METADATA (so other partitions in the request can still be served via delayed fetch) instead of scheduling a remote read (ReplicaManager.scala:1917). Otherwise it builds a RemoteStorageFetchInfo and the fetch becomes a DelayedRemoteFetch; processRemoteFetch calls rlm.asyncRead(fetchInfo, callback) (ReplicaManager.scala:1610).

asyncRead submits a RemoteLogReader to the bounded remoteStorageReaderThreadPool (RemoteLogManager.java:2145). RemoteLogReader.call() times rlm.read(fetchInfo) against the remoteReadTimer, records bytes against the fetch quota, and invokes the callback with a RemoteLogReadResult (RemoteLogReader.java:61). The heavy lifting is RemoteLogManager.read (:1834):

1. Resolve the leader epoch for the fetch offset from the leader-epoch cache and look up the covering RemoteLogSegmentMetadata via the RLMM; if none, throw OffsetOutOfRangeException (:1852).
2. lookupPositionForOffset uses RemoteIndexCache.lookupOffset on the cached offset index to get a byte startPos (:1933); open RSM.fetchLogSegment(meta, startPos) and wrap it in a RemoteLogInputStream.
3. findFirstBatch scans forward, skipping batches whose lastOffset < offset, summing skippedBytes (:2077). If a segment is exhausted (e.g. the offset was compacted away), findNextSegmentMetadata advances to the next segment and repeats (:1870).
4. Copy the first batch plus as many subsequent bytes as fit into a min(fetchMaxBytes, partition.maxBytes) buffer; honour minOneMessage by returning at least the first batch even if it exceeds the cap, or an empty result if not (:1887).
5. If isolation is TXN_COMMITTED, addAbortedTransactions walks the txn indexes of this and following segments (then local segments) up to the fetch upper bound, accumulating aborted-transaction markers for the client to filter (:1937, :1979).

RemoteIndexCache

Re-downloading index files for every remote fetch would be ruinous, so RemoteIndexCache keeps offset/time/txn indexes on local disk under $logdir/remote-log-index-cache and memory-maps them (RemoteIndexCache.java:78). It is a Caffeine cache weighted by entry byte size with maximum weight remote.log.index.file.cache.total.size.bytes (default 1 GiB) and optional expireAfterAccess = remote.log.index.file.cache.ttl.ms (default 900 000 ms; -1 disables) (:171). On a miss, loadIndexFile downloads via RSM.fetchIndex into a .tmp file and atomically renames it into place, then opens and sanity-checks the index; a missing transaction index (RemoteResourceNotFoundException) is tolerated by writing an empty file (:362, :441). On startup, init() deletes leftover .tmp/.deleted files and re-hydrates entries from any complete triplet on disk (:275).

Eviction is two-phase to avoid mutating files under in-flight reads. The Caffeine evictionListener only enqueues the entry; the dedicated remote-log-index-cleaner KafkaScheduler thread runs markForCleanup (rename to .deleted) then cleanup (delete) after a fileDeleteDelayMs (10 s) delay (:195, :248). Concurrency uses a class-level ReadWriteLock (reads take the read lock; close/resize take the write lock) plus a per-Entry ReentrantReadWriteLock so cleanup never races a lookup (:99, :518). The class doc notes the policy is Caffeine's Window-TinyLfu.

The default RLMM: TopicBasedRemoteLogMetadataManager

The default metadata store persists every event as a record in the internal compaction-disabled topic __remote_log_metadata (TopicBasedRemoteLogMetadataManagerConfig.java:44). The topic defaults: 50 partitions, replication factor 3, min.insync.replicas 2, cleanup policy delete, retention -1 (unlimited) (:54, newRemoteLogMetadataTopic at TopicBasedRemoteLogMetadataManager.java:432). Because retention is unlimited by default and the topic is delete-not-compact, the full event history is replayable.

Producing & the read-your-writes guarantee

storeRemoteLogMetadata publishes the event through ProducerManager (an idempotent, acks=all KafkaProducer; ProducerManager.java:48, TopicBasedRemoteLogMetadataManagerConfig.java:238) and then chains a stage that blocks until the local ConsumerManager has consumed up to that record's offset (TopicBasedRemoteLogMetadataManager.java:153). That is how the RLM's blocking add(...).get() / update(...).get() calls get strong consistency: the future does not complete until the in-memory cache reflects the write. The deterministic partition for a user partition is computed by RemoteLogMetadataTopicPartitioner.metadataPartition = murmur2(hash(topicId, partition)) mod numMetadataPartitions (RemoteLogMetadataTopicPartitioner.java:35), both producer and consumer use it so all events for a user partition land on one metadata partition.

Consuming & the in-memory cache

A single ConsumerTask thread polls the assigned metadata partitions and applies each deserialized event to a RemotePartitionMetadataEventHandler (ConsumerTask.java:65, :166). Assignment is keyed by metadata partition, not user partition: when partitions are assigned, the task computes the set of metadata partitions, consumer.assign(...)s them, seeks newly-assigned ones to the beginning (full replay) and others to the last processed offset (:224). A user partition is marked ready only once the consumer has read up to the metadata partition's end offset at assignment time, or the metadata partition is empty (maybeMarkUserPartitionsAsReady, :185); until then isReady(tpId) is false and the RLM skips that partition's tasks and read requests.

The handler delegates to a per-user-partition RemoteLogMetadataCache (RemoteLogMetadataCache.java:96). It holds two maps: idToSegmentMetadata (segment-id → metadata for all non-terminal segments) and leaderEpochEntries (leader epoch → RemoteLogLeaderEpochState tracking the floor structure and highest offset per epoch). Lookups respect state: remoteLogSegmentMetadata(epoch, offset) returns only COPY_SEGMENT_FINISHED segments and verifies the offset lies within the epoch's boundary inside that segment (:132); listRemoteLogSegments(epoch) returns started, finished, and delete-started (for cleanup) but never delete-finished. State updates run through checkStateTransition and drop invalid transitions rather than corrupting the cache (:344); a DELETE_SEGMENT_FINISHED event removes the segment entirely (:240).

Concurrency & threading

The RLM runs several distinct pools, all created in the constructor (RemoteLogManager.java:256):

The scheduled pools set removeOnCancelPolicy(true) and disable running delayed/periodic tasks after shutdown (RemoteLogManager.java:2324). Each scheduled task starts with a fast isCancelled() / isReady() guard (:826) so a cancelled-then-rescheduled task exits cheaply. State that crosses threads is protected as follows:

• Task registries, leaderCopyRLMTasks, leaderExpirationRLMTasks, followerRLMTasks are ConcurrentHashMaps, mutated with computeIfAbsent/computeIfPresent so install/cancel are atomic (:184).
• In-flight copies, segmentIdsBeingCopied is a concurrent set; the expiration task consults it to avoid deleting a segment mid-copy (:187, :1465).
• Copy quota, a fair ReentrantLock copyQuotaManagerLock with a Condition serializes copy threads so they wait and re-check the byte-rate together (:169, :1028).
• Per-task offsets, copiedOffsetOption, isLogStartOffsetUpdated, logDirectory are volatile within each RLMCopyTask (single executor thread mutates them; reads may come from cancel paths) (:870).
• Index cache, class-level ReadWriteLock + per-entry lock as described above.

Quotas & throttling

Two independent RLMQuotaManagers bound aggregate (broker-global, all-partition) byte rates: copy (local→remote) and fetch (remote→local), both defaulting to Long.MAX_VALUE (unbounded) (RemoteLogManager.java:366, :372). Each uses a SimpleRate sensor over num windows of window.size.seconds (defaults 11 windows × 1 s) and a Quota bound; getThrottleTimeMs calls sensor.checkQuotas() and, on QuotaViolationException, returns QuotaUtils.throttleTime(...) (RLMQuotaManager.java:80). The two managers differ in where they apply: the copy quota makes the copy task wait before uploading a segment (back-pressure), whereas the fetch quota makes the read path shed the request, returning empty so the consumer retries later rather than queueing remote reads (ReplicaManager.scala:1917). Throttle times are surfaced via remote-copy-throttle-time / remote-fetch-throttle-time sensors (RemoteLogManager.java:244).

Configuration reference

Failure modes, edge cases & recovery

• Crash mid-copy. If the broker dies after COPY_SEGMENT_STARTED but before FINISHED, the local segment is retained (deletion guard never advanced), and the started metadata sits dangling. The new leader simply re-copies with a fresh uuid; the dangling started-record is harmless and excluded from size/visibility, and the expiration task can clean up segments stuck in DELETE_SEGMENT_STARTED.
• RSM retriable error. RetriableRemoteStorageException propagates up and the scheduled task retries on its next tick; the run() wrapper logs it at debug and does not cancel (RemoteLogManager.java:848).
• RSM hard error during copy. The code best-effort deletes the partially-copied segment, marks failedRemoteCopyRequestRate, and rethrows; the started-metadata remains until the next cleanup pass (:1114).
• Read for a compacted/missing offset. read iterates to the next segment if the first batch is not found; if the offset truly is not in remote, it throws OffsetOutOfRangeException, returned to the client (:1856, :1870).
• Read queue full. asyncRead throws RejectedExecutionException when the 100-slot queue is full; the fetch is served with an error rather than blocking (:2143).
• Missing transaction index. Old metadata may report txnIdxEmpty=false yet have no remote txn index; getTransactionIndex uses ofNullable for backward compatibility, and the index cache substitutes an empty index on RemoteResourceNotFoundException (:2007, RemoteIndexCache.java:444).
• Partition deletion. stopPartitions cancels all three tasks, and when deleteRemoteLog is set, deleteRemoteLogPartition publishes DELETE_STARTED for every segment, deletes data via the RSM, clears the index cache, then publishes DELETE_FINISHED (:524, :576).
• Metadata topic unavailable at boot. The default RLMM exits the JVM after the init timeout (see Caution above); operationally this means the metadata topic must be healthy for tiered brokers to start.

Invariants & guarantees

• A local segment is deleted only after its data is durably in remote (highestOffsetInRemoteStorage advanced post-FINISHED), no data loss window (UnifiedLog.java:1857, RemoteLogManager.java:1160).
• Metadata is read-your-writes consistent: the RLM's blocking metadata calls return only after the local cache reflects the write (TopicBasedRemoteLogMetadataManager.java:158).
• Only COPY_SEGMENT_FINISHED segments are readable; started/deleting segments are invisible to fetch but available for cleanup (RemoteLogMetadataCache.java:132).
• Each copy attempt has a globally-unique id, so the copy/delete RSM operations are idempotent and overwriting (RemoteLogSegmentId.java:24).
• Retention & size accounting are evaluated within the current leader-epoch lineage; segments outside it are treated as unreferenced (RemoteLogManager.java:1728).
• Local effective retention is always ≤ total retention: log.local.retention.{ms,bytes} defaults of -2 resolve to the corresponding total-retention value (RemoteLogManagerConfig.java:159).

Interactions with other subsystems

• The Log Storage Engine, UnifiedLog supplies sealed LogSegments, the leader-epoch cache, producer-state snapshots, and highestOffsetInRemoteStorage; tiered storage reuses the same on-disk index/segment formats (Record Format & Batches) when streaming remote data.
• Log Management, Retention & Compaction, local deletion is performed by UnifiedLog.deleteOldSegments() driven by local retention, gated by the remote copy state; tiered topics use cleanup.policy=delete only.
• The Fetch Path & Replica Fetchers, ReplicaManager translates an out-of-range fetch into a DelayedRemoteFetch and consults the fetch quota; followers fetching a remote-only offset receive OffsetMovedToTieredStorageException (ReplicaManager.scala:1914).
• Replication, ISR & High Watermark, copy eligibility is bounded by the last-stable-offset so only committed data is tiered; leader/follower transitions install/cancel RLM tasks.
• Metadata Propagation & Broker Lifecycle, leadership and stop-partition events from the controller drive onLeadershipChange/stopPartitions; the default RLMM defers init until the broker is ready.
• Transactions & Exactly-Once Semantics, the remote read path reconstructs aborted-transaction lists from remote (then local) txn indexes for read_committed consumers.
• Quotas, Throttling & Client Metrics, RLM copy/fetch quotas are a separate QuotaType (RLM_COPY, RLM_FETCH) from client request/bandwidth quotas but share the metrics/sensor machinery.

Design rationale & evolution

Tiered storage was introduced by KIP-405, deliberately structured around two narrow SPIs so cloud object stores (S3/GCS/ABS) and HDFS-like systems can be plugged in without touching the broker. The metadata/data split, strongly-consistent metadata, eventually-consistent data, lets the broker reason about correctness from a replicated log of events (the default RLMM) while treating the object store as a dumb blob sink. Tiered-storage quotas (KIP-956) added the global copy/fetch byte-rate limiters to keep catch-up traffic from overwhelming the object store or the broker. The follower thread pool was split out from the legacy single thread pool and the old remote.log.manager.thread.pool.size deprecated in 4.2 (RemoteLogManagerConfig.java:109). Separating copier and expiration pools lets uploads and deletes scale independently.

Gotchas / operational notes