2.5. 将操作应用到 Kafka 连接中的数据

流程

1. 编辑 Kamelet Binding，使其包含 topic-name-matches-filter-action Kamelet 作为 source 和 sink Kamelets 之间的中间步骤。

   通常，您可以使用 kafka-source Kamelet 作为源 Kamelet，并提供一个主题作为所需 主题参数的值。

   在以下 Kamelet Binding 示例中，kafka-source Kamelet 指定 test-topic、test-topic-2 和 test-topic-3 Kafka 主题，以及 topic-name-matches-filter-action Kamelet 指定过滤来自 topic-test 主题的事件数据：

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

   Copy to Clipboard Toggle word wrap

   如果要过滤来自 kafka-source Kamelet 以外的源 Kamelet 的主题，您必须提供 Kafka 主题信息。您可以使用 insert-header-action Kamelet 将 Kafka 主题字段添加为中间步骤，在 Kamelet Binding 中的 topic-name-matches-filter-action 步骤前，如下例所示：

   apiVersion: camel.apache.org/v1alpha1
   kind: KameletBinding
   metadata:
     name: coffee-to-log-by-topic
   spec:
     source:
       ref:
         kind: Kamelet
         apiVersion: camel.apache.org/v1alpha1
         name: coffee-source
       properties:
         period: 5000
   steps:
     - ref:
         kind: Kamelet
         apiVersion: camel.apache.org/v1alpha1
         name: insert-header-action
       properties:
         name:  "KAFKA.topic"
         value:  "test-topic"
     - ref:
         kind: Kamelet
         apiVersion: camel.apache.org/v1alpha1
         name: topic-name-matches-filter-action
       properties:
         regex:  "test-topic"
   sink:
       ref:
         kind: Kamelet
         apiVersion: camel.apache.org/v1alpha1
         name: log-sink
       properties:
         showStreams: true

   Copy to Clipboard Toggle word wrap
2. 保存 Kamelet Binding YAML 文件。