第 5 章 针对 Kafka 验证 Camel K
您可以针对 Apache Kafka 验证 Camel K。
以下示例演示了如何设置 Kafka 主题,并在简单的 Producer/Consumer 模式集成中使用它。
5.1. 设置 Kafka
要设置 Kafka,您必须:
- 安装所需的 OpenShift operator
- 创建 Kafka 实例
- 创建 Kafka 主题
使用以下提到的红帽产品来设置 Kafka:
- Red Hat Advanced 消息队列(AMQ)流 - 自我管理的 Apache Kafka 产品。AMQ Streams 基于开源 Strimzi,并作为 Red Hat Integration 的一部分包含在内。AMQ Streams 是一个分布式、可扩展的流平台,它基于 Apache Kafka,其中包括发布/订阅消息传递代理。Kafka Connect 提供了一个框架,用于将基于 Kafka 的系统与外部系统集成。使用 Kafka Connect,您可以配置源和接收器连接器,将来自外部系统的数据流传输到 Kafka 代理。
5.1.1. 使用 AMQ 流设置 Kafka
AMQ Streams 简化了在 OpenShift 集群中运行 Apache Kafka 的过程。
5.1.1.1. 为 AMQ Streams 准备 OpenShift 集群
要使用 Camel K 或 Kamelets 和 Red Hat AMQ Streams,您必须安装以下 operator 和工具:
- Red Hat Integration - AMQ Streams operator - 管理 Openshift Cluster 和 AMQ Streams for Apache Kafka 实例之间的通信。
- Red Hat Integration - Camel K operator - 安装和管理 Camel K - 在 OpenShift 上原生运行的轻量级集成框架。
- Camel K CLI 工具 - 允许您访问所有 Camel K 功能。
先决条件
- 熟悉 Apache Kafka 概念。
- 您可以使用正确访问级别访问 OpenShift 4.6 (或更新版本)集群、创建项目和安装操作器的功能,以及在本地系统上安装 OpenShift 和 Camel K CLI。
-
已安装 OpenShift CLI 工具(
oc
),以便可以在命令行中与 OpenShift 集群交互。
流程
使用 AMQ Streams 设置 Kafka:
- 登录您的 OpenShift 集群的 Web 控制台。
- 创建或打开您要在其中创建集成的项目,如 my-camel-k-kafka。
- 安装 Camel K operator 和 Camel K CLI,如 安装 Camel K 所述。
安装 AMQ Streams Operator:
- 从任何项目中,选择 Operators > OperatorHub。
- 在 Filter by Keyword 字段中,键入 AMQ Streams。
点 Red Hat Integration - AMQ Streams 卡,然后点 Install。
此时会打开 Install Operator 页面。
- 接受默认值,然后点 Install。
- 选择 Operators > Installed Operators 来验证是否安装了 Camel K 和 AMQ Streams operator。
5.1.1.2. 使用 AMQ Streams 设置 Kafka 主题
Kafka 主题提供在 Kafka 实例中存储数据的目标。在向它发送数据前,您必须设置 Kafka 主题。
先决条件
- 您可以访问 OpenShift 集群。
- 已安装 Red Hat Integration - Camel K 和 Red Hat Integration - AMQ Streams operator,如 准备 OpenShift 集群 中所述。
-
已安装 OpenShift CLI (
oc
)和 Camel K CLI (kamel
)。
流程
使用 AMQ Streams 设置 Kafka 主题:
- 登录您的 OpenShift 集群的 Web 控制台。
- 选择 Projects,然后单击在其中安装 Red Hat Integration - AMQ Streams operator 的项目。例如,单击 my-camel-k-kafka 项目。
- 选择 Operators > Installed Operators,然后点 Red Hat Integration - AMQ Streams。
创建 Kafka 集群:
- 在 Kafka 下,点 Create instance。
- 为集群输入一个名称,如 kafka-test。
接受其他默认值,然后单击 Create。
创建 Kafka 实例的过程可能需要几分钟时间来完成。
当状态就绪时,继续下一步。
创建 Kafka 主题:
- 选择 Operators > Installed Operators,然后点 Red Hat Integration - AMQ Streams。
- 在 Kafka 主题 下,点 Create Kafka Topic。
- 输入主题的名称,如 test-topic。
- 接受其他默认值,然后单击 Create。
5.1.2. 使用 OpenShift 流设置 Kafka
要使用 OpenShift Streams for Apache Kafka,您必须登录到您的红帽帐户。
5.1.2.1. 为 OpenShift Streams 准备 OpenShift 集群
要使用受管云服务,您必须安装以下 operator 和工具:
- OpenShift Application Services (RHOAS) CLI - 允许您从终端管理应用程序服务。
- Red Hat Integration - Camel K operator 安装并管理 Camel K - 在 OpenShift 上原生运行的轻量级集成框架。
- Camel K CLI 工具 - 允许您访问所有 Camel K 功能。
先决条件
- 熟悉 Apache Kafka 概念。
- 您可以使用正确访问级别访问 OpenShift 4.6 (或更新版本)集群、创建项目和安装操作器的功能,以及在本地系统上安装 OpenShift 和 Apache Camel K CLI。
-
已安装 OpenShift CLI 工具(
oc
),以便可以在命令行中与 OpenShift 集群交互。
流程
- 使用集群管理员帐户登录 OpenShift Web 控制台。
为您的 Camel K 或 Kamelets 应用程序创建 OpenShift 项目。
- 选择 Home > Projects。
- 点击 Create Project。
-
键入项目的名称,如
my-camel-k-kafka
,然后单击 Create。
- 下载并安装 RHOAS CLI,如 Getting started with the rhoas CLI 所述。
- 安装 Camel K operator 和 Camel K CLI,如 安装 Camel K 所述。
- 要验证是否安装了 Red Hat Integration - Camel K Operator,点 Operators > Installed Operators。
后续步骤
5.1.2.2. 使用 RHOAS 设置 Kafka 主题
Kafka 整理有关 主题的消息。每个主题都有一个名称。应用向主题发送消息并从主题检索消息。Kafka 主题提供在 Kafka 实例中存储数据的目标。在向它发送数据前,您必须设置 Kafka 主题。
先决条件
- 您可以使用正确访问级别访问 OpenShift 集群、创建项目和安装操作器,以及在本地系统上安装 OpenShift 和 Camel K CLI。
-
已安装 OpenShift CLI (
oc
)、Camel K CLI (kamel
)和 RHOAS CLI (rhoas
)工具,如 准备 OpenShift 集群 中所述。 - 已安装 Red Hat Integration - Camel K operator,如 准备 OpenShift 集群 中所述。
- 您已登录到 Red Hat Cloud 站点。
流程
设置 Kafka 主题:
- 在命令行中登录到您的 OpenShift 集群。
打开您的项目,例如:
oc project my-camel-k-kafka
验证 Camel K Operator 是否已安装到项目中:
oc get csv
结果列出了 Red Hat Camel K operator,并表示它处于
Succeeded
阶段。准备 Kafka 实例并将其连接到 RHOAS:
使用以下命令登录到 RHOAS CLI:
RHOAS 登录
创建一个 kafka 实例,如 kafka-test :
rhoas kafka create kafka-test
创建 Kafka 实例的过程可能需要几分钟时间来完成。
检查 Kafka 实例的状态:
RHOAS 状态
您还可以在 web 控制台中查看状态:
https://cloud.redhat.com/application-services/streams/kafkas/
当状态 就绪时,继续下一步。
创建新的 Kafka 主题:
rhoas kafka topic create --name test-topic
将 Kafka 实例(集群)与 Openshift Application Services 实例连接:
RHOAS 集群连接
按照获取凭证令牌的脚本说明进行操作。
您应该看到类似如下的输出:
Token Secret "rh-cloud-services-accesstoken-cli" created successfully Service Account Secret "rh-cloud-services-service-account" created successfully KafkaConnection resource "kafka-test" has been created KafkaConnection successfully installed on your cluster.
后续步骤
5.1.2.3. 获取 Kafka 凭证
要将应用程序或服务连接到 Kafka 实例,您必须首先获取以下 Kafka 凭证:
- 获取 bootstrap URL。
- 使用凭证(用户名和密码)创建服务帐户。
对于 OpenShift Streams,身份验证协议是 SASL_SSL。
前提条件
- 您已创建了 Kafka 实例,它处于 ready 状态。
- 您已创建了 Kafka 主题。
流程
获取 Kafka Broker URL (Bootstrap URL):
RHOAS 状态
这个命令返回类似如下的输出:
Kafka --------------------------------------------------------------- ID: 1ptdfZRHmLKwqW6A3YKM2MawgDh Name: my-kafka Status: ready Bootstrap URL: my-kafka--ptdfzrhmlkwqw-a-ykm-mawgdh.kafka.devshift.org:443
要获取用户名和密码,请使用以下语法创建服务帐户:
rhoas service-account create --name "<account-name>" --file-format json
注意在创建服务帐户时,您可以选择文件格式和位置来保存凭证。如需更多信息,请键入
rhoas service-account create --help
例如:
rhoas service-account create --name "my-service-acct" --file-format json
服务帐户已创建并保存到 JSON 文件中。
要验证您的服务帐户凭证,请查看
credentials.json
文件:cat credentials.json
这个命令返回类似如下的输出:
{"clientID":"srvc-acct-eb575691-b94a-41f1-ab97-50ade0cd1094", "password":"facf3df1-3c8d-4253-aa87-8c95ca5e1225"}
授予向 Kakfa 主题或从 Kakfa 发送和接收消息的权限。使用以下命令,其中
clientID
是credentials.json
文件中提供的值(从第 3 步中)。rhoas kafka acl grant-access --producer --consumer --service-account $CLIENT_ID --topic test-topic --group all
例如:
rhoas kafka acl grant-access --producer --consumer --service-account srvc-acct-eb575691-b94a-41f1-ab97-50ade0cd1094 --topic test-topic --group all
5.1.2.4. 使用 SASL/Plain 验证方法创建 secret
您可以使用您获取的凭证创建 secret (Kafka bootstrap URL、服务帐户 ID 和服务帐户 secret)。
流程
编辑
application.properties
文件并添加 Kafka 凭证。application.properties 文件
camel.component.kafka.brokers = <YOUR-KAFKA-BOOTSTRAP-URL-HERE> camel.component.kafka.security-protocol = SASL_SSL camel.component.kafka.sasl-mechanism = PLAIN camel.component.kafka.sasl-jaas-config=org.apache.kafka.common.security.plain.PlainLoginModule required username='<YOUR-SERVICE-ACCOUNT-ID-HERE>' password='<YOUR-SERVICE-ACCOUNT-SECRET-HERE>'; consumer.topic=<TOPIC-NAME> producer.topic=<TOPIC-NAME>
运行以下命令,在
application.properties
文件中创建一个包含敏感属性的 secret:oc create secret generic kafka-props --from-file application.properties
在运行 Camel K 集成时,您可以使用此 secret。
5.1.2.5. 使用 SASL/OAUTHBearer 身份验证方法创建 secret
您可以使用您获取的凭证创建 secret (Kafka bootstrap URL、服务帐户 ID 和服务帐户 secret)。
流程
编辑
application-oauth.properties
文件并添加 Kafka 凭证。application-oauth.properties file
camel.component.kafka.brokers = <YOUR-KAFKA-BOOTSTRAP-URL-HERE> camel.component.kafka.security-protocol = SASL_SSL camel.component.kafka.sasl-mechanism = OAUTHBEARER camel.component.kafka.sasl-jaas-config = org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \ oauth.client.id='<YOUR-SERVICE-ACCOUNT-ID-HERE>' \ oauth.client.secret='<YOUR-SERVICE-ACCOUNT-SECRET-HERE>' \ oauth.token.endpoint.uri="https://identity.api.openshift.com/auth/realms/rhoas/protocol/openid-connect/token" ; camel.component.kafka.additional-properties[sasl.login.callback.handler.class]=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler consumer.topic=<TOPIC-NAME> producer.topic=<TOPIC-NAME>
运行以下命令,在
application.properties
文件中创建一个包含敏感属性的 secret:oc create secret generic kafka-props --from-file application-oauth.properties
在运行 Camel K 集成时,您可以使用此 secret。