9.6. 配置 Kafka Connect

流程

1. 编辑 KafkaUser 资源中的 authorization 属性，为用户提供访问权限。

   使用 字面 名称值为 Kafka Connect 主题和集群组配置访问权限。下表显示了为主题和集群组 ID 配置的默认名称。

   表 9.2. 访问权限配置的名称

   属性	名称
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   在本例中，默认名称用于指定访问权限。如果您要为 Kafka Connect 实例使用不同的名称，请在 ACL 配置中使用这些名称。

   简单授权配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operations:
             - Create
             - Describe
             - Read
             - Write
           host: "*"
         # cluster group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

2. 创建或更新资源。

   oc apply -f KAFKA-USER-CONFIG-FILE

流程

1. 查找控制您要暂停或停止连接器的 KafkaConnector 自定义资源的名称：

   oc get KafkaConnector

2. 编辑 KafkaConnector 资源，以停止或暂停连接器。

   停止 Kafka Connect 连接器的配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector
     labels:
       strimzi.io/cluster: my-connect-cluster
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector
     tasksMax: 2
     config:
       file: "/opt/kafka/LICENSE"
       topic: my-topic
     state: stopped
     # ...

   将 state 配置更改为 stopped 或 paused。当此属性没有设置时，连接器的默认状态为 running。

3. 对 KafkaConnector 配置应用更改。

   您可以通过将 state 改为 running，或删除配置来恢复连接器。

流程

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

   oc get KafkaConnector

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

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

   重启 注解设置为 true。

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

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

流程

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

   oc get KafkaConnector

2. 查找从 KafkaConnector 自定义资源重启的任务 ID：

   oc describe KafkaConnector <kafka_connector_name>

   任务 ID 是非负整数，从 0 开始。

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

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

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

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

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