5.2.5. 部署示例 KafkaConnector 资源

流程

1. 编辑 example /connect/source-connector.yaml 文件：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector 

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster 

   2

   
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 

   3

   
     tasksMax: 2 

   4

   
     config: 

   5

   
       file: "/opt/kafka/LICENSE" 

   6

   
       topic: my-topic 

   7

   
       # ...

   Copy to Clipboard Toggle word wrap

   1

   KafkaConnector 资源的名称，用作连接器的名称。使用对 OpenShift 资源有效的任何名称。

   2

   用于在其中创建连接器实例的 Kafka Connect 集群的名称。连接器必须部署到与其链接的 Kafka Connect 集群相同的命名空间。

   3

   连接器类的全名或别名.这应该存在于由 Kafka Connect 集群使用的镜像中。

   4

   连接器可以创建的 Kafka Connect 任务 的最大数量。

   5

   连接器配置，作为键值对.

   6

   这个示例源连接器配置从 /opt/kafka/LICENSE 文件中读取数据。

   7

   将源数据发布到的 Kafka 主题。

2. 在 OpenShift 集群中创建源 KafkaConnector ：

   oc apply -f examples/connect/source-connector.yaml

   Copy to Clipboard Toggle word wrap

3. 创建一个 示例/connect/sink-connector.yaml 文件：

   touch examples/connect/sink-connector.yaml

   Copy to Clipboard Toggle word wrap

4. 将以下 YAML 粘贴到 sink-connector.yaml 文件中：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 

   1

   
     tasksMax: 2
     config: 

   2

   
       file: "/tmp/my-file" 

   3

   
       topics: my-topic 

   4

   Copy to Clipboard Toggle word wrap

   1

   连接器类的全名或别名.这应该存在于由 Kafka Connect 集群使用的镜像中。

   2

   连接器配置，作为键值对.

   3

   发布源数据的临时文件。

   4

   从中读取源数据的 Kafka 主题。

5. 在 OpenShift 集群中创建 sink KafkaConnector ：

   oc apply -f examples/connect/sink-connector.yaml

   Copy to Clipboard Toggle word wrap

6. 检查是否创建了连接器资源：

   oc get kctr --selector strimzi.io/cluster=MY-CONNECT-CLUSTER -o name
   
   my-source-connector
   my-sink-connector

   Copy to Clipboard Toggle word wrap

   将 MY-CONNECT-CLUSTER 替换为您的 Kafka 连接集群。

7. 在容器中，执行 kafka-console-consumer.sh 读取由源连接器写入主题的消息：

   oc exec MY-CLUSTER-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server MY-CLUSTER-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   Copy to Clipboard Toggle word wrap