2.4. 在 Kamelet Binding 中将 Kafka 主题连接到一个数据接收器

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

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

为 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:

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

例如，当 Kafka 集群位于 OpenShift Streams 中时（您不需要设置 securityProtocol 参数）：

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"
  sink:

例如，当 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:

对于 Kamelet Binding 的 sink，请指定数据使用者 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

将 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"}

要停止正在运行的集成，请删除相关的 Kamelet Binding 资源：

oc delete kameletbindings/<kameletbinding-name>

例如：

oc delete kameletbindings/kafka-to-log