Page History

...

RLM will ship all the LogSegments and the corresponding OffsetIndex to RemoteStorage. A new index file, **<code>RemoteLogSegmentIndex,</code></strong> is maintained locally on the Kafka broker per topic-partition like all the existing index files are stored today as shown below:   

{log.dirs}/{topic-partition}/0000001000121.remoteindex

...

Remote index maintains the LogSegment starting offsets for a completed segment in 4-byte number relative to the starting offset of remote index file as shown below:

FileName : 00000001000121.remoteindex 

Contents:
SegmentStartOffset
1000121 
1500024 
2000011 
…
2999999

FileName : 00000003000000.remoteindex 
Contents:
SegmentStartOffset
3000000 
3500024 
4000011 
…

RLM maintains these RemoteLogSegmentIndexes per topic-partition in local files on the Kafka broker. These files are rolled on a periodic basis with starting index of first LogSegment in the file name. Note that the RemoteLogSegmentIndex can be constructed by listing all the log segments stored on the remote storage. Maintaining a local file is an optimization to avoid such listing operations that may be slow and expensive depending on the external store. RemoteLogSegmentIndex files are MMAP'ed files and will follow a similar binary search mechanism as OffsetIndex files to find a LogSegment to serve a read operation.

On `OutOfRangeOffsetException`, ReplicaManager delegates the read request to RLM which does the following:

...

Compacted topics will not have remote storage support. 

Configs

RemoteLogManager (RLM)

RemoteLogManager is a new interface added to the broker. It is responsible to copy the completed log segments to the remote storage and update RemoteOffsetIndex file. The default implementation of interface supports HDFS as remote storage. Additional remote storage support, such as S3 can be added later by providing other implementations using the configuration `Remote.log.manager.class`.

...

ReplicaManager.readLocalLog works as it does today. But only in case of OffsetOutOfRange of exception and RLM is configured we will delegate the read request to RLM which returns LogReadResult

def readFromLocaLog(): Seq[(TopicPartition, LogReadResult)] = {
catch {
case e@ (_: OffsetOutOfRangeException) =>
    RemoteLogManager.read(fetchMaxBytes: Int,hardMaxBytesLimit: Boolean, readPartitionInfo: Seq[(TopicPartition, PartitionData)], quota: ReplicaQuota)
}

Proposed Changes

When an RLM class is configured and all the required configs are present, RLM will send a list of topic-partitions and invoke the

...