将 Pub/Sub 连接到 Apache Kafka

本文档介绍如何使用 Pub/Sub Group Kafka Connector 集成 Apache Kafka 和 Pub/Sub。

注意 ：您还可以将此连接器与 Pub/Sub Lite 搭配使用。请参阅将 Pub/Sub Lite 连接到 Apache Kafka。

Pub/Sub Group Kafka Connector 简介

Apache Kafka 是一个用于流式处理事件的开源平台。它通常用于分布式架构，以实现松散耦合组件之间的通信。Pub/Sub 是一种用于异步发送和接收消息的代管式服务。与 Kafka 一样，您可以使用 Pub/Sub 在云架构中的组件之间进行通信。

借助 Pub/Sub Group Kafka Connector，您可以集成这两个系统。以下连接器打包在连接器 JAR 中：

• 接收器连接器从一个或多个 Kafka 主题读取记录并将其发布到 Pub/Sub。
• 源连接器从 Pub/Sub 主题中读取消息并将其发布到 Kafka。

以下是您可能会使用 Pub/Sub Group Kafka Connector 的一些场景：

• 您正在将基于 Kafka 的架构迁移到 Google Cloud。
• 您有一个前端系统将事件存储在 Google Cloud 之外的 Kafka，但您也可以使用 Google Cloud 运行一些需要接收 Kafka 事件的后端服务。
• 您从本地 Kafka 解决方案收集日志，并将其发送到 Google Cloud 进行数据分析。
• 您有一个使用 Google Cloud 的前端系统，但您还使用 Kafka 在本地存储数据。

连接器需要使用 Kafka Connect，这是一个在 Kafka 与其他系统之间流式传输数据的框架。如需使用连接器，您必须将 Kafka Connect 与 Kafka 集群一起运行。

本文档假定您熟悉 Kafka 和 Pub/Sub。在阅读本文档之前，建议您先完成其中一个 Pub/Sub 快速入门。

Pub/Sub 连接器不支持 Google Cloud IAM 与 Kafka Connect ACL 之间的任何集成。

开始使用连接器

本部分将引导您完成以下任务：

1. 配置 Pub/Sub Group Kafka Connector。
2. 将事件从 Kafka 发送到 Pub/Sub。
3. 将消息从 Pub/Sub 发送到 Kafka。

注意 ：本部分在独立模式下使用适合测试的 Kafka Connect。对于生产系统，建议使用分布式模式。如需了解如何在分布式模式下运行 Kafka Connect，请参阅 Apache Kafka 文档中的运行 Kafka Connect。

前提条件

安装 Kafka

按照 Apache Kafka 快速入门中的说明，在本地机器上安装单节点 Kafka。完成快速入门中的下列步骤：

1. 下载最新的 Kafka 版本并将其解压缩。
2. 启动 Kafka 环境。
3. 创建 Kafka 主题。

身份验证

Pub/Sub 群组 Kafka 连接器必须通过 Pub/Sub 进行身份验证，才能发送和接收 Pub/Sub 消息。如需设置身份验证，请执行以下步骤：

1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。
2. 安装 Google Cloud CLI。

3. 如需初始化 gcloud CLI，请运行以下命令：

   gcloud init

4. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

5. 为您的 Google 账号创建本地身份验证凭据：

   gcloud auth application-default login

6. 向您的 Google 账号授予角色。对以下每个 IAM 角色运行以下命令一次： roles/pubsub.admin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

   • 将 PROJECT_ID 替换为您的项目 ID。
   • 将 EMAIL_ADDRESS 替换为您的电子邮件地址。
   • 将 ROLE 替换为每个角色。
7. 安装 Google Cloud CLI。

8. 如需初始化 gcloud CLI，请运行以下命令：

   gcloud init

9. Create or select a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   • Create a Google Cloud project:

     gcloud projects create PROJECT_ID

     Replace PROJECT_ID with a name for the Google Cloud project you are creating.

   • Select the Google Cloud project that you created:

     gcloud config set project PROJECT_ID

     Replace PROJECT_ID with your Google Cloud project name.

10. 为您的 Google 账号创建本地身份验证凭据：

    gcloud auth application-default login

11. 向您的 Google 账号授予角色。对以下每个 IAM 角色运行以下命令一次： roles/pubsub.admin

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

    • 将 PROJECT_ID 替换为您的项目 ID。
    • 将 EMAIL_ADDRESS 替换为您的电子邮件地址。
    • 将 ROLE 替换为每个角色。

注意 ：对于生产工作负载，我们建议使用工作负载身份联合进行身份验证。

下载连接器 JAR

将连接器 JAR 文件下载到本地计算机。如需了解详情，请参阅 GitHub 自述文件中的获取连接器。

复制连接器配置文件

1. 克隆或下载连接器的 GitHub 代码库。

   git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
   cd java-pubsub-group-kafka-connector
   

2. 将 config 目录的内容复制到 Kafka 安装的 config 子目录中。

   cp config/* [path to Kafka installation]/config/
   

这些文件包含连接器的配置设置。

更新您的 Kafka Connect 配置

1. 导航到包含您下载的 Kafka Connect 二进制文件的目录。
2. 在 Kafka Connect 二进制目录中，用文本编辑器打开名为 config/connect-standalone.properties 的文件。
3. 如果 plugin.path property 已被注释掉，请取消注释。

4. 更新 plugin.path property 以包含连接器 JAR 的路径。

   示例：

   plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar
   

5. 将 offset.storage.file.filename 属性设置为本地文件名。在独立模式下，Kafka 使用此文件来存储偏移数据。

   示例：

   offset.storage.file.filename=/tmp/connect.offsets
   

将事件从 Kafka 转发到 Pub/Sub

本部分介绍如何启动接收器连接器、将事件发布到 Kafka，以及从 Pub/Sub 读取转发的消息。

1. 使用 Google Cloud CLI 创建包含订阅的 Pub/Sub 主题。

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC
   

   请替换以下内容：

   • PUBSUB_TOPIC：从 Kafka 接收消息的 Pub/Sub 主题的名称。
   • PUBSUB_SUBSCRIPTION：主题的 Pub/Sub 订阅名称。

2. 在文本编辑器中打开 /config/cps-sink-connector.properties 文件。为以下属性添加值，它们在注释中标记为 "TODO"：

   topics=KAFKA_TOPICS
   cps.project=PROJECT_ID
   cps.topic=PUBSUB_TOPIC
   

   请替换以下内容：

   • KAFKA_TOPICS：要读取的 Kafka 主题的逗号分隔列表。
   • PROJECT_ID：您的 Pub/Sub 主题所属的 Google Cloud 项目。
   • PUBSUB_TOPIC：从 Kafka 接收消息的 Pub/Sub 主题。

3. 从 Kafka 目录运行以下命令：

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-sink-connector.properties
   

4. 按照 Apache Kafka 快速入门中的步骤，将一些事件写入您的 Kafka 主题。

5. 使用 gcloud CLI 从 Pub/Sub 中读取事件。

   gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack
   

将消息从 Pub/Sub 转发到 Kafka

本部分介绍如何启动源连接器、将消息发布到 Pub/Sub，以及读取从 Kafka 转发的消息。

1. 使用 gcloud CLI 创建包含订阅的 Pub/Sub 主题。

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC
   

   请替换以下内容：

   • PUBSUB_TOPIC：Pub/Sub 主题的名称。
   • PUBSUB_SUBSCRIPTION：Pub/Sub 订阅的名称。

2. 在文本编辑器中打开名为 /config/cps-source-connector.properties 的文件。为以下属性添加值，它们在注释中标记为 "TODO"：

   kafka.topic=KAFKA_TOPIC
   cps.project=PROJECT_ID
   cps.subscription=PUBSUB_SUBSCRIPTION
   

   请替换以下内容：

   • KAFKA_TOPIC：用于接收 Pub/Sub 消息的 Kafka 主题。
   • PROJECT_ID：您的 Pub/Sub 主题所属的 Google Cloud 项目。
   • PUBSUB_TOPIC：Pub/Sub 主题。

3. 从 Kafka 目录运行以下命令：

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-source-connector.properties
   

4. 使用 gcloud CLI 向 Pub/Sub 发布消息。

   gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"
   

5. 从 Kafka 读取消息。按照 Apache Kafka 快速入门中的步骤读取 Kafka 主题中的消息。

短信转化

Kafka 记录包含一个键和一个值，它们是可变长度的字节数组。（可选）Kafka 记录还可以具有标头，标头是键值对。Pub/Sub 消息包含两个主要部分：消息正文以及零个或零个以上的键值对属性。

Kafka Connect 使用转换器对传入和传出 Kafka 的键和值进行序列化。如需控制序列化，请在连接器配置文件中设置以下属性：

• key.converter：用于对记录键进行序列化的转换器。
• value.converter：用于对记录值进行序列化的转换器。

Pub/Sub 消息的正文是 ByteString 对象，因此最高效的转换是直接复制载荷。因此，我们建议您尽可能使用可生成原始数据类型（整数、浮点数、字符串或字节架构）的转换器，以防止对同一消息正文进行反序列化和重新序列化。

从 Kafka 到 Pub/Sub 的转换

接收器连接器将 Kafka 记录转换为 Pub/Sub 消息，如下所示：

• Kafka 记录键存储为 Pub/Sub 消息中名为 "key" 的特性。
• 默认情况下，连接器会丢弃 Kafka 记录中的所有标头。但是，如果您将 headers.publish 配置选项设置为 true，则连接器会将标头作为 Pub/Sub 属性写入。连接器会跳过任何超出 Pub/Sub 消息属性限制的标头。
• 对于整数、浮点、字符串和字节架构，连接器会将 Kafka 记录值的字节直接传递到 Pub/Sub 消息正文中。
• 对于结构体架构，连接器将每个字段写入 Pub/Sub 消息的属性。例如，如果该字段为 { "id"=123 }，则生成的 Pub/Sub 消息具有 "id"="123" 属性。字段值始终转换为字符串。系统不支持将映射和结构体类型用作结构体中的字段类型。
• 对于映射架构，连接器将每个键值对作为 Pub/Sub 消息的属性写入。例如，如果映射为 {"alice"=1,"bob"=2}，则生成的 Pub/Sub 消息具有两个属性："alice"="1" 和 "bob"="2"。键和值将转换为字符串。

结构体架构和映射架构还有一些其他行为：

• （可选）您可以通过设置 messageBodyName 配置属性，将特定结构体字段或映射键指定为消息正文。字段或键的值以 ByteString 形式存储在消息正文中。如果您未设置 messageBodyName，则结构体和映射架构的消息正文为空。

• 对于数组值，连接器仅支持基元数组类型。数组中的值序列会串联为单个 ByteString 对象。

从 Pub/Sub 到 Kafka 的转换

源连接器按如下方式将 Pub/Sub 消息转换为 Kafka 记录：

• Kafka 记录密钥：默认情况下，密钥设置为 null。或者，您可以通过设置 kafka.key.attribute 配置选项来指定要用作键的 Pub/Sub 消息属性。在这种情况下，连接器会查找具有该名称的属性，并将记录键设置为属性值。如果指定的属性不存在，则记录键设置为 null。

• Kafka 记录值。连接器按如下方式写入记录值：

  • 如果 Pub/Sub 消息没有自定义属性，则连接器使用 value.converter 指定的转换器将 Pub/Sub 消息正文以 byte[] 类型直接写入 Kafka 记录值。

  • 如果 Pub/Sub 消息具有自定义属性，并且 kafka.record.headers 为 false，则连接器会将结构体写入记录值。该结构体中，每个属性对应一个字段，以及一个名为 "message" 的字段，其值为 Pub/Sub 消息正文（以字节形式存储）：

    {
      "message": "<Pub/Sub message body>",
      "<attribute-1>": "<value-1>",
      "<attribute-2>": "<value-2>",
      ....
    }
    

    在这种情况下，您必须使用与 struct 架构兼容的 value.converter，例如 org.apache.kafka.connect.json.JsonConverter。

  • 如果 Pub/Sub 消息具有自定义属性且 kafka.record.headers 为 true，则连接器会将这些属性写入 Kafka 记录标头。它使用 value.converter 指定的转换器，将 Pub/Sub 消息正文以 byte[] 类型直接写入 Kafka 记录值。

• Kafka 记录标头。默认情况下，标头为空，除非您将 kafka.record.headers 设置为 true。

注意 ：在 Kafka Broker 配置中，将 message.max.bytes 设置为足以处理 Pub/Sub 消息。Pub/Sub 具有消息大小上限。

配置选项

除了 Kafka Connect API 提供的配置之外，Pub/Sub Group Kafka Connector 还支持以下配置。

接收器连接器配置选项

接收器连接器支持以下配置选项。

设置	数据类型	说明
connector.class	String	必填。连接器的 Java 类。对于 Pub/Sub 接收器连接器，该值必须为 com.google.pubsub.kafka.sink.CloudPubSubSinkConnector。
cps.endpoint	String	要使用的 Pub/Sub 端点。 默认值："pubsub.googleapis.com:443"。
cps.project	String	必需。包含 Pub/Sub 主题的 Google Cloud。
cps.topic	String	必需。将 Kafka 记录发布到的 Pub/Sub 主题。
gcp.credentials.file.path	String	可选。指向存储用于对 Pub/Sub Lite 进行身份验证的 Google Cloud 凭据的文件的路径。
gcp.credentials.json	String	可选。包含 Google Cloud 的 JSON blob，用于对 Pub/Sub Lite 进行身份验证。
headers.publish	Boolean	如果为 true，请将所有 Kafka 记录标头添加为 Pub/Sub 消息特性。 默认值：false。
maxBufferBytes	Long	在将主题 Kafka 分区发布到 Pub/Sub 之前，要在主题 Kafka 分区上接收的最大字节数。 默认值：10000000。
maxBufferSize	Integer	在将 Kafka 主题分区发布到 Pub/Sub 之前，要在 Kafka 主题分区上接收的记录数上限。 默认值：100。
maxDelayThresholdMs	Integer	在将未完成的记录发布到 Pub/Sub 之前等待达到 maxBufferSize 或 maxBufferBytes 的最长时间（以毫秒为单位）。 默认值：100。
maxOutstandingMessages	Long	在发布者阻止进一步发布之前，待处理的记录（包括未完成和待处理批次）的数量上限。 默认值：Long.MAX_VALUE。
maxOutstandingRequestBytes	Long	在发布者阻止进一步发布之前，未完成的总字节数（包括未完成和待处理批次）的最大字节数。 默认值：Long.MAX_VALUE。
maxRequestTimeoutMs	Integer	向 Pub/Sub 发出的单个发布请求的超时（以毫秒为单位）。 默认值：10000。
maxTotalTimeoutMs	Integer	用于发布到 Pub/Sub 的调用（包括重试）的总超时时间（以毫秒为单位）。 默认值：60000。
metadata.publish	Boolean	如果为 true，请将 Kafka 主题、分区、偏移量和时间戳添加为 Pub/Sub 消息特性。 默认值：false。
messageBodyName	String	使用结构体或映射值架构时，请指定要用作 Pub/Sub 消息正文的字段或键的名称。请参阅从 Kafka 到 Pub/Sub 的转换。 默认值："cps_message_body"。
orderingKeySource	String	指定如何在 Pub/Sub 消息中设置排序键。可以是下列值之一： none：不要设置排序键。 key：使用 Kafka 记录键作为排序键。 partition：将分区号（已转换为字符串）用作排序键。请仅对低吞吐量主题或具有数千个分区的主题使用此设置。 默认值：none。
topics	String	必需。要读取的 Kafka 主题的逗号分隔列表。

来源连接器配置选项

源连接器支持以下配置选项。

设置	数据类型	说明
connector.class	String	必填。连接器的 Java 类。对于 Pub/Sub 源连接器，该值必须为 com.google.pubsub.kafka.source.CloudPubSubSourceConnector。
cps.endpoint	String	要使用的 Pub/Sub 端点。 默认值："pubsub.googleapis.com:443"。
cps.makeOrderingKeyAttribute	Boolean	如果为 true，则使用与 Pub/Sub 消息属性相同的格式将排序键写入 Kafka 记录。请参阅从 Pub/Sub 到 Kafka 记录的转换。 默认值：false。
cps.maxBatchSize	Integer	发送到 Pub/Sub 的每个拉取请求要批处理的消息数量上限。 默认值：100
cps.project	String	必填。包含 Pub/Sub 主题的 Google Cloud 项目。
cps.subscription	String	必需。要从中拉取消息的 Pub/Sub 订阅的名称。
gcp.credentials.file.path	String	可选。指向存储用于对 Pub/Sub Lite 进行身份验证的 Google Cloud 凭据的文件的路径。
gcp.credentials.json	String	可选。包含 Google Cloud 的 JSON blob，用于对 Pub/Sub Lite 进行身份验证。
kafka.key.attribute	String	用作发布到 Kafka 的消息键的 Pub/Sub 消息属性。如果设置为 "orderingKey"，则使用消息的排序键。如果为 null，则表示 Kafka 记录没有密钥。 默认值：null。
kafka.partition.count	Integer	发布消息的 Kafka 主题的 Kafka 分区数。如果分区架构为 "kafka_partitioner"，则忽略此参数。 默认值：1。
kafka.partition.scheme	String	用于将消息分配给 Kafka 中的分区的方案。可以是以下值之一： round_robin：以轮循方式分配分区。 hash_key：通过对记录键进行哈希处理来查找分区。 hash_value：通过对记录值进行哈希处理来查找分区。 kafka_partitioner：将分区逻辑委托给 Kafka 生产者。默认情况下，Kafka 生产者会自动检测分区数量，并执行基于 murmur 哈希的分区映射或轮循，具体取决于是否提供记录键。 ordering_key：使用消息排序键的哈希代码。如果没有排序键，请使用 round_robin。 默认值：round_robin。
kafka.record.headers	Boolean	如果为 true，则将 Pub/Sub 消息特性写入 Kafka 标头。
kafka.topic	String	必需。从 Pub/Sub 接收消息的 Kafka 主题。

获取支持

如果您需要帮助，请创建支持服务工单。 对于一般性问题和讨论，请在 GitHub 代码库中创建问题。

后续步骤

• 了解 Kafka 和 Pub/Sub 之间的差异。
• 详细了解 Pub/Sub Group Kafka Connector。
• 请参阅 Pub/Sub Group Kafka Connector GitHub 代码库。