Kafka 操作之分层存储（Tiered Storage）

一. 前言

二. 分层存储（Tiered Storage）

2.1. 分层存储概述（Tiered Storage Overview）

2.2. 配置（Configuration）

2.2.1. Broker 配置（Broker Configurations）

2.2.2. Topic 配置（Topic Configurations）

2.3. 快速入门示例（Quick Start Example）

2.4. 局限性（Limitations）

一. 前言

Kafka 分层存储通常是通过配置 Kafka 的日志保留策略来实现的，而不是直接通过代码来实现的。Kafka 支持基于时间和基于大小的日志保留策略。

二. 分层存储（Tiered Storage）

2.1. 分层存储概述（Tiered Storage Overview）

原文引用：Kafka data is mostly consumed in a streaming fashion using tail reads. Tail reads leverage OS's page cache to serve the data instead of disk reads. Older data is typically read from the disk for backfill or failure recovery purposes and is infrequent.

Kafka 数据主要是以流式方式使用尾部读取来消耗的。尾部读取利用操作系统的页面缓存来提供数据，而不是磁盘读取。较旧的数据通常是为了回填或故障恢复而从磁盘中读取的，而且这种情况并不常见。

原文引用：In the tiered storage approach, Kafka cluster is configured with two tiers of storage - local and remote. The local tier is the same as the current Kafka that uses the local disks on the Kafka brokers to store the log segments. The new remote tier uses external storage systems, such as HDFS or S3, to store the completed log segments. Please check KIP-405 for more information.

在分层存储方法中，Kafka 集群配置有两层存储——本地和远程。本地层与当前的 Kafka 相同，后者使用 Kafka Broker 上的本地磁盘来存储日志段。新的远程层使用外部存储系统（如 HDFS 或S3）来存储已完成的日志段。请查看 KIP-405 了解更多信息。

原文引用：Note: Tiered storage is considered as an early access feature, and is not recommended for use in production environments

注意：分层存储被视为早期访问功能，不建议在生产环境中使用。

2.2. 配置（Configuration）

2.2.1. Broker 配置（Broker Configurations）

原文引用：By default, Kafka server will not enable tiered storage feature. remote.log.storage.system.enable is the property to control whether to enable tiered storage functionality in a broker or not. Setting it to "true" enables this feature.

默认情况下，Kafka 服务器不会启用分层存储功能。remote.log.storage.system.enable 是用于控制是否在 Broker 中启用分层存储功能的属性。将其设置为“true”将启用此功能。

原文引用：RemoteStorageManager is an interface to provide the lifecycle of remote log segments and indexes. Kafka server doesn't provide out-of-the-box implementation of RemoteStorageManager. Configuring remote.log.storage.manager.class.name and remote.log.storage.manager.class.path to specify the implementation of RemoteStorageManager.

RemoteStorageManager 是一个提供远程日志段和索引的生命周期的接口。Kafka 服务器不提供 RemoteStorageManager 的开箱即用实现。配置 remote.log.storage.manager.class.name 和remote.log.store.manager.class.path 以指定 RemoteStorageManager 的实现。

原文引用：RemoteLogMetadataManager is an interface to provide the lifecycle of metadata about remote log segments with strongly consistent semantics. By default, Kafka provides an implementation with storage as an internal topic. This implementation can be changed by configuring remote.log.metadata.manager.class.name and remote.log.metadata.manager.class.path. When adopting the default kafka internal topic based implementation, remote.log.metadata.manager.listener.name is a mandatory property to specify which listener the clients created by the default RemoteLogMetadataManager implementation.

RemoteLogMetadataManager 是一个接口，用于提供具有强一致语义的远程日志段元数据的生命周期。默认情况下，Kafka 提供了一个将存储作为内部 Topic 的实现。可以通过配置remote.log.metadata.manager.class.name 和 remote.log.etadata.manager.class.path 来更改此实现。当采用默认的基于 Kafka 内部 Topic 的实现时，remote.log.metadata.manager.listener.name是一个强制属性，用于指定由默认 RemoteLogMetadataManager 实现创建的客户端的侦听器。

2.2.2. Topic 配置（Topic Configurations）

原文引用：After correctly configuring broker side configurations for tiered storage feature, there are still configurations in topic level needed to be set. remote.storage.enable is the switch to determine if a topic wants to use tiered storage or not. By default it is set to false. After enabling remote.storage.enable property, the next thing to consider is the log retention. When tiered storage is enabled for a topic, there are 2 additional log retention configurations to set:

local.retention.ms
retention.ms
local.retention.bytes
retention.bytes

在为分层存储功能正确配置了 Broker 端配置之后，仍然需要设置 Topic 级别的配置。remote.storage.enable 是用于确定 Topic 是否要使用分层存储的开关。默认情况下，它设置为false。启用 remote.storage.enable 属性后，接下来要考虑的是日志保留。当为某个 Topic 启用分层存储时，还需要设置2个额外的日志保留配置：

local.retension.ms
retension.ms
local.retension.bytes
retention.bytes

原文引用：The configuration prefixed with local are to specify the time/size the "local" log file can accept before moving to remote storage, and then get deleted. If unset, The value in retention.ms and retention.bytes will be used.

前缀为 local 的配置用于指定“本地”日志文件在移动到远程存储之前可以接受的时间/大小，然后删除。如果未设置，将使用 retention.ms 和 retention.bytes 中的值。

2.3. 快速入门示例（Quick Start Example）

原文引用：Apache Kafka doesn't provide an out-of-the-box RemoteStorageManager implementation. To have a preview of the tiered storage feature, the LocalTieredStorage implemented for integration test can be used, which will create a temporary directory in local storage to simulate the remote storage.

To adopt the `LocalTieredStorage`, the test library needs to be built locally

Apache Kafka 没有提供现成的 RemoteStorageManager 实现。要预览分层存储功能，可以使用为集成测试实现的 LocalTieredStorage，它将在本地存储中创建一个临时目录来模拟远程存储。

要采用“LocalTieredStorage”，需要在本地构建测试库

# please checkout to the specific version tag you're using before building it
# ex: `git checkout 3.6.1`
./gradlew clean :storage:testJar

原文引用：After build successfully, there should be a `kafka-storage-x.x.x-test.jar` file under `storage/build/libs`. Next, setting configurations in the broker side to enable tiered storage feature.

构建成功后，“storage/build/libs”下应该有一个“kafka-storage-x.x-test.jar”文件。接下来，在 Broker 端设置配置以启用分层存储功能。

# Sample Zookeeper/Kraft broker server.properties listening on PLAINTEXT://:9092
remote.log.storage.system.enable=true# Setting the listener for the clients in RemoteLogMetadataManager to talk to the brokers.
remote.log.metadata.manager.listener.name=PLAINTEXT# Please provide the implementation info for remoteStorageManager.
# This is the mandatory configuration for tiered storage.
# Here, we use the `LocalTieredStorage` built above.
remote.log.storage.manager.class.name=org.apache.kafka.server.log.remote.storage.LocalTieredStorage
remote.log.storage.manager.class.path=/PATH/TO/kafka-storage-x.x.x-test.jar# These 2 prefix are default values, but customizable
remote.log.storage.manager.impl.prefix=rsm.config.
remote.log.metadata.manager.impl.prefix=rlmm.config.# Configure the directory used for `LocalTieredStorage`
# Note, please make sure the brokers need to have access to this directory
rsm.config.dir=/tmp/kafka-remote-storage# This needs to be changed if number of brokers in the cluster is more than 1
rlmm.config.remote.log.metadata.topic.replication.factor=1# Try to speed up the log retention check interval for testing
log.retention.check.interval.ms=1000

原文引用：Following quick start guide to start up the kafka environment. Then, create a topic with tiered storage enabled with configs:

按照快速入门指南启动 Kafka 环境。然后，创建一个 Topic，其中使用配置启用了分层存储：

# remote.storage.enable=true -> enables tiered storage on the topic
# local.retention.ms=1000 -> The number of milliseconds to keep the local log segment before it gets deleted.Note that a local log segment is eligible for deletion only after it gets uploaded to remote.
# retention.ms=3600000 -> when segments exceed this time, the segments in remote storage will be deleted
# segment.bytes=1048576 -> for test only, to speed up the log segment rolling interval
# file.delete.delay.ms=10000 -> for test only, to speed up the local-log segment file delete delaybin/kafka-topics.sh --create --topic tieredTopic --bootstrap-server localhost:9092 \
--config remote.storage.enable=true --config local.retention.ms=1000 --config retention.ms=3600000 \
--config segment.bytes=1048576 --config file.delete.delay.ms=1000

尝试将消息发送到“tieredTopic”以滚动日志段（Try to send messages to the `tieredTopic` topic to roll the log segment）：

bin/kafka-producer-perf-test.sh --topic tieredTopic --num-records 1200 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=localhost:9092

原文引用：Then, after the active segment is rolled, the old segment should be moved to the remote storage and get deleted. This can be verified by checking the remote log directory configured above. For example:

然后，在滚动活动段之后，应将旧段移动到远程存储并删除。这可以通过检查上面配置的远程日志目录来验证。例如：

> ls /tmp/kafka-remote-storage/kafka-tiered-storage/tieredTopic-0-jF8s79t9SrG_PNqlwv7bAA
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.index
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.snapshot
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.leader_epoch_checkpoint
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.timeindex
00000000000000000000-knnxbs3FSRyKdPcSAOQC-w.log

原文引用：Lastly, we can try to consume some data from the beginning and print offset number, to make sure it will successfully fetch offset 0 from the remote storage.

最后，我们可以尝试从一开始就消耗一些数据并打印偏移量编号，以确保它能成功地从远程存储中获取偏移量 0。

bin/kafka-console-consumer.sh --topic tieredTopic --from-beginning --max-messages 1 --bootstrap-server localhost:9092 --property print.offset=true

原文引用：Please note, if you want to disable tiered storage at the cluster level, you should delete the tiered storage enabled topics explicitly. Attempting to disable tiered storage at the cluster level without deleting the topics using tiered storage will result in an exception during startup.

请注意，如果要在集群级别禁用分层存储，则应明确删除启用分层存储的 Topic。尝试在集群级别禁用分层存储而不删除使用分层存储的 Topic，将导致启动过程中出现异常。

bin/kafka-topics.sh --delete --topic tieredTopic --bootstrap-server localhost:9092

原文引用：After topics are deleted, you're safe to set remote.log.storage.system.enable=false in the broker configuration.

删除主题后，可以安全地在 Broker 配置中设置 remote.log.storage.system.enable=false。

2.4. 局限性（Limitations）

原文引用：While the early access release of Tiered Storage offers the opportunity to try out this new feature, it is important to be aware of the following limitations:

No support for clusters with multiple log directories (i.e. JBOD feature)
No support for compacted topics
Cannot disable tiered storage at the topic level
Deleting tiered storage enabled topics is required before disabling tiered storage at the broker level
Admin actions related to tiered storage feature are only supported on clients from version 3.0 onwards

For more information, please check Tiered Storage Early Access Release Note.

虽然分层存储的早期访问版本提供了尝试这一新功能的机会，但重要的是要注意以下限制：