6.4. 部署 Kafka 连接

流程

1. 通过在 .spec.build.output 中指定容器 registry 来配置 KafkaConnect 自定义资源，并在 .spec.build.plugins 中指定其他连接器：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 1
     #...
     build:
       output: 2
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 3
         - name: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
               sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
               sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
     #...

   1

   Kafka Connect 集群的规格。

   2

   （必需）推送新镜像的容器 registry 的配置。

   3

   （必需）添加到新容器镜像的连接器插件及其工件列表。每个插件必须配置至少一个 工件。

2. 创建或更新资源：

   $ oc apply -f <kafka_connect_configuration_file>

3. 等待新容器镜像构建，并且部署 Kafka Connect 集群。
4. 使用 Kafka Connect REST API 或 KafkaConnector 自定义资源使用您添加的连接器插件。

流程

1. 使用 registry.redhat.io/amq-streams/kafka-34-rhel8:2.4.0 作为基础镜像创建新的 Dockerfile ：

   FROM registry.redhat.io/amq-streams/kafka-34-rhel8:2.4.0
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   插件文件示例

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   COPY 命令指向要复制到容器镜像的插件文件。

   本例为 Debezium 连接器(MongoDB、MySQL 和 PostgreSQL)添加了插件，但并非所有文件都被列为 brevity。在 Kafka Connect 中运行的 Debezium 与任何其他 Kafka Connect 任务相同。

2. 构建容器镜像。
3. 将自定义镜像推送到容器 registry。

4. 指向新容器镜像。

   您可以使用以下方法之一指向镜像：

   • 编辑 KafkaConnect 自定义资源的 KafkaConnect.spec.image 属性。

     如果设置，此属性会覆盖 Cluster Operator 中的 STRIMZI_KAFKA_CONNECT_IMAGES 环境变量。

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 1
       #...
       image: my-new-container-image 2
       config: 3
         #...

     1

     Kafka Connect 集群的规格。

     2

     pod 的 docker 镜像。

     3

     配置 Kafka Connect worker （而不是连接器）。

   • 编辑 install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 文件中的 STRIMZI_KAFKA_CONNECT_IMAGES 环境变量，然后重新安装 Cluster Operator。

流程

1. 使用以下方法之一将 FileStreamSourceConnector 和 FileStreamSinkConnector 插件添加到 Kafka Connect 中：

   • 配置 Kafka Connect 以自动使用插件构建新容器镜像
   • 从基本 Kafka Connect 镜像创建 Docker 镜像 （手动或使用持续集成）

2. 在 Kafka Connect 配置中，将 strimzi.io/use-connector-resources 注解设置为 true。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true"
   spec:
       # ...

   启用 KafkaConnector 资源后，Cluster Operator 会监视它们。

3. 编辑 examples/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
     autoRestart: 5
       enabled: true
     config: 6
       file: "/opt/kafka/LICENSE" 7
       topic: my-topic 8
       # ...

   1

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

   2

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

   3

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

   4

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

   5

   启用自动重启失败的连接器和任务。

   6

   连接器配置 作为键值对。

   7

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

   8

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

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

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

5. 创建 examples/connect/sink-connector.yaml 文件：

   touch examples/connect/sink-connector.yaml

6. 将以下 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

   1

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

   2

   连接器配置 作为键值对。

   3

   用于将源数据发布到的临时文件。

   4

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

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

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

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

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   将 <my_connect_cluster> 替换为 Kafka Connect 集群的名称。

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

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

   将 <my_kafka_cluster> 替换为 Kafka 集群的名称。

流程

1. 查找控制您要重启的 Kafka 连接器的 KafkaConnector 自定义资源的名称：

   oc get KafkaConnector

2. 通过在 OpenShift 中注解 KafkaConnector 资源来重启连接器。

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart=true

   restart 注解设置为 true。

3. 等待下一个协调发生（默认为两分钟）。

   Kafka 连接器会重启，只要协调过程检测到注解。当 Kafka Connect 接受重启请求时，注解会从 KafkaConnector 自定义资源中删除。

流程

1. 查找控制您要重启的 Kafka 连接器任务的 KafkaConnector 自定义资源的名称：

   oc get KafkaConnector

2. 从 KafkaConnector 自定义资源查找要重启的任务 ID。任务 ID 是非负整数，从 0 开始：

   oc describe KafkaConnector <kafka_connector_name>

3. 通过在 OpenShift 中注解 KafkaConnector 资源，使用 ID 重启连接器任务：

   oc annotate KafkaConnector <kafka_connector_name> strimzi.io/restart-task=0

   在本例中，任务 0 被重启。

4. 等待下一个协调发生（默认为两分钟）。

   只要协调过程检测到注解，Kafka 连接器任务会被重启。当 Kafka Connect 接受重启请求时，注解会从 KafkaConnector 自定义资源中删除。

1. 使用配置部署 KafkaConnector 资源，以创建您的连接器实例。
2. 通过将 strimzi.io/use-connector-resources 注解设置为 true，在 Kafka Connect 配置中启用 KafkaConnector 资源。