2.3. 将 Kafka 主题连接到 Kamelet Binding 中的数据接收器

要将 Kafka 主题连接到数据接收器，您可以创建一个 Kamelet Binding，如图 2.3 所示。

Connecting a Kafka topic to a data sink 图 2.3 将 Kafka 主题连接到数据接收器

先决条件

您知道您要从中发送事件的 Kafka 主题的名称。此流程中的示例使用 test-topic 来发送事件。您用来从将数据源连接到 Kamelet Binding 中的 Kafka 主题中的 coffee 源中接收事件的主题相同。
您知道 Kafka 实例的以下参数值：
- bootstrapServers - 以逗号分隔的 Kafka Broker URL 列表。
- password - Kafka 验证的密码。
- user - 向 Kafka 进行身份验证的用户名。
您知道与 Kafka 代理通信的安全协议。对于 AMQ 流上的 Kafka 集群，它是 PLAINTEXT。
您知道您要添加到 Camel K 集成和所需的实例参数中的 Kamelets。此流程的示例 Kamelets 在 Kamelet Catalog 中提供：
- kafka-source Kamelet - 使用 kafka-source Kamelet，因为 Kafka 主题在此绑定中发送数据（这是数据制作者）。所需的参数的值示例有：
  - bootstrapServers - "broker.url:9092"
  - password - "testpassword"
  - user - "testuser"
  - topic - "test-topic"
  - securityProtocol - 对于 AMQ 流上的 Kafka 集群，这个参数值为 "PLAINTEXT "。
- log-sink Kamelet - 使用 log-sink 来记录它从 kafka-source Kamelet 接收的数据。（可选）指定 showStreams 参数，以显示数据的消息正文。log-sink Kamelet 可用于调试目的。

流程

要将 Kafka 主题连接到数据接收器，请创建一个 Kamelet Binding：

在您选择的编辑器中，使用以下基本结构创建一个 YAML 文件：

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name:
spec:
  source:
  sink:

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name:
spec:
  source:
  sink:

Copy to Clipboard

Toggle word wrap

为 Kamelet Binding 添加一个名称。在本例中，名称是 kafka-to-log，因为绑定将 kafka-source Kamelet 连接到 log-sink Kamelet。

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
  sink:

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
  sink:

Copy to Clipboard

Toggle word wrap

对于 Kamelet Binding 的源，指定 kafka-source Kamelet 并配置其参数。

例如，当 Kafka 集群位于 AMQ Streams 中时，必须将 securityProtocol 参数设置为 "PLAINTEXT" ：

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT"
  sink:

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT"
  sink:

Copy to Clipboard

Toggle word wrap

对于 Kamelet Binding 的接收器，指定数据消费者 Kamelet （例如，log-sink Kamelet）并为 Kamelet 配置任何参数，例如：

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT" // only for AMQ streams
  sink:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: log-sink
    properties:
      showStreams: true

apiVersion: camel.apache.org/v1alpha1
kind: KameletBinding
metadata:
  name: kafka-to-log
spec:
  source:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: kafka-source
    properties:
      bootstrapServers: "broker.url:9092"
      password: "testpassword"
      topic: "test-topic"
      user: "testuser"
      securityProtocol: "PLAINTEXT" // only for AMQ streams
  sink:
    ref:
      kind: Kamelet
      apiVersion: camel.apache.org/v1alpha1
      name: log-sink
    properties:
      showStreams: true

Copy to Clipboard

Toggle word wrap

保存 YAML 文件（如 kafka-to-log.yaml）。
登录您的 OpenShift 项目。
将 Kamelet Binding 作为资源添加到 OpenShift 命名空间中：
oc apply -f <kamelet binding filename>
例如：
oc apply -f kafka-to-log.yaml
Camel K operator 使用 KameletBinding 资源生成并运行 Camel K 集成。构建可能需要几分钟时间。
查看 KameletBinding 资源的状态：
oc get kameletbindings
查看其集成的状态：
oc get integrations

查看集成日志：

kamel logs <integration> -n <project>

例如：

kamel logs kafka-to-log -n my-camel-k-kafka

在输出中，您应该看到 coffee 事件，例如：

INFO  [log-sink-E80C5C904418150-0000000000000001] (Camel (camel-1) thread #0 - timer://tick) {"id":7259,"uid":"a4ecb7c2-05b8-4a49-b0d2-d1e8db5bc5e2","blend_name":"Postmodern Symphony","origin":"Huila, Colombia","variety":"Kona","notes":"delicate, chewy, black currant, red apple, star fruit","intensifier":"balanced"}

INFO  [log-sink-E80C5C904418150-0000000000000001] (Camel (camel-1) thread #0 - timer://tick) {"id":7259,"uid":"a4ecb7c2-05b8-4a49-b0d2-d1e8db5bc5e2","blend_name":"Postmodern Symphony","origin":"Huila, Colombia","variety":"Kona","notes":"delicate, chewy, black currant, red apple, star fruit","intensifier":"balanced"}

Copy to Clipboard

Toggle word wrap

要停止正在运行的集成，请删除关联的 Kamelet Binding 资源：
oc delete kameletbindings/<kameletbinding-name>
例如：
oc delete kameletbindings/kafka-to-log

另请参阅

2.3. 将 Kafka 主题连接到 Kamelet Binding 中的数据接收器

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links