What are the RemoteStorageManager (RSM) and RemoteLogMetadataManager (RLMM) plugins, and how do their responsibilities differ?

## Why two plugins Tiered storage has two fundamentally different jobs: 1. **Move and retrieve bytes** to/from an external store (which could be S3, GCS, Azure Blob, HDFS, or a vendor system). 2. **Remember what was moved** — for each remote segment, which offsets it covers, what leader epochs it spans, when it was written, and what lifecycle state it's in — so that a read for offset X can be routed to the right remote object. KIP-405 deliberately splits these into two interfaces so each can be implemented and scaled independently. ## RemoteStorageManager (RSM) — the data plane Interface: **org.apache.kafka.server.log.remote.storage.RemoteStorageManager**. Core methods: - **copyLogSegmentData(metadata, segmentData)** — uploads a closed segment file plus its associated index files: the **offset index**, **time index**, **transaction index**, **producer-snapshot**, and the **leader-epoch checkpoint**. These indexes are needed so reads can locate positions without the broker holding the data locally. - **fetchLogSegment(metadata, startPosition[, endPosition])** — returns an InputStream over a byte range of the remote segment (this is what serves a remote read). - **fetchIndex(metadata, indexType)** — retrieves a specific index for a remote segment. - **deleteLogSegmentData(metadata)** — removes the remote object when retention expires. The RSM hides all object-store specifics (auth, bucket layout, multipart upload, retries). ## RemoteLogMetadataManager (RLMM) — the control plane Interface: **org.apache.kafka.server.log.remote.storage.RemoteLogMetadataManager**. It manages **RemoteLogSegmentMetadata** records, each carrying: - a unique **RemoteLogSegmentId**, - **startOffset / endOffset**, - **leader-epoch → start-offset** map (segment epoch lineage), - **maxTimestamp**, broker/event timestamps, - a **state** in a small state machine: COPY_SEGMENT_STARTED → COPY_SEGMENT_FINISHED, and later DELETE_SEGMENT_STARTED → DELETE_SEGMENT_FINISHED. Key lookups it serves: - **remoteLogSegmentMetadata(partition, leaderEpoch, offset)** — 'which remote segment contains this offset for this epoch?' (drives the read path). - **highestOffsetForEpoch / listRemoteLogSegments** — used for retention and recovery. The default implementation, **TopicBasedRemoteLogMetadataManager**, stores these records in an internal compacted-style Kafka topic called **__remote_log_metadata**, partitioned so each user partition's metadata maps to a metadata partition. This makes the metadata itself replicated and durable using Kafka's own machinery. ## How they fit together On each broker, the **RemoteLogManager (RLM)** component runs background tasks per leader partition. It (1) detects closed segments eligible to offload, (2) writes a COPY_SEGMENT_STARTED metadata record via RLMM, (3) calls RSM.copyLogSegmentData, (4) writes COPY_SEGMENT_FINISHED. On reads it asks RLMM for the segment covering the requested offset, then streams bytes via RSM.fetchLogSegment. For retention it marks DELETE_SEGMENT_STARTED, calls RSM.deleteLogSegmentData, then DELETE_SEGMENT_FINISHED. ## Edge cases - The two-phase STARTED/FINISHED states make offload and delete **idempotent and crash-safe**: if a broker dies mid-copy, a segment left in COPY_SEGMENT_STARTED is not yet readable and can be retried. - Metadata correctness is critical: a segment in COPY_SEGMENT_FINISHED but whose bytes are missing remotely would break reads, so RSM must guarantee durability before FINISHED is written. - Both plugins are configured via classpath + config (remote.log.storage.manager.class.name, remote.log.metadata.manager.class.name).