了解 Kafka-on-Pulsar (KoP)：昨天、今天和明天

深入了解 KoP 概念，回答常见问题，以及 KoP 社区已经和将要对项目进行的最新和未来改进。

每日分享最新，最流行的软件开发知识与最新行业趋势，希望大家能够一键三连，多多支持，跪求关注，点赞，留言。

Apache Pulsar 和Apache Kafka围绕日志共享相似的数据模型。这使得实现与 Kafka 兼容的协议处理程序成为可能，以便 Kafka 用户可以将现有的 Kafka 应用程序和服务迁移到 Pulsar，而无需修改代码。它还允许这些应用程序利用各种 Pulsar 功能，例如多租户、无限事件流保留和无服务器事件处理。这就是开源项目 Kafka-on-Pulsar (KoP) 的用武之地。

社区为什么要开发 KoP？

首先，我认为回答一个关键问题很重要：“你为什么开发它？” 作为 KoP 项目的维护者，我注意到这通常是新社区成员首先想到的事情。为了回答这个问题，让我们快速了解一下 Pulsar 和 Kafka 的基础知识。

Pulsar 和 Kafka 有一些共同的概念。例如，他们都有生产者和消费者，经纪人托管不同的主题来为他们服务。此外，它们都支持主题分区以实现更好的并行性。

在 Kafka 中，客户端通过请求向领导者代理发送消息PRODUCE。这些消息在本地持久化到节点。每个追随者通过请求从领导者那里读取数据FETCH以存储消息的副本。在这种领导者-从者架构中，每个代理都需要处理数据处理和存储。

该设计的一个缺点是它保证最新和相关的数据副本仅存储在领导代理上，该代理为生产者和消费者提供服务。这意味着 Kafka 集群在流量爆发期间可能会不堪重负，因为负载可能不会分布在追随者之间。

相比之下，Pulsar 将服务（brokers）和存储（bookies）分成不同的层。在计算层，所有的 Pulsar broker 都是无状态的，并且彼此等价。如下图所示，客户端通过SEND请求向代理发送消息。在处理完消息后，broker 通过ADD_ENTRY请求将它们传递给不同的 bookie。具体来说，根据配置的写入策略（即 ensemble size、write quorum 和 ack quorum 的值）将数据写入 bookie。这有助于实现跨存储节点的数据高可用性。最重要的是，经纪人和博彩公司都可以在自己的层轻松扩展，而不会影响另一层。

Pulsar 的这种云原生架构具有出色的可扩展性、可用性和弹性，为用户提供了 Kafka 中一些关键痛点的解决方案。因此，许多用户正在寻找一个从 Kafka 迁移到 Pulsar 的优雅计划。

迁移计划

在开发 KoP 之前，从 Kafka 迁移到 Pulsar 并非易事。我注意到一些社区成员使用并且可能仍在使用以下方式：

更新客户端

用户需要重写他们的代码并优化一些客户端配置。诚然，这不是一个理想的解决方案，因为它可能会产生额外的费用。除此之外，与 Kafka 相比，Pulsar 的生态系统还有很长的路要走。后者在许多方面享有更成熟的生态系统，例如多语言客户端支持和第三方集成。这意味着当您尝试使用 Kafka 生态系统中的工具时，重写代码可能会导致许多问题。

Apache Kafka 的 Pulsar 适配器

最初，Pulsar 社区试图通过为 Apache Kafka 开发一个名为 Pulsar Adapter 的工具来解决迁移问题。它允许用户用 Pulsar Kafka 包装器替换 Kafka 客户端依赖项。它不需要对现有代码进行任何更改。然而，它的缺点是显而易见的：

1. 仅适用于基于 Java 的客户端
2. 处理 Kafka 偏移量的问题
3. 用户仍然需要学习一些 Pulsar 客户端配置

脉冲星上的卡夫卡 (KoP)

为了给用户提供更流畅的迁移体验，KoP 社区提出了一个新的解决方案。他们决定通过在 Pulsar 代理上引入 Kafka 协议处理程序来为 Pulsar 带来原生 Kafka 协议支持。协议处理程序是 Pulsar 2.5.0 中引入的一项新功能。它们允许 Pulsar 代理支持其他消息传递协议，包括 Kafka、AMQP 和MQTT。

与上述迁移计划相比，KoP 具有以下主要优势：

1. 无需修改代码：用户无需修改 Kafka 应用程序中的任何代码，包括不同语言编写的客户端、应用程序本身和第三方组件
2. 兼容性强： KoP 与 Kafka 生态系统中的大多数工具兼容。目前支持Kafka 0.9+
3. 与 Pulsar Brokers 的直接交互：在设计 KoP 之前，一些用户试图通过在中间创建代理层来使 Pulsar 客户端服务于 Kafka 客户端发送的请求。这可能会影响性能，因为它需要额外的路由请求。相比之下，KoP 允许客户直接与 Pulsar 代理进行通信，而不会影响性能

要详细了解开发 KoP 的原因，请参阅 KoP 白皮书。

KoP 的工作原理

KoP 被实现为协议名称为“Kafka”的协议处理程序插件。它在 Pulsar 代理重新启动时加载。默认情况下，Kafka 和 Pulsar 客户端可以同时工作。

如果你查看 Pulsar 的源代码，你会发现一个名为 ServerCnx 的类，它暴露了 6650 端口。Pulsar 客户端向该端口发送请求后，ServerCnx 会解析这些请求，以便代理可以采取相应的操作，例如读写来自和来自 BookKeeper 的数据。

为集群启用 KoP 后，它会公开端口 9092，以便 Kafka 客户端可以向其发送请求。与 ServerCnx 类似，KoP 处理这些请求，然后要求代理做出响应。ServerCnx 和 KoP 都可以访问所有代理资源，例如主题和订阅。

实施与设计

让我们检查一些关键概念，尤其是它们是如何在 KoP 中实现的。

主题和分区

在 Kafka 中，所有主题都存储在一个平面命名空间中。在 Pulsar 的多租户架构中，消息被组织在租户、命名空间和主题的三层层次结构中。Pulsar 主题可以是持久的或非持久的。随着 Kafka 的持久化，KoP 中的所有消息和所有主题都是持久的。以下是 Pulsar 主题命名约定（分区是主题名称的一部分作为后缀）：

{persistent | non-persistent}://tenant/namespace/topic-partition-0

为了使 Kafka 用户能够利用 Pulsar 的多租户特性，KoP 允许他们直接使用简短的主题名称，因为它提供了默认租户和命名空间。同时，用户还可以指定 Pulsar 租户和命名空间，以便客户端可以访问其他主题。详情见下表：

在此表中，kafkaTenant定义kafkaNamespaceKoP 用于短主题名称的默认名称，默认为public和default。

验证

在生产环境中，Pulsar 客户端和 Kafka 客户端可能有不同的命名空间。为了提高安全性，您可以为 KoP 启用身份验证。它使用Kafka SASL 机制和Pulsar 中基于令牌的身份验证提供程序进行身份验证。因此，您需要为以下组件启用身份验证：

1. 脉冲星经纪人
2. KoP（需要通过配置来指定机制saslAllowedMechanisms，目前只支持PLAIN和OAUTHBEARER）
3. 卡夫卡客户

身份验证的工作原理见下图：

SASL_HANDSHAKEKafka 客户端使用为 Kafka 客户端配置的机制发送请求。然后 KoP 验证是否支持该机制。如果是，KoP 基于该机制创建 SaslServer。

KoP 响应 Kafka 客户端后，将SASL_AUTHENTICATE带有令牌的请求发送回 KoP。KoP 根据之前创建的 SaslServer 验证令牌。

有关如何配置身份验证的详细信息，请参阅 Pulsar 上的 Kafka 中的安全性。

授权

您还可以在 KoP 中配置授权，以定义不同客户端拥有的权限。例如，一些客户端只能产生消息，而其他客户端可能能够消费消息。Kafka 使用 ACL（访问控制列表）和资源类型来控制可用于不同资源的操作。KoP 支持Topic资源类型。请看下图，了解它如何映射不同的 ACL 操作：

身份验证完成后，KoP 获取一个授权 ID（或 Pulsar 角色）并将其发送到AuthorizationService代理上。

注意：一个 Pulsar 角色是一个字符串，可以代表一个或多个客户端。角色用于管理客户端权限，例如生成或使用某些主题的消息。Pulsar 使用身份验证提供程序来建立客户端的身份，然后将角色令牌分配给该客户端。然后，此角色令牌用于授权和 ACL，以确定客户端有权执行的操作。

小组协调员

为 KoP 实现 Kafka 组协调器非常具有挑战性。这是因为 Pulsar 没有一个集中的组协调器来将分区分配给消费者或管理偏移量。小组协调员有两个主要职责：

1. 再平衡：为同一组中的消费者分配不同的分区（订阅）
2. Commit offsets : 坚持一个特殊的offset主题的最新offset信息

在 KoP 中，组协调器逻辑是用 Java 重写的，但与 Kafka 中的原始实现略有不同。KoP 引入了一个名为 Namespace Bundle Listener 的新组件。在 Kafka 中，follower 节点通知 leader ISR 变化。这与 Pulsar 不同，因为 Pulsar 代理无法相互通信。Pulsar 使用包作为分片机制。主题通过主题名称的散列分配给特定的包。

注意：这篇文章没有详细解释捆绑包。在这里，您只需要知道该组件监视捆绑包的更改，以便组协调器可以分配分区。

读取和写入

对于 offset 主题，Pulsar 生产者和消费者直接执行读取和写入。对于普通消息，KoP 使用托管分类帐和代理上的托管游标进行读取和写入。

托管分类帐在 BookKeeper 之上提供了额外的存储抽象。托管账本管理多个 Pulsar 段（在 BookKeeper 中也称为账本，用于在 bookie 上存储消息）。它包含它们的大小和状态信息。托管游标是托管分类帐中的持久游标。它从托管分类帐中读取，并在消费者完成消费消息时发出信号。

请记住：KoP 使用代理资源直接从 BookKeeper 读取和写入消息，而不是创建客户端。这也减少了额外的网络跃点。

抵消实施

Kafka 为分区中的记录维护一个数字偏移量，表示消费者在分区中的位置。在 Pulsar 中，没有与 Kafka 偏移量等效的概念（Pulsar 使用游标来跟踪订阅的消费和确认信息。请注意，游标的工作方式比 Kafka 偏移量更复杂）。

早期的 KoP 版本使用简单的转换方法处理偏移量，但它不允许连续偏移量，很容易导致问题。为了解决这个问题，KoP 社区在 KoP 2.8.0 中引入了一个新概念“代理条目元数据”，以实现连续偏移。由于这是一个非常复杂的话题，我建议您参考我之前的一篇博客 Offset Implementation in Kafka-on-Pulsar 来全面了解 KoP 中的 offset 实现。

开始使用 KoP

如果您有 Apache Pulsar 集群，您可以通过直接在 Pulsar 代理上下载和安装 KoP 协议处理程序，在现有 Pulsar 集群上启用 KoP。要开始，请执行以下步骤：

1. 在Releases页面上下载 nar 文件，并将其移动到protocols存储 Pulsar 包的目录中的文件夹（手动创建）。或者，克隆 KoP GitHub 存储库并在本地构建项目
2. 在broker.conf或standalone.conf中，配置以下字段：

messagingProtocols=kafka allowAutoTopicCreationType=partitioned listeners=PLAINTEXT://127.0.0.1:9092 brokerEntryMetadataInterceptors=\ org.apache.pulsar.common.intercept.AppendlndexMetadatalntercepor

1. 启动您的 Pulsar 代理。

在选择不同版本的 KoP 时，请记住：

1. 一个 KoP 版本有四个次要数字，即 xyzm 前三个数字指的是这个 KoP 版本兼容的 Pulsar 版本（例如 KoP 2.9.1.2 兼容 Pulsar 2.9.1）
2. 当 Pulsar 次要版本升级时，将不会维护相应的旧版本 KoP。例如，Pulsar 2.9.2 发布后，KoP 社区没有维护 KoP 2.9.1.m
3. 不推荐 2.8 之前的 Pulsar 版本
4. 从源代码构建 KoP 项目时，不要使用master分支。使用特定于您正在使用的 Pulsar 版本的分支

关于 KoP 的常见问题

我从 KoP 社区收集了一些问题，并在此处列出了我的解决方案和建议。

如何指定entryFormat？

entryFormat定义 KoP 条目的格式。允许的值为pulsar、kafka和mixed_kafka。它默认为pulsar，这意味着 Kafka 客户端可以使用 Pulsar 客户端生成的消息，并且 Pulsar 客户端也可以使用 Kafka 客户端生成的消息。但是，消息解码和编码过程可能很耗时。

1. Kafka 格式的消息需要解压以重新创建 Pulsar 格式的消息
2. Pulsar 格式的消息需要解压以重新创建 Kafka 格式的消息

对于生产中的大多数应用程序，我建议您配置entryFormat=kafka. 通过这种方式，KoP 在消息的元数据中添加一个键值，表明它来自 Kafka 客户端，然后将其写入 BookKeeper，而无需进行不必要的编码和解码。这种方法的缺点是 Pulsar 客户端无法消费 Kafka 客户端生成的消息。由于这些消息以 Kafka 的原始格式写入 bookie，因此 Pulsar 客户端无法识别它们。为了解决这个问题，社区在 KoP 2.9.0 中为 Pulsar 客户端设计了一个消息负载处理器。它允许 Pulsar 消费者消费 Kafka 生产者产生的消息，即使你设置entryFormat为kafka. messagePayloadProcessor您可以使用如下方式在您的消费者应用程序中配置处理器：

爪哇1最终 消费者<字节[] > 消费者 = 客户端。新消费者()23 . 主题（“我的主题”）45 . 订阅名称（“我的订阅”）67 . messagePayloadProcessor (新的 KafkaPayloadProcessor ())89 . 订阅（）；

要导入KafkaPayloadProcessor，您需要添加以下依赖项。请注意，它pulsar.version应该与您的pulsar-client依赖项的版本相同。

io.streamnative.pulsar.handlers kafka-payload-processor ${pulsar.version}

有关详细信息，请参阅KoP 文档。

消费者跟不上消息产生的速度

解决这个问题的一个简单方法是增加 的值maxreadEntriesNum：

maxreadEntriesNum=5

此属性表示 KoP 每次处理FETCH请求时读取的最大条目数。增加此值会减少网络上所需的请求数。但是，如果它变得太大，可能会给内存带来额外的压力。请注意，ManagedCursor 当前不检查基于字节的读取限制。

为什么使用 Pulsar-Admin 时看不到消费统计？

社区移除了 KoP 2.8.0 及更高版本中的话题消费统计。如果您使用pulsar-admin查看某个主题的信息，则看不到任何可用的消费详情。删除的原因是：

1. 最初，当 KoP 提交偏移量时，它需要找到偏移量的消息 ID，以便确认消息。这个过程太耗时
2. Kafka 通过组协调器管理偏移量。但是，协调代理可能不是领导代理。在 Pulsar 中，只有主题所有者 broker 可以确认消息

KoP 现在提供了多种基于 Prometheus 的指标来显示消费信息，可以通过 Grafana 访问。有关详细信息，请参阅 KoP 指标。

错误信息

NOT_LEADER_OR_FOLLOWER

(org.apache.kafka.clients.producer.internals.Sender)

当 Pulsar 自动删除分区主题时（默认情况下，在 Pulsar 中不活动的主题会被自动删除），它不会删除分区号元数据，它存储在 ZooKeeper 中。因此，当 KoP 从客户端收到有关元数据的请求时，它仍然可以找到主题。然后 KoP 将 leader broker 信息发送给客户端，导致如下错误。

Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)

要解决此问题，请使用以下任一方法：

1. 禁用自动删除非活动主题broker.conf：brokerDeleteInactiveTopicsEnabled=false
2. 启用自动删除非活动分区主题的元数据broker.conf。如果您不想在生产中禁用主题自动删除，这是首选解决方案：brokerDeleteInactivePartitionedTopicMetadataEnabled=true

KoP 有什么新功能？

在 OpenMessaging Benchmark Framework 中支持 KoP

首先，让我简单解释一下什么是 OpenMessaging Benchmark Framework，以及如果你想使用它需要的脚本和配置。

OpenMessaging Benchmark Framework 是一套工具，可以轻松地对云中的分布式消息系统进行基准测试。OpenMessaging 基准测试套件目前可用于 Apache Pulsar、Apache Kafka、Apache RocketMQ、Redis 和 Pravega 等系统。它为每个受支持的消息系统提供以下内容：

1. 一个 Terraform 脚本，在云提供商（例如 AWS）上自动申请和创建资源
2. 一个 Ansible 脚本，可在远程机器上自动安装所有必要的组件（MQ、监控和基准测试服务）
3. 一个基于 Java 的驱动程序，可让您轻松运行 MQ 客户端。您可以在 YAML 文件中自定义必要的客户端配置

以前，当您使用 OpenMessaging Benchmark Framework 测试 KoP 的性能时，您必须通过添加与 KoP 相关的逻辑来手动更改 Pulsar Ansible 脚本。这是因为脚本默认不部署 Kafka 协议处理程序。此外，您必须使用 Kafka 驱动程序和配置来启动测试中的 Kafka 生产者和消费者。

为了使用基准简化 KoP 的性能测试，KoP 社区专门为 KoP 创建了一个驱动程序。它支持 Kafka 和 Pulsar 客户端。换句话说，您可以同时使用 Kafka 和 Pulsar 生产者和消费者的混合组合。它包含有效负载处理器的逻辑，如果您设置entryFormat为kafka.

这是使用 KoP 驱动程序的示例：

./bin/benchmark -d driver-kop/kafka_to_pulsar.yaml -o kop.json workloads/1-topic-1-partition-100b.yaml ./bin/benchmark -d driver-kop/kafka_to_kafka.yaml -o kop.json workloads/1-topic-1-partition-100b.yaml ./bin/benchmark -d driver-kop/pulsar_to_kafka.yaml -o kop.json workloads/1-topic-1-partition-100b.yaml

有关更多信息，请参阅 OpenMessaging Benchmark Framework GitHub 存储库上的 driver-kop 目录。

注意：协议处理程序的配置也得到了简化。用户可以protocol_handlers通过指定协议类型、配置文件和下载 URL 轻松自定义变量。请参阅此拉取请求和Apache Pulsar 基准以了解更多详细信息。

自动删除 Znode 上的非活动组信息

KoP 在处理来自消费者的 FETCH 请求时利用 ZooKeeper 来存储消费者的组信息。然后可以使用存储在 znode 上的组信息来更新消费者指标。KoP 之所以为此使用 ZooKeeper，是因为您无法直接在 FETCH 请求中获取组信息。如下图所示，一个consumer向leader broker和coordinator broker发送不同的请求，而这两个broker可能并不相同。

以前，znode 上的组信息从未被删除，这可能会变得非常大，并导致 KoP 集群出现其他问题。为了解决这个问题，社区找到了一种在群组不再活跃时自动删除群组信息的方法。

除了上述改进之外，KoP 社区在修复现有 bug 和添加一些小功能方面也取得了一些进展。如果您有兴趣，可以看看其中的一些。

错误修复：PR 1038、PR 973、PR 1125、PR 1230和PR 1276。

新功能：PR 1006和PR 1125。

下一步是什么？

KoP 社区希望通过以下方式改进项目。

1. 添加模式支持：
   • Confluent Schema Registry API 兼容性
   • 添加兼容的序列化器和反序列化器，以便 Kafka 客户端可以与 Pulsar 模式交互
2. 改进交易功能
3. 对其他语言的 OAuth2 插件进行一些研究，例如 C、C++、Go 和 Python
4. 完成KoP的性能测试报告

如果您退后一步思考一下 KoP 已经走了多远，就会发现越来越多的用户加入 KoP 社区并以不同的方式做出贡献也就不足为奇了。与 Apache Pulsar 生态系统中的其他工具一样，如果没有所有相关社区成员的努力，KoP 不可能走到这一步。如果您对 KoP 感兴趣，请随时提交拉取请求或在其 GitHub 存储库上打开问题。