原文链接：https://cwiki.apache.org/confluence/display/KAFKA/KIP-405%3A+Kafka+Tiered+Storage

背景

Kafka是基础数据重要的组成部分，并且已经得到用户广泛的认可，增长势头迅猛。随着集群规模的增加，越来越多的数据将会被存储在Kafka上，其消息的保留时长、集群的弹缩、性能以及运维等日益变得越来越重要

Kafka采用append-only的日志追加模式，将数据存在在本地磁盘中。消息保留时长通过配置项log.retention来进行控制，既可以设置全局层面的，同时也可以设计某个topic维度的。消息保留时长能否确保数据持久化不丢失，即便是consumer短暂性宕机或不可用，当其成功重启后，只要时间没有超过log.retention，消息依旧能够读取

总的消息的存储量，与topic/partition数量、消息存储速率、消息保留时长相关，一个Kafka的Broker通常在本地磁盘上存储了大量的数据，例如10TB，这种大量本地存储的现象给Kafka的维护带来了巨大挑战

Kafka作为一种长期的存储服务

Kafka的普及率越来越高，也逐渐成为了很多数据的入口。它会将数据持久化下来，因此允许用户进行一些非实时的消费操作。很多用户因为Kafka协议的简单以及消费者API的广泛采用，且允许用户将数据保留很长一段时间，这些特性都有助于Kafka日益成为了数据的source of data（SOT）

目前，Kafka一般会配置一个较短的保留时长（例如3天），然后更老的数据可以通过数据管道拷贝至更具弹缩能力的外部存储（例如HDFS）以便长期使用，结果就是客户端需要建立2种机制去读取数据，相对新的数据读取Kafka，老数据则读取HDFS

Kafka存储的提高，一般是依赖增加更多的Broker节点来实现的，但是这样同样也会导致新增了更多的内存+cpu，相对比可弹缩的外部存储来讲，这样无疑是增加了全局的开销，并且一个很多节点的集群同样增加了运维、部署的难度

Kafka本地存储以及维护的复杂性

当Kafka的一个broker坏掉了，将会用一个新的broker来替代，然后这个新节点必须从其他节点上拉取旧节点的全量数据。同样，当新添加一个broker来横向扩展集群存储时，集群的rebalance会为新节点分配分区，这同样需要复制大量的数据。恢复及rebalance的耗时与kafka broker上的数据量呈正相关。许多多broker的集群（例如100个broker），节点故障是非常常见的情况，在恢复过程中消耗了大量的时间，这使得运维操作变得非常困难

减少每个broker上的存储数据量能够减少recovery及rebalance时间，但是这样操作的话同样需要减少消息的保留时长，这样就使得Kafka可提供的消息回溯时间变得更少

Kafka上云

本地部署的Kafka一般都会使用多个具备硬件SKU的高容量磁盘，从而最大程度提高I/O的吞吐量。而在云上，具有类似SKU的本地磁盘，要么不可用，要么非常昂贵。如果Kafka能够使用容量较小的SKU作为本地存储，那么它就更适合上云

解决方案 - Kafka分层存储

Kafka数据主要以流式方式使用尾部读取来进行消费，提供读取的层，一般都是操作系统的Page Cache，而不是穿透到磁盘。而旧的数据一般是为了回溯或者是因为consumer故障后重启后读取的，而这种情况一般不太常见。

在分层存储方法中，Kafka集群配置有两层存储：本地和远程（local and remote）。本地存储层与当前的Kafka相同，使用Kafka Broker上的本地磁盘来存储日志段。而新的远端存储层则使用一些外部存储，例如HDFS或者S3来实现。不同的存储层使用不同的日志过期时间。当开启远程存储时，本地消息的保留时长将会从几天缩短至几小时，而远端存储的消息保留时长则可能会保留更长的时间，例如几周甚至几个月。当本地日志段发生了滚动 （译者：这里所谓的滚动rolled，可以简单理解为某个日志段写满1G了，即数据已经不会再发生变化了），它可能就会被拷贝至远端存储，当然包含日志段相关的索引文件。这样即便是延迟敏感的数据也能获得高效的消费，因为数据都是尾部读取，且数据都会高概率命中page cache。而那些读取历史消息，或者对消息进行回溯的场景，很有可能数据已经不在本地存储了，那么它们将会去远端存储上读取

此解决方案允许在Kafka集群扩容存储时，将不再依赖于内存和CPU，使Kafka成为一个长期存储的解决方案。同时也减少了每个broker上本地存储的数据量，从而减少了集群recovery及rebalance时需要复制的数据量。broker不需要恢复远程存储层中的日志段，也不存在惰性恢复，而是远程存储层直接提供服务。这样，增加消息保留时长就不需要再扩展Kafka集群的broker数量了，同时消息总体的保留时长还可以更长，不用像当前很多集群部署的策略，需要启动一个单独的管道，将数据从Kafka拷贝至外部存储了

Goals

通过将旧数据存储在外部存储（如HDFS或S3）中，实现了将Kafka的存储扩展到了集群之外，不过Kafka的内部的协议不能有太大的变动。对于那么没有启用分层存储功能的现有用户，Kafka各类行为及操作复杂性决不能改变

Non-Goals

• 分层存储不能取代ETL管道任务。现有的ETL管道继续按原样消费Kafka的数据，尽管Kafka有更长的消息保留时长
• 二级存储不适用于compact类型的topic。即便是将compact类型的topic的配置项remote.storage.enable设置为true，也不能将其类型由delete改为compact
• 二级存储不支持JBOD特性

变更

高层设计

RemoteLogManager (RLM) 是一个新引入的组件：

• 处理leader变更、topic partition删除等回调事件
• 可插拔的存储管理器（即RemoteStorageManager）将处理segments的copy、read、delete事件，且其需要维护远端segments日志段的元数据（它需要知道哪些segments存储在了远端）

RemoteLogManager 是一个内部组件，不会向外暴露API

RemoteStorageManager 本身是一个接口，它定义了远端日志段及索引的生命周期。具体细节下文还会说明，我们将提供一个简单的RSM的实现来帮助大家更好的理解它。而诸如HDFS或者S3的实现应该放在他们产品的仓库中，Apache Kafka自身的仓库不会包含其具体的实现。这个设计与Kafka connnector保持一致

RemoteLogMetadataManager 本身也是个接口，它同样定义了具有强一致语义的远端元数据的生命周期。它的默认实现是一个kafka系统内部的topic，用户如果需要使用其他远程存储介质来存储元数据的话，需要自己去扩展它

RemoteLogManager (RLM)

RLM为leader及follower启动了很多任务，具体解析可见下文

• RLM Leader 职责

• 它会不断地检查非active状态的LogSegments（这些LogSegments中最大的offset需要严格小于LSO，才能进行拷贝），然后将这些LogSegments及索引文件（offset/time/transaction/producer-snapshot）、leader epoch均拷贝至远端存储层
• 提供从远端存储层查询旧数据的服务（当查询的数据在local log存储中没有时）
• 即便是local存储已经不足（或存储的日志已经超时 ？这里存疑），也要先将日志段LogSegments拷贝至远端后，再删除

• RLM Follower 职责

• 通过访问RemoteLogMetdataManager来获取远端存储的log及index数据
• 同时，它也会提供从远端存储层查询旧数据的服务

RLM提供了一个本地的有界缓存（可能是LRU淘汰策略）来存储远端的索引文件，这样可避免频繁的访问远端存储。它们存储在log dir目录下的remote-log-index-cache子目录，这些索引可以像local索引一样使用，用户可以通过设置配置项remote.log.index.file.cache.total.size.mb来设定此缓存的上限

在早期的设计中，还包含了通过远端存储的API拉取LogSegments元数据的章节，（译者：这应该是曾经讨论的某次中间版本）它在HDFS接入时，看起来一切运行的很好。依赖远端存储来维护元数据的问题之一是：整个分层存储是需要强一致性的，它不仅影响元数据，还影响Segments日志段数据本身。其次也要考虑远端存储中存储元数据的耗时，在S3中，frequent LIST APIs导致了巨大的开销

因此需要将远端的数据本身，与元数据进行分离，其对应的管理类分别为RemoteStorageManager、RemoteLogMetadataManager

本地及远端offset约束

以下是leader offset相关描述图

Lx = Local log start offset Lz = Local log end offset Ly = Last stable offset(LSO)

Ry = Remote log end offset Rx = Remote log start offset

Lz >= Ly >= Lx and Ly >= Ry >= Rx

Replica Manager

如果配置了RLM，那么ReplicaManager将调用RLM来分配或删除topic-partition

如果某个Broker从Leader切换为了Follower，而正在此时，RLM正在工作，它正在将某个Segment拷贝至远端，我们这个时候不会直接将其放弃掉，而是会等它完成工作。这个操作可能会导致Segment片段的重复，但是没关系，在远端存储的这些日志过期后，均会删除

Follower Replication

Overview

目前，followers从leaders拉取消息数据，并且尽力尝试追上leader的log-end-offset(LEO)，从而将自己的状态变为in-sync副本。如果需要，follower可能还会截断自己的日志从而与leader的数据保持一致

而在多级存储中，follower同样需要与leader的数据保持一致，follower仅复制leader中已经可用的本地存储的消息。但是他们需要为远端的Segment构建诸如「leader epoch cache」、「producer id snapshot」这些状态，甚至有必要，它们还需要对其进行截断

下面这张图对leader、follower、remote log、metadata storage 4者的关系进行了简明的概述，具体的细节将在下文展开

1. Leader将Segment日志端及AuxiliaryState（含leader epoch及producer-id snapshots）拷贝至远端存储
2. Leader将刚才上传的Segment日志段的元数据发布出去
3. Follower从Leader拉取消息，并遵循一定的规范，这个规范在下文具体说明
4. Follower等待Leader将元数据放入RemoteLogSegmentMetadataTopic后将其拉取下来
5. Follower抓取相应的远端存储的元数据，并构建状态AuxiliaryState

Follower拉取消息协议细节

Leader epoch概念的引入，是为了解决在KIP-101及KIP-279中提到的leader切换的场景中，可能存储日志差异的问题。它（Leader epoch）是partition下的一个单调递增的整数，每当leader进行了切换，那么这个值将会累加，并且它也会存储在消息的message batch中

Leader epoch文件存在于每个broker的每个partition中，然后所有状态是in-sync的副本需要保证其有同样的leader epoch历史信息，以及相同的日志数据

Leader epoch的作用：

• 决定日志截断（KIP-101）
• 保证副本间的一致性（KIP-279）
• 在发生截断后，重置消费位点（KIP-320)

在使用远端存储时，我们应该像使用本地存储一样，来处理日志及leader epoch

目前，纯本地存储的场景，follower从leader拉取消息后，通过读取message batch来构建AuxiliaryState状态。

而在多级存储中，follower需要读取leader构建出来的AuxiliaryState，从而获取起始offset及leader epoch。然后follower将会从这个起始offset开始拉取数据。这个起始offset可能是「local-log-start-offset」或「last-tiered-offset」。local-log-start-offset是本地存储的开始offset；last-tiered-offset是已经拷贝至远端存储的最大offset。我们来讨论下使用这两者的利弊

last-tiered-offset

• 用这个策略明显的好处就是follower能否非常快的追上leader，因为follower只需要同步那些存在于leader本地存储中，且还没来得及放在远端的日志段
• 而这样做的一个缺点是，follower相对于leader缺少很多本地日志段，当这个follower成为leader后，其他follower将会根据新leader的log-start-offset来截断它们的日志段

local-log-start-offset

• 在发生leader切换时，将会保留本地日志
• follower追赶leader，这将会花费较长的时间，当为某个partition新增一个全新follower时，就命中了这个case

基于上述原因，我们更倾向使用「local-log-start-offset」

在多层存储中，当follower来拉取数据时，leader只会返回在本地存储中存在的数据。那些已经存在在远端，且本地已经没有的日志段，follower是不会进行拉取复制的。根据「local-log-start-offset」机制，如果有必要的话，follower可能会截断自己的日志

当一个follower从leader拉取一个leader的本地存储已经不存在的offset时，leader将会发送一个错误码OFFSET_MOVED_TO_TIERED_STORAGE，然后follower将会重新从leader获取「local-log-start-offset」及「leader eopch」。follower收到leader的local-log-start-offset后，需要基于这个offset构建远端日志段的AuxiliaryState，「译者：此处注意，在纯local存储的模式下，follower是通过拉取leader的全量日志，并且在这个拉取过程中，逐步构建并维护leader-epoch-checkpoint文件的。而在多层存储的环境中，因为follower不再需要从leader处拉取全量日志，但是follower自身的leader-epoch-checkpoint文件还需要全量维护，因此就需要额外花精力去构建这个文件，否则当这个follower成为leader后，leader-epoch-checkpoint文件的部分缺失，会使其无法做出正常的判断」这个AuxiliaryState其实就是leader的「leader eopch」及「producer-snapshot-ids」。可以通过两种方式来实现：

• 引入一个新的协议，专门从leader中拉取这个AuxiliaryState
• 从远端存储中获取这个AuxiliaryState

这里更推荐后者，因为本身远端存储已经保留了这个字段，且不需要在于leader的交互中引入新的协议

获取目标offset的之前的日志段的AuxiliaryState状态需要以下2个步骤：

• 需要拉取远端日志段的元数据
• 需要在相应日志段中拉取诸如leader epoch的记录

当将一个日志段（segment）搬移至远端存储后，leader broker同时需要将「leader epoch sequence」以及「producer id snapshot」追加到segment所在的目录下。这些数据将会帮助follower来构建自己的「leader epoch sequence」以及「producer id snapshot」

因此，我们需要为这个副本引入一个相对应的新状态，可以将其定义为BuildingRemoteLogAuxState。follower的拉取线程就如同切换Fetching或Truncating states状态一样，在每次执行时，都需要判断一下，需要切换至哪个状态

当一个follower尝试拉取一个已经不在leader local 存储的offset时，会收到leader返回的OffsetMovedToRemoteStorage错误，如果follower收到了这个状态，将会：

1. 通过调用API ListOffset来获取leader的Earliest Local Offset (ELO) 以及 leader epoch (ELO-LE) 译者：注意，ListOffset这个API将会发生改变，其返回的出参中将会携带这些信息
2. 截断自己的本地日志以及AuxiliaryState
3. 从Fetching状态切换至BuildingRemoteLogAux状态

处于BuildingRemoteLogAux状态时，follower可以在以下两个方案中二选一：

• 方案1：

• 通过不断反复调用FetchEarliestOffsetFromLeader API，从而获取ELO-LE至leader中最早的leader epoch，然后构建follower本地的leader epoch。当远端存储上有很多任leader切换时，这个方案可能并不会很高效。不过这个方案的好处是，获取leader epoch的操作完全在kafka内部，当远端存储出现短暂不可用时，follower仍然可以追赶leader并进入ISR

• 方案2：

• RLMM（RemoteLogMetadataManager）等待远端的元数据，直到等到某个segment包含了ELO-LE
• 抓取远端存储的leader epoch以及producer snapshot（使用远端fetcher线程）译者：多层存储引入的工作线程
• 获取远端存储的leader epoch数据后，截取 [LSO, ELO] 部分，然后构建follower自己的cache

在构建完follower自己的leader epoch后，follower状态转换为Fetching，然后继续从leader的ELO开始拉取数据。我们更倾向使用方案2，即从远端存储来获取所需数据

Follower fetch 场景(包含日志截断的场景)

让我们讨论一下follower在尝试从leader复制并从远程存储构建AuxiliaryState状态时可能遇到的几种情况

名词定义：

OMTS : OffsetMovedToTieredStorage 译者：offset已经不在leader中，通常是一个错误

ELO : Earliest-Local-Offset 译者：local存储中最早的offset

LE-x : Leader Epoch x, 译者：leader epoch，不赘述

HW : High Watermark 译者：高水位，kafka发明的词，不赘述

seg-a-b: a remote segment with first-offset = a and last-offset = b 译者：远端存储的某个segment日志段，它的offse的区间

LE-x, y : A leader epoch sequence entry indicates leader-epoch x starts from offset y 译者：leader epoch的某个区间

场景1：全新follower

现在假设某个全新的broker刚被加入集群，然后将其指派为某个partition的follower replica，这个follower肯定是没有任何本地存储数据的。它将会从offset为0的位置开始从leader抓取数据，如果offset为0的位点在leader中不存在的话，follower将会收到错误OFFSET_MOVED_TO_TIERED_STORAGE，然后follower将会给leader发送ListOffset API，并且在入参中携带参数timestamp = EARLIEST_LOCAL_TIMESTAMP，接着会收到leader返回的ELO（Earliest-Local-Offset） 译者：多层存储需要修改ListOffset协议

follower需要等待这个offset（leader的ELO）的返回，然后构建AuxiliaryState状态，然后才能从leader拉取数据 译者：又强调了构建复核状态的必要

步骤1：

抓取远端segment信息，然后构建leader epoch

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 6: msg 6 LE-2 7: msg 7 LE-3 (HW) leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	1. Fetch LE-1, 0 2. Receives OMTS 3. Receives ELO 3, LE-1 4. Fetch remote segment info and build local leader epoch sequence until ELO leader_epochs LE-0, 0 LE-1, 3	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-5, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 5	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-3-5, uuid-2 segment epochs LE-1, 3 LE-2, 5

步骤2：

继续从leader拉取数据

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 6: msg 6 LE-2 7: msg 7 LE-3 (HW) leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	Fetch from ELO to HW 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 6: msg 6 LE-2 7: msg 7 LE-3 (HW) leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-5, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 5	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-3-5, uuid-2 segment epochs LE-1, 3 LE-2, 5

场景2：out-of-sync follower catching up

一个follower正在尝试追赶leader，然后leader对应的日志段segment已经转移至了远端存储。我们以目标日志段是否在本地存储来分为2种情况来讨论

• 本地segment存在，而且本地最新的offset要比leader的ELO大

• 这种场景，本地存储已有，follower跟常规方式一样进行拉取即可

• 本地segment不存在，或者最新的offset要比leader的ELO小

• 这种场景，本地的日志段可能因为日志过期已经删除，或者是因为follower已经离线了很长一段时间。然后follower拉取数据时，将会收到OFFSET_MOVED_TO_TIERED_STORAGE错误，然后follower将不得不截断自己所有的本地日志，因为这些数据在leader已经标记为过期

步骤1：

out-of-sync follower (broker B) 本地的offset存储到了3

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 6: msg 6 LE-2 7: msg 7 LE-3 8: msg 8 LE-3 9: msg 9 LE-3 (HW) leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 leader_epochs LE-0, 0 LE-1, 3 1. Because the latest leader epoch in the local storage (LE-1) does not equal the current leader epoch (LE-3). The follower starts from the Truncating state. 2. fetchLeaderEpochEndOffsets(LE-1) returns 5, which is larger than the latest local offset. With the existing truncation logic, the local log is not truncated and it moves to Fetching state.	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-5, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 5	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-3-5, uuid-2 segment epochs LE-1, 3 LE-2, 5

步骤2：

leader的本地日志段因为数据过期而已经删除，然后follower开始尝试追上leader

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
9: msg 9 LE-3 10: msg 10 LE-3 11: msg 11 LE-3 (HW) [segments till offset 8 were deleted] leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 leader_epochs LE-0, 0 LE-1, 3 <Fetch State> 1. Fetch from leader LE-1, 4 2. Receives OMTS, truncate local segments. 3. Fetch ELO, Receives ELO 9, LE-3 and moves to BuildingRemoteLogAux state	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-5, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 5 Seg 6-8, uuid-3, LE-3 log: 6: msg 6 LE-2 7: msg 7 LE-3 8: msg 8 LE-3 epochs: LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-3-5, uuid-2 segment epochs LE-1, 3 LE-2, 5 seg-6-8, uuid-3 segment epochs LE-2, 5 LE-3, 7

步骤3：

删除本地数据后，将会转换为场景1一样的case

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
9: msg 9 LE-3 10: msg 10 LE-3 11: msg 11 LE-3 (HW) [segments till offset 8 were deleted] leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	1. follower rebuilds leader epoch sequence up to LE-3 using remote segment metadata and remote data leader_epochs LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7 2. follower continue fetching from the leader from ELO (9, LE-3) 9: msg 9 LE-3 10: msg 10 LE-3 11: msg 11 LE-3 (HW)	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-5, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 5 Seg 6-8, uuid-3, LE-3 log: 6: msg 6 LE-2 7: msg 7 LE-3 8: msg 8 LE-3 epochs: LE-0, 0 LE-1, 3 LE-2, 5 LE-3, 7	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-3-5, uuid-2 segment epochs LE-1, 3 LE-2, 5 seg-6-8, uuid-3 segment epochs LE-2, 5 LE-3, 7

场景3：Multiple hard failures

步骤1：

Broker A已经将第一个segment转移至了远端存储

Broker A (Leader)	Broker B	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 (HW) leader_epochs LE-0, 0	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 (HW) leader_epochs LE-0, 0	seg-0-1: log: 0: msg 0 LE-0 1: msg 1 LE-0 epoch: LE-0, 0	seg-0-1, uuid-1 segment epochs LE-0, 0

步骤2：

Broker A与Broker B同时崩溃，在Broker B上的一些消息（msg1及msg2）还没有及时刷盘，然后丢失了。在这种场景下，我们是可以接受数据丢失的，但是我们需要保证与KIP-101具有相同的语义

Broker A (stopped)	Broker B (Leader)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 (HW) leader_epochs LE-0, 0	0: msg 0 LE-0 (HW) 1: msg 3 LE-1 leader_epochs LE-0, 0 LE-1, 1	seg-0-1: log: 0: msg 0 LE-0 1: msg 1 LE-0 epoch: LE-0, 0	seg-0-1, uuid-1 segment epochs LE-0, 0

在Broker B重启后，B丢失了msg1及msg2，然后B变成了leader，这时收到了一条新的消息 msg3 （LE-1, offset 1）

（注意：严格来讲，这个应该不属于unclean-leader-election，因为B并没有从ISR中移除，因为发生问题时，A B同时崩溃掉了）

步骤3：

重启后，broker A截断offset 1、2，然后这时收到了新的数据（LE-1, offset 1）

Broker A (follower)	Broker B (Leader)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 1: msg 3 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	0: msg 0 LE-0 1: msg 3 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	seg-0-1: log: 0: msg 0 LE-0 1: msg 1 LE-0 epoch: LE-0, 0	seg-0-1, uuid-1 segment epochs LE-0, 0

步骤4：

Broker A (follower)	Broker B (Leader)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 3 LE-1 2: msg 4 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	0: msg 0 LE-0 1: msg 3 LE-1 2: msg 4 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	seg-0-1: log: 0: msg 0 LE-0 1: msg 1 LE-0 epoch: LE-0, 0 seg-1-1 log: 1: msg 3 LE-1 epoch: LE-0, 0 LE-1, 1	seg-0-1, uuid-1 segment epochs LE-0, 0 seg-1-1, uuid-2 segment epochs LE-1, 1

收到一个新消息msg 4，然后B broker上的第二个segment(seg-1-1)传输到了远端存储

考虑在两个broker上删除offset为2的本地segment：

• consumer拉取offset 0, LEO-0。根据本地leader epoch 缓存，offset 0 LE-0是有效的，因此broker会基于segment 0-1 返回msg 0
• consumer拉取 offset 1，没有携带leader epoch信息。根据本地leader epoch缓存，offset 1是属于 LE-1的。因此broker将会基于segment 1-1 返回 msg 3，而不是seg-0-1的LE-0 的offset 1
• consumer拉取 LE-0的offset 2将会被拒绝
• consumer拉取 LE-1的offset 1将会收到远端日志段segmeng 1-1 对应的msg 3

场景4：unclean leader election including truncation

步骤1：

Broker A (Leader)	Broker B (out-of-sync)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 (HW) leader_epochs LE-0, 0	0: msg 0 LE-0 (HW) leader_epochs LE-0, 0	seg 0-2: log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epoch: LE-0, 0	seg-0-2, uuid-1 segment epochs LE-0, 0

步骤2：

Broker A (Stopped)	Broker B (Leader)	Remote Storage	RL metadata storage
	0: msg 0 LE-0 1: msg 4 LE-1 2: msg 5 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	seg 0-2: log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epoch: LE-0, 0 seg 0-1: 0: msg 0 LE-0 1: msg 4 LE-1 epoch: LE-0, 0 LE-1, 1	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-0-1, uuid-2 segment epochs LE-0, 0 LE-1, 1

Broker A停止了，然后一个 out-of-sync的副本（broker B）成为了新的leader。基于unclean-leader-election策略，是允许数据丢失的，但是我们需要保证已有的Kafka的行为没有发生变化

我们假设 min.in_sync = 1

在HW变为2后，Broker B将其本地的日志段seg-0-1搬运至远端

步骤3：

Broker A (Stopped)	Broker B (Leader)	Remote Storage	RL metadata storage
	2: msg 5 LE-1 (HW) leader_epochs LE-0, 0 LE-1, 1	seg 0-2: log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epoch: LE-0, 0 seg 0-1: 0: msg 0 LE-0 1: msg 4 LE-1 epoch: LE-0, 0 LE-1, 1	seg-0-2, uuid-1 segment epochs LE-0, 0 seg-0-1, uuid-2 segment epochs LE-0, 0 LE-1, 1

Broker B上的第一个本地日志段已经过期

consumer拉取offset 0 LE-0收到msg 0。这个消息既可以由远端日志段seg-0-2提供，也可以由seg-0-1提供

consumer拉取offset 1，broker发现offset 1属于leader eopch 1，因此它返回msg 4 而不是msg 1

consumer拉取offset 1 LE-1，收到远端日志段segmeng 0-1的msg 4

consumer拉取offset 2 LE-0将被拒

场景5：log divergence in remote storage - unclean leader election

步骤1：

Broker A (Leader)	Broker B	Remote Storage	Remote Segment Metadata
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 4: msg 4 LE-0 (HW) leader_epochs LE-0, 0 broker A shipped one segment to remote storage	0: msg 0 LE-0 1: msg 1 LE-0 leader_epochs LE-0, 0 broker B is out-of-sync	seg-0-3 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 epoch: LE0, 0	seg-0-3, uuid1 segment epochs LE-0, 0

步骤2：

在broker A宕机后，out-of-sync的broker B变成了新leader（unclean leader election）

Broker A (stopped)	Broker B (Leader)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 4: msg 4 LE-0 leader_epochs LE-0, 0	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 4 LE-1 3: msg 5 LE-1 4: msg 6 LE-1 leader_epochs LE-0, 0 LE-1, 2 After becoming the new leader, B received several new messages, and shipped one segment to remote storage.	seg-0-3 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 epoch: LE-0, 0 Seg-0-3 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 4 LE-1 3: msg 5 LE-1 epoch: LE-0, 0 LE-1, 2	seg-0-3, uuid1 segment epochs LE-0, 0 seg-0-3, uuid2 segment epochs LE-0, 0 LE-1, 2

步骤3：

Broker B宕机，Broker A重启，但是不知道LE-1的存在（另一个unclean leader election）

Broker A (Leader)	Broker B (stopped)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 4: msg 4 LE-0 5: msg 7 LE-2 6: msg 8 LE-2 leader_epochs LE-0, 0 LE-2, 5 1. Broker A receives two new messages in LE-2 2. Broker A ships seg-4-5 to remote storage	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 4 LE-1 3: msg 5 LE-1 4: msg 6 LE-1 leader_epochs LE-0, 0 LE-1, 2	seg-0-3 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 epoch: LE-0, 0 seg-0-3 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 4 LE-1 3: msg 5 LE-1 epoch: LE-0, 0 LE-1, 2 seg-4-5 epoch: LE-0, 0 LE-2, 5	seg-0-3, uuid1 segment epochs LE-0, 0 seg-0-3, uuid2 segment epochs LE-0, 0 LE-1, 2 seg-4-5, uuid3 segment epochs LE-0, 0 LE-2, 5

步骤4：

Broker B重启后丢失了所有本地数据

Broker A (Leader)	Broker B (started, follower)	Remote Storage	RL metadata storage
6: msg 8 LE-2 leader_epochs LE-0, 0 LE-2, 5	1. Broker B fetches offset 0, and receives OMTS error. 2. Broker B receives ELO=6, LE-2 3. in BuildingRemoteLogAux state, broker B finds seg-4-5 has LE-2. So, it builds local LE cache from seg-4-5: leader_epochs LE-0, 0 LE-2, 5 4. Broker B continue fetching from local messages from ELO 6, LE-2 5. Broker B joins ISR	seg-0-3 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-0 epoch: LE-0, 0 seg-0-3 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 4 LE-1 3: msg 5 LE-1 epoch: LE-0, 0 LE-1, 2 seg-4-5 epoch: LE-0, 0 LE-2, 5	seg-0-3, uuid1 segment epochs LE-0, 0 seg-0-3, uuid2 segment epochs LE-0, 0 LE-1, 2 seg-4-5, uuid3 segment epochs LE-0, 0 LE-2, 5

consumer从broker B拉取offset 3，LE-1 被拒

consumer从broker B拉取offet 2，将会收到 msg 2

Follower转换为Leader

controller会根据某个follower的相关配置信息来判断，follower是可以被转换为leader的。当一个follower成为leader后，它需要判断从哪个日志段开始将数据拷贝至远端存储，通过遍历leader epoch的历史信息，直至到最近一次的leader epoch，然后找到已经拷贝至远端存储的最大offset。如果找不到对应的entry，那么就从前一个leader epoch中寻找，如果一直到最早的leader epoch仍然没有找到，那么就从最早的epoch开始拷贝

步骤1：

Broker A (Leader)	Broker B (Follower)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-1 6: msg 6 LE-2 (HW) 7: msg 7 LE-2 8: msg 8 LE-2 leader_epochs LE-0, 0 LE-1, 3 LE-2, 6	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-1 6: msg 6 LE-2 (HW) leader_epochs LE-0, 0 LE-1, 3 LE-2, 6	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg 3-4, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 epochs: LE-0, 0 LE-1, 3	seg-0-2, uuid-1 Segment epochs LE-0, 0 seg-3-4, uuid-2 Segment epochs LE-1, 3

步骤2：

此时Broker A宕机了，然后Broker B成为了新的leader，它现在的leader epoch是3，因此需要在远端元数据中寻找leader epoch 为2的offset的最大值，如果不存在，那么就需要找leader epoch 1 的数据，以此类推。在本例中，它找到了epoch为1，offset=4的记录，因此它需要拷贝包含了offset 5的日志段segment，因此，从seg-4-6日志段开始拷贝 译者：此处存疑，感觉可能是作者写错了，因为只有当某个日志段segment成为了非active状态时，才能从本地拷贝至远端，也就是如果seg 3-4已经拷贝到了远端，那么就表明seg 3-4只包含2条消息，所以应该是从seg-5-6拷贝才对

Broker A (Stopped)	Broker B (Leader)	Remote Storage	RL metadata storage
0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-1 6: msg 6 LE-2 (HW) 7: msg 7 LE-2 8: msg 8 LE-2 leader_epochs LE-0, 0 LE-1, 3 LE-2, 6	0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 3: msg 3 LE-1 4: msg 4 LE-1 5: msg 5 LE-1 6: msg 6 LE-2 (HW) 7: msg 8 LE-3 leader_epochs LE-0, 0 LE-1, 3 LE-2, 6 LE-3, 7	seg-0-2, uuid-1 log: 0: msg 0 LE-0 1: msg 1 LE-0 2: msg 2 LE-0 epochs: LE-0, 0 seg-3-4, uuid-2 log: 3: msg 3 LE-1 4: msg 4 LE-1 epochs: LE-0, 0 LE-1, 3 Seg-4-6, uuid-3 4: msg 4 LE-1 5: msg 5 LE-1 6: msg 6 LE-2 epochs: LE-0, 0 LE-1, 3 LE-2, 6	seg-0-2, uuid-1 Segment epochs LE-0, 0 seg-3-4, uuid-2 Segment epochs LE-1, 3 seg-4-6, uuid-3 Segment epochs LE-1, 3 LE-2, 6

事务支持

RemoteLogManager拷贝的数据均是小于LSO（last-stable-offset）的。follower可以返回那些事务的取消消息

Consumer抓取请求

任何消费请求，ReplicaManager都会去本地存储查询，如果本地存储返回OffsetOutOfRange异常，那么将会去远端存储查询，如果远端存储依旧没有对应的数据，那么kafka将会扔出TIERED_STORAGE_NOT_AVAILABLE的错误

其他API

DeleteRecords

ListOffsets

LeaderAndIsr

Stopreplica

OffsetForLeaderEpoch

LogStartOffset

RLM/RSM tasks and thread pools

远端存储（例如HDFS/S3/GCP）一般相对比本地来说，都具有高I/O延迟、低性能的特点

当远端存储变得临时不可用（长达数小时）或者延迟变高（长达几分钟），Kafka应该依旧能够正常运转。所有Kafka的操作（produce/consume local data/create topic/etc.）不应该被远端存储所影响。当远端存储不可用或超时时，consumer消费远端数据，应该收到一个响应错误

为了实现这点，我们必须在专用线程池中处理远端存储的操作，而不是在Kafka I/O线程和fetcher线程中

Remote Log Manager (RLM) Thread Pool

RLM维护其管理的topic-partition的list，当topic-partition添加/删除时，这个list将被Kafka的I/O线程更新维护。这个list中每个topic-partition都被分配了一个计划处理的时间，某个topic-partition到达处理时间后，RLM线程池就会将其发起调度

当一个新的topic-partition被指派给broker后，topic-partition就会加入这个list，于此同时，这个partition的执行时间将被初始化为0（scheduled processing time = 0），也就意味着这个partition将会被立即调度执行，然后从远端存储中检索查询信息

当一个partition执行结束后，它的调度时间将会被置为（now() + remote.log.manager.task.interval.ms），配置项remote.log.manager.task.interval.ms可以在broker.conf中进行配置

如果某次partition由于远端存储错误导致执行失败，它会依据规避重试算法进行重试，重试初始配置项为remote.log.manager.task.retry.interval.ms，最大重试时间为remote.log.manager.task.retry.backoff.max.ms，以及抖动配置remote.log.manager.task.retry.jitter

当一个partition在broker上注销时，如果线程池没有调度这个partition，那么直接将其从list中移除，否则将此partition标记位deleted，然后会在当前调度结束后直接将其删除

线程池中的线程，在同一时刻，只会处理一个partition，遵循如下步骤：

• 拷贝日志段segment至远端存储（leader）

• 拷贝的日志段有如下特征

• 非active 译者：拷贝的所有segment都不会再产生新数据了
• 拷贝的offset range，没有被远端存储段完全覆盖
• last offset < last-stable-offset 译者：拷贝LSO以下的数据

• 如果日志段都已经就绪，那么它们将会逐个拷贝至远端存储，按照时间升序，依次拷贝。它为每个segment生成一个普遍唯一的RemoteLogSegmentId，它调用RLMM.putRemoteLogSementData（RemoteLogSegentMetadata RemoteLogSegmentMetadata），并在RSM上调用copyLogSegment（RemoteLogSegmentMetadata remoteLogSegmentMetadata，LogSegmentData LogSegmentData）。如果成功，将会调用RLMM.putRemoteLogSegmentData，入参为update state

• 处理过期的远端日志段segment（leader）

• RLM leader会根据远端超时参数配置，来计算判断segment是否应该删除。并且它会实时维护RLMM的最早的offset，它会提供包含segment ids的远端日志段的全量列表（实现类为RemoteStorageManager），同样它也会删除对应的元数据（实现类为RemoteLogMetadataManager）

Remote Storage Fetcher Thread Pool

当处理consumer拉取请求时，如果请求的offset落在了远端存储上，那么这个请求将会被加入至RemoteFetchPurgatory（方便处理超时），而RemoteFetchPurgatory是kafka原始定义的延迟处理器kafka.server.DelayedOperationPurgatory的一种实例，与现有的producer/fetch回调类似。与此同时，这个请求将会被放入命名为“remote storage fetcher thread pool”的队列中

在这个线程池中的每个线程在同一时刻，只会处理一个fetch操作，而这个从远端拉取数据的线程：

1. 从RLMM中找出相应的RemoteLogSegmentId，从offset索引中找出startPosition和endPosition
2. 尝试从方法RSM.fetchLogSegmentData(RemoteLogSegmentMetadata remoteLogSegmentMetadata, Long startPosition, Long endPosition)构建拉取的数据记录

1. 如果成功，RemoteFetchPurgatory将会被通知，然后返回数据给client端
2. 如果远端日志已经被删了，RemoteFetchPurgatory也将会被通知到，然后给client端返回一个错误
3. 如果远端操作失败（例如远端存储短暂性不可用），拉取操作将会自动重试，直至consumer的fetch操作超时

Remote Log Metadata State transitions

COPY_SEGMENT_STARTED - 这个状态表明segment正在向远端拷贝，但是还没有完成

COPY_SEGMENT_FINISHED - 这个状态表明segment已经向远端拷贝完成

leader broker将segment拷贝至远端，然后将元数据状态修改为COPY_SEGMENT_STARTED，而一旦拷贝完成，则将状态修改为COPY_SEGMENT_FINISHED

DELETE_SEGMENT_STARTED - 这个状态表明segment正在进行删除操作，但是还没完成

DELETE_SEGMENT_FINISHED - 这个状态表明segment已经成功删除完毕

当远端存储的消息过时后，Leader partition负责发布以上两个状态。 Remote Partition Removers也会发布这两个事件

DELETE_PARTITION_MARKED - 当某个partition被controller删除时，将会被修改为这个状态。 也就意味着，所有拥有这个partition远端日志删除的操作者，都有权力去删除这个远端日志

DELETE_PARTITION_STARTED - 这个状态表明这个partition启动了删除日志操作，但是还未完成

DELETE_PARTITION_FINISHED - 这个状态表明这个partition完成了删除日志操作

Remote Partition Removers发布这两个状态

当一个partition被删，controller在RLMM中修改其状态为DELETE_PARTITION_MARKED，预期RLMM能有一个清理远端日志的机制

RemoteLogMetadataManager implemented with an internal topic

远端存储的segment的元数据存放在一个内部的非compact的topic中，topic name为__remote_log_metadata，这个topic默认初始化了50个分区，用户可以通过修改配置参数来修改这个选项

在本设计中， RemoteLogMetadataManager(RLMM)的职责是存储及拉取远端的元数据，它提供如下功能

• 为一个partition的日志段segment提供存储元数据的功能 译者：一个用户定义的partition一般都会有多个日志段
• 根据一个leader epoch + offset抓取远端元数据 译者：给定leader epoch及offet来定位元数据信息
• 通过读取这个元数据topic，来构建partition的元数据缓存

RemoteLogMetadataManager(RLMM)可能由以下组件组成

• Cache - 缓存
• Producer - 生产者
• Consumer - 消费者

某个用户创建的topic的元数据存储在 __remote_log_metadata 的partition为：

Utils.toPositive(Utils.murmur2(tp.toString().getBytes(StandardCharsets.UTF_8))) % no_of_remote_log_metadata_topic_partitions

不论是当前broker是__remote_log_metadata对应partition的leader还是follower，RLMM均会注册，当然这些partition包含远端元数据信息

RLMM通过订阅__remote_log_metadata对应的partition，维护了一套自己的缓存。无论何时，当一个partition被指派到了一个新broker上，并且这个新broker的RLMM并没有订阅这个partition的元数据信息，那么这时新broker上的RLMM就一定会去订阅对应的partition远端元数据，并将其维护在自己的cache中。因此，在最坏的场景中，某个broker的RLMM可能订阅了topic上大部分的partition。在原始版本中，无论RLMM何时启动，我们都将会有一个基于文件的缓存，这个缓存保存了这个实例已经消费的全部消息。每个partition文件都有这样一个独立的文件。它将会帮助我们看到已经读取的数据，从而定位到commit offset，commit offset也可以存储在本地文件中，当我们重启broker时，也就避免了重新读取消息

RLMM segment的存储开销

Topic partition's topic-id : uuid : 2 longs.

remoteLogSegmentId : uuid : 2 longs.

remoteLogSegmentMetadata : 5 longs + 1 int +1 byte + ~3 epochs(approx avg)

It has leader epochs in-memory which will be much less.

On avg: 10 longs : 10 * 8 = 80 *(other overhead 1.25) = 100 bytes

When a segment is rolled on a broker per sec. // 译者：这里假设每秒生成1G文件

retention as 30days : 60*60*24*30 ~ 2.6MM

2.6MM segments would take ~ 260MB. (This is 1% in our production env)

这个开销可能并没有那么大，因为随意一个borker就会使用几个GB的内存

我们同样可以设置一个类似懒加载、有界缓存的模式来控制内存消耗。需要的话，我们甚至基于操作文件来维护都行

Message Format

RLMM实例会发布key为null、value格式如下的消息

type : 相对应的value类型，就像schema中定义的apikey，类型是字节byte
version : schema中定义的version，类型是字节byte
data : kafka消息协议格式，以下给出schema定义
在data被序列化之前，type跟version其实已经定义好了，通过添加一个新的version，很容易扩展Schema，同样也可以比较容易地添加一个新的type及其version

Schema

{
    "apiKey": 0,
    "type": "data",
    "name": "RemoteLogSegmentMetadataRecord",
    "validVersions": "0",
    "flexibleVersions": "none",
    "fields": [
    {
        "name": "RemoteLogSegmentId",
        "type": "RemoteLogSegmentIdEntry",
        "versions": "0+",
        "about": "Unique representation of the remote log segment",
        "fields": [
        {
            "name": "TopicIdPartition",
            "type": "TopicIdPartitionEntry",
            "versions": "0+",
            "about": "Represents unique topic partition",
            "fields": [
            {
              "name": "Name",
              "type": "string",
              "versions": "0+",
              "about": "Topic name"
            },
            {
              "name": "Id",
              "type": "uuid",
              "versions": "0+",
              "about": "Unique identifier of the topic"
            },
            {
              "name": "Partition",
              "type": "int32",
              "versions": "0+",
              "about": "Partition number"
            }
          ]
        },
        {
          "name": "Id",
          "type": "uuid",
          "versions": "0+",
          "about": "Unique identifier of the remote log segment"
        }
      ]
    },
    {
      "name": "StartOffset",
      "type": "int64",
      "versions": "0+",
      "about": "Start offset  of the segment."
    },
    {
      "name": "EndOffset",
      "type": "int64",
      "versions": "0+",
      "about": "End offset  of the segment."
    },
    {
      "name": "LeaderEpoch",
      "type": "int32",
      "versions": "0+",
      "about": "Leader epoch from which this segment instance is created or updated"
    },
    {
      "name": "MaxTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Maximum timestamp with in this segment."
    },
    {
      "name": "EventTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Event timestamp of this segment."
    },
    {
      "name": "SegmentLeaderEpochs",
      "type": "[]SegmentLeaderEpochEntry",
      "versions": "0+",
      "about": "Leader epoch cache.",
      "fields": [
        {
          "name": "LeaderEpoch",
          "type": "int32",
          "versions": "0+",
          "about": "Leader epoch"
        },
        {
          "name": "Offset",
          "type": "int64",
          "versions": "0+",
          "about": "Start offset for the leader epoch"
        }
      ]
    },
    {
      "name": "SegmentSizeInBytes",
      "type": "int32",
      "versions": "0+",
      "about": "Segment size in bytes"
    },
    {
      "name": "RemoteLogSegmentState",
      "type": "int8",
      "versions": "0+",
      "about": "State of the remote log segment"
    }
  ]
}
 
 
{
  "apiKey": 1,
  "type": "data",
  "name": "RemoteLogSegmentMetadataRecordUpdate",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    {
      "name": "RemoteLogSegmentId",
      "type": "RemoteLogSegmentIdEntry",
      "versions": "0+",
      "about": "Unique representation of the remote log segment",
      "fields": [
        {
          "name": "TopicIdPartition",
          "type": "TopicIdPartitionEntry",
          "versions": "0+",
          "about": "Represents unique topic partition",
          "fields": [
            {
              "name": "Name",
              "type": "string",
              "versions": "0+",
              "about": "Topic name"
            },
            {
              "name": "Id",
              "type": "uuid",
              "versions": "0+",
              "about": "Unique identifier of the topic"
            },
            {
              "name": "Partition",
              "type": "int32",
              "versions": "0+",
              "about": "Partition number"
            }
          ]
        },
        {
          "name": "Id",
          "type": "uuid",
          "versions": "0+",
          "about": "Unique identifier of the remote log segment"
        }
      ]
    },
    {
      "name": "LeaderEpoch",
      "type": "int32",
      "versions": "0+",
      "about": "Leader epoch from which this segment instance is created or updated"
    },
    {
      "name": "EventTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Event timestamp of this segment."
    },
    {
      "name": "RemoteLogSegmentState",
      "type": "int8",
      "versions": "0+",
      "about": "State of the remote segment"
    }
  ]
}
 
 
 
{
  "apiKey": 2,
  "type": "data",
  "name": "RemotePartitionDeleteMetadataRecord",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    {
      "name": "TopicIdPartition",
      "type": "TopicIdPartitionEntry",
      "versions": "0+",
      "about": "Represents unique topic partition",
      "fields": [
        {
          "name": "Name",
          "type": "string",
          "versions": "0+",
          "about": "Topic name"
        },
        {
          "name": "Id",
          "type": "uuid",
          "versions": "0+",
          "about": "Unique identifier of the topic"
        },
        {
          "name": "Partition",
          "type": "int32",
          "versions": "0+",
          "about": "Partition number"
        }
      ]
    },
    {
      "name": "Epoch",
      "type": "int32",
      "versions": "0+",
      "about": "Epoch (controller or leader) from which this event is created. DELETE_PARTITION_MARKED is sent by the controller. DELETE_PARTITION_STARTED and DELETE_PARTITION_FINISHED are sent by remote log metadata topic partition leader."
    },
    {
      "name": "EventTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Event timestamp of this segment."
    },
    {
      "name": "RemotePartitionDeleteState",
      "type": "int8",
      "versions": "0+",
      "about": "Deletion state of the remote partition"
    }
  ]
}
 
package org.apache.kafka.server.log.remote.storage;
...
/**
 * It indicates the deletion state of the remote topic partition. This will be based on the action executed on this
 * partition by the remote log service implementation.
 */
public enum RemotePartitionDeleteState {
 
    /**
     * This is used when a topic/partition is determined to be deleted by controller.
     * This partition is marked for delete by controller. That means, all its remote log segments are eligible for
     * deletion so that remote partition removers can start deleting them.
     */
    DELETE_PARTITION_MARKED((byte) 0),
 
    /**
     * This state indicates that the partition deletion is started but not yet finished.
     */
    DELETE_PARTITION_STARTED((byte) 1),
 
    /**
     * This state indicates that the partition is deleted successfully.
     */
    DELETE_PARTITION_FINISHED((byte) 2);
...
}
 
 
package org.apache.kafka.server.log.remote.storage;
...
/**
 * It indicates the state of the remote log segment or partition. This will be based on the action executed on this
 * segment or partition by the remote log service implementation.
 * <p>
 */
public enum RemoteLogSegmentState {
 
    /**
     * This state indicates that the segment copying to remote storage is started but not yet finished.
     */
    COPY_SEGMENT_STARTED((byte) 0),
 
    /**
     * This state indicates that the segment copying to remote storage is finished.
     */
    COPY_SEGMENT_FINISHED((byte) 1),
 
    /**
     * This state indicates that the segment deletion is started but not yet finished.
     */
    DELETE_SEGMENT_STARTED((byte) 2),
 
    /**
     * This state indicates that the segment is deleted successfully.
     */
    DELETE_SEGMENT_FINISHED((byte) 3),
...
}

Configs

remote.log.metadata.topic.replication.factor	topic的副本因子 默认值: 3
remote.log.metadata.topic.num.partitions	topic的分区数 默认值: 50
remote.log.metadata.topic.retention.ms	topic过期时间 默认值: -1, 没有过期限制。 用户可以根据自己的情况来配置这个选项，为了避免数据丢失，这个选项的值应该大于多层存储中配置的topic的过期时间
remote.log.metadata.manager.listener.name	此项配置为了通过RemoteLogMetadataManager的实现类与本地broker建联，本地实现了接口RemoteLogMetadataManager的类名，默认配置为`org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager`，可以手动修改。相对应的endpoint接入点为配置项"bootstrap.servers"
remote.log.metadata.*	默认RLMM的实现类会创建producer及consumer的实例，常规的client端配置通常是以remote.log.metadata.common.client. 作为前缀的，用户也可以通过指定remote.log.metadata.producer. 及remote.log.metadata.consumer. 来覆盖common配置。这些配置都将通过方法RemoteLogMetadataManager#configure(Map<String, ?> props) 来实现 例如：“rlmm.config.remote.log.metadata.producer.batch.size=100”将会设置producer的batch.size配置
remote.partition.remover.task.interval.ms	删除远端分区的任务，在执行前后两次删除任务的时间间隔，默认3600000，即 1 小时

Committed offsets file format

已经提交的位点信息会存储在一个名为_rlmm_committed_offsets的本地文件，这个文件在log dir目录下。这个文件为每个分区都创建了一个键值对：“<partition-no> <offset>”。_rlmm_committed_offsets的文件内容举例：

0 2022
4 104
2 498

Internal flat-file store format of remote log metadata

RLMM存储远端日志的元数据，并为用户topic的每一个partition构建一个物化实例并存储在单独的打平文件中

打平的文件格式如下

<magic><topic-name><topic-id><metadata-topic-offset><sequence-of-serialized-entries>
 
magic:                 
    unsigned var int, version of this file format.
topic-name:            
    string, topic name.
topic-id:              
    uuid, uuid of topic
metadata-topic-offset: 
    var long, offset of the remote log metadata topic partition upto which this topic partition's remote log metadata is fetched.
serialized-entries:
   sequence of serialized entries defined as below, more types can be added later if needed.
 
Serialization of entry is done as mentioned below. This is very similar to the message format mentioned earlier for storing into the metadata topic. 
 
length    : unsigned var int, length of this entry which is sum of sizes of type, version, and data.
type      : unsigned var int, represents the value type. This value is 'apikey' as mentioned in the schema. 
version   : unsigned var int, the 'version' number of the type as mentioned in the schema. 
data      : record payload in kafka protocol message format, the schema is given below.
 
Both type and version are added before the data is serialized into record value.  Schema can be evolved by adding a new version with the respective changes. A new type can also be supported by adding the respective type and its version.
 
 
{
  "apiKey": 0,
  "type": "data",
  "name": "RemoteLogSegmentMetadataRecordStored",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    {
      "name": "SegmentId",
      "type": "uuid",
      "versions": "0+",
      "about": "Unique identifier of the log segment"
    },
    {
      "name": "StartOffset",
      "type": "int64",
      "versions": "0+",
      "about": "Start offset  of the segment."
    },
    {
      "name": "EndOffset",
      "type": "int64",
      "versions": "0+",
      "about": "End offset  of the segment."
    },
    {
      "name": "LeaderEpoch",
      "type": "int32",
      "versions": "0+",
      "about": "Leader epoch from which this segment instance is created or updated"
    },
    {
      "name": "MaxTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Maximum timestamp with in this segment."
    },
    {
      "name": "EventTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Event timestamp of this segment."
    },
    {
      "name": "SegmentLeaderEpochs",
      "type": "[]SegmentLeaderEpochEntry",
      "versions": "0+",
      "about": "Event timestamp of this segment.",
      "fields": [
        {
          "name": "LeaderEpoch",
          "type": "int32",
          "versions": "0+",
          "about": "Leader epoch"
        },
        {
          "name": "Offset",
          "type": "int64",
          "versions": "0+",
          "about": "Start offset for the leader epoch"
        }
      ]
    },
    {
      "name": "SegmentSizeInBytes",
      "type": "int32",
      "versions": "0+",
      "about": "Segment size in bytes"
    },
    {
      "name": "RemoteLogSegmentState",
      "type": "int8",
      "versions": "0+",
      "about": "State of the remote log segment"
    }
  ]
}
 
 
{
  "apiKey": 1,
  "type": "data",
  "name": "DeletePartitionStateRecord",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    {
      "name": "Epoch",
      "type": "int32",
      "versions": "0+",
      "about": "Epoch (controller or leader) from which this event is created. DELETE_PARTITION_MARKED is sent by the controller. DELETE_PARTITION_STARTED and DELETE_PARTITION_FINISHED are sent by remote log metadata topic partition leader."
    },
    {
      "name": "EventTimestamp",
      "type": "int64",
      "versions": "0+",
      "about": "Event timestamp of this segment."
    },
    {
      "name": "RemotePartitionDeleteState",
      "type": "int8",
      "versions": "0+",
      "about": "Deletion state of the remote partition"
    }
  ]
}

Message Formatter for the internal topic

当从远端元数据topic中消费到消息后，`org.apache.kafka.server.log.remote.storage.RemoteLogMetadataFormatter`这个类用来格式化消息，用户可以指定格式化的property，如下文所示。这个对于debug来说非常有帮助

Internal message format

partition:<val><sep>message-offset:<val><sep>type:<RemoteLogSegmentMetadata | RemoteLogSegmentMetadataUpdate | DeletePartitionState><sep>version:<_no_><vs>event-value:<string representation of the event>
 
val: represents the respective value of the key.
sep: represents the separator, default value is: ","
 
partition : Remote log metata topic partition number. This is optional.
Use print.partition property to print it, default is false
 
message-offset : Offset of this message in remote log metadata topic. This is optional.
Use print.message.offset property to print it, default is false
 
type: Event value type, which can be one of RemoteLogSegmentMetadata, RemoteLogSegmentMetadataUpdate, DeletePartitionState values.
 
version: Version number of the event value type. This is optional.
Use print.version property to print it, default is false
 
Use print.all.event.value.fields to print the string representation of the event which will include all the fields in the data, default property value is false.
 
Event value can be of any of the types below:
 
remote-log-segment-id is represented as "{id:<><sep>topicId:<val><sep>topicName:<val><sep>partition:<val>}" in the event value.
topic-id-partition is represented as "{topicId:<val><sep>topicName:<val><sep>partition:<val>}" in the event value.
 
For RemoteLogSegmentMetadata
default representation is "{remote-log-segment-id:<val><sep>start-offset:<val><sep>end-offset:<val><sep>leader-epoch:<val><sep>remote-log-segment-state:<COPY_SEGMENT_STARTED | COPY_SEGMENT_FINISHED | DELETE_SEGMENT_STARTED | DELETE_SEGMENT_FINISHED>}"
 
For RemoteLogSegmentMetadataUpdate
default representation is "{remote-log-segment-id:<val><sep>leader-epoch:<val><sep>remote-log-segment-state:<COPY_SEGMENT_STARTED | COPY_SEGMENT_FINISHED | DELETE_SEGMENT_STARTED | DELETE_SEGMENT_FINISHED>}"
 
For DeletePartitionState
default representation is "{topic-id-partition:<val><sep>epoch:<val><sep>remote-partition-delete-state:<DELETE_PARTITION_MARKED | DELETE_PARTITION_STARTED | DELETE_PARTITION_FINISHED>

Topic deletion lifecycle

当一个controller收到了一个删除topic的请求，那么将会遵循现有的Kafka删除协议，将此topic对应的所有的replicas标记为离线offline，并且停止一切拉取请求。当所有的副本replica都达到了offline状态后，controller向RLMM发布一个删除事件（调用方法RemoteLogMetadataManager.updateRemotePartitionDeleteMetadata），将topic标记为deleted，即设置为RemotePartitionDeleteState#DELETE_PARTITION_MARKED。因为为topic引入了uuid，因此topic的删除操作可变成异步操作 译者：这里其实引入了topicId，避免快速删除topic后又新建同名topic 。这个设计允许以后通过将删除标记发布到远程日志元数据topic中来回收远程日志。RLMM则收到DELETE_PARTITION_MARKED状态后，就触发了异步删除远端元数据的操作

默认RLMM处理删除操作使用的类为RemotePartitionRemover(RPRM)

如果某个broker是topic __remote_log_metadata对应partition的leader，RPRM的实例将会在这个broker上创建。当某个topic被标记为删除，那么RPRM就要负责将其从远端存储上删除。RLMM消费远端partition元数据的消息，然后从中过滤删除partition的事件。它收集那些待删除的partitions，然后调用对应的RemoteStorageManager类来删除日志segment。这个执行间隔的时间是通过配置remote.partition.remover.task.interval.ms来设置的（默认1小时）。一旦删除操作成功执行，那么它将会提交消费位点