9.9. 配置 Kafka MirrorMaker 2

流程

1. 配置两个 Kafka 资源，一个用于保护源 Kafka 集群，另一个用于保护目标 Kafka 集群。

   您可以为身份验证和启用授权添加监听程序配置。

   在本例中，为带有 TLS 加密和 mTLS 身份验证的 Kafka 集群配置了内部监听程序。启用 Kafka 简单 授权。

   使用 TLS 加密和 mTLS 身份验证的源 Kafka 集群配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.8.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.8"
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   使用 TLS 加密和 mTLS 身份验证的目标 Kafka 集群配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.8.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.8"
       storage:
         type: jbod
         volumes:
           - id: 0
             type: persistent-claim
             size: 100Gi
             deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 在单独的命名空间中创建或更新 Kafka 资源。

   Copy to Clipboard Toggle word wrap

   oc apply -f <kafka_configuration_file> -n <namespace>

   Cluster Operator 创建监听程序并设置集群和客户端证书颁发机构(CA)证书，以便在 Kafka 集群中启用身份验证。

   证书在 secret < cluster_name> -cluster-ca-cert 中创建。

3. 配置两个 KafkaUser 资源，一个用于源 Kafka 集群的用户，另一个用于目标 Kafka 集群的用户。

   1. 配置与对应的源和目标 Kafka 集群相同的身份验证和授权类型。例如，如果您在源 Kafka 集群的 Kafka 配置中使用了 tls 验证和 simple 授权类型，请在 KafkaUser 配置中使用相同的。
   2. 配置 MirrorMaker 2 所需的 ACL，以允许对源和目标 Kafka 集群执行操作。

   mTLS 验证的源用户配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-source-user
     labels:
       strimzi.io/cluster: my-source-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # MirrorSourceConnector
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Create
             - DescribeConfigs
             - Read
             - Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - DescribeConfigs
             - Read
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operations:
             - Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Read

   mTLS 验证的目标用户配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-target-user
     labels:
       strimzi.io/cluster: my-target-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # cluster group
         - resource:
             type: group
             name: mirrormaker2-cluster
           operations:
             - Read
         # access to config.storage.topic
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # access to status.storage.topic
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # access to offset.storage.topic
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - Create
             - Alter
             - AlterConfigs
             - Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operations:
             - Create
             - Describe
             - Read
             - Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operations:
             - Read
             - Describe
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operations:
             - Create
             - Describe
             - Write

   注意

   您可以通过将 type 设置为 tls-external 来使用 User Operator 外部发布的证书。如需更多信息，请参阅 KafkaUserSpec 模式参考。

4. 在您为源和目标 Kafka 集群创建的每个命名空间中创建或更新 KafkaUser 资源。

   Copy to Clipboard Toggle word wrap

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   User Operator 根据所选的验证类型创建代表客户端(MirrorMaker)的用户，以及用于客户端身份验证的安全凭证。

   User Operator 创建一个名称与 KafkaUser 资源相同的新 secret。secret 包含 mTLS 验证的私钥和公钥。公钥包含在用户证书中，该证书由客户端 CA 签名。

5. 使用身份验证详情配置 KafkaMirrorMaker2 资源，以连接到源和目标 Kafka 集群。

   带有 TLS 加密和 mTLS 身份验证的 MirrorMaker 2 配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.8.0
     replicas: 1
     connectCluster: "my-target-cluster"
     clusters:
       - alias: "my-source-cluster"
         bootstrapServers: my-source-cluster-kafka-bootstrap:9093
         tls: 

   1

   
           trustedCertificates:
             - secretName: my-source-cluster-cluster-ca-cert
               pattern: "*.crt"
         authentication: 

   2

   
           type: tls
           certificateAndKey:
             secretName: my-source-user
             certificate: user.crt
             key: user.key
       - alias: "my-target-cluster"
         bootstrapServers: my-target-cluster-kafka-bootstrap:9093
         tls: 

   3

   
           trustedCertificates:
             - secretName: my-target-cluster-cluster-ca-cert
               pattern: "*.crt"
         authentication: 

   4

   
           type: tls
           certificateAndKey:
             secretName: my-target-user
             certificate: user.crt
             key: user.key
         config:
           # -1 means it will use the default replication factor configured in the broker
           config.storage.replication.factor: -1
           offset.storage.replication.factor: -1
           status.storage.replication.factor: -1
     mirrors:
       - sourceCluster: "my-source-cluster"
         targetCluster: "my-target-cluster"
         sourceConnector:
           config:
             replication.factor: 1
             offset-syncs.topic.replication.factor: 1
             sync.topic.acls.enabled: "false"
         heartbeatConnector:
           config:
             heartbeats.topic.replication.factor: 1
         checkpointConnector:
           config:
             checkpoints.topic.replication.factor: 1
             sync.group.offsets.enabled: "true"
         topicsPattern: "topic1|topic2|topic3"
         groupsPattern: "group1|group2|group3"

   1

   源 Kafka 集群的 TLS 证书。如果它们位于单独的命名空间中，请将集群 secret 从 Kafka 集群的命名空间中复制。

   2

   使用 TLS 机制访问源 Kafka 集群的用户身份验证。

   3

   目标 Kafka 集群的 TLS 证书。

   4

   访问目标 Kafka 集群的用户身份验证。

6. 在与目标 Kafka 集群相同的命名空间中创建或更新 KafkaMirrorMaker2 资源。

   Copy to Clipboard Toggle word wrap

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>

流程

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

   Copy to Clipboard Toggle word wrap

   oc get KafkaMirrorMaker2

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

   停止 MirrorMaker 2 连接器的配置示例

   Copy to Clipboard Toggle word wrap

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.8.0
     replicas: 3
     connectCluster: "my-cluster-target"
     clusters:
       # ...
     mirrors:
     - sourceCluster: "my-cluster-source"
       targetCluster: "my-cluster-target"
       sourceConnector:
         tasksMax: 10
         autoRestart:
           enabled: true
         state: stopped
     # ...

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

3. 将更改应用到 KafkaMirrorMaker2 配置。

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

流程

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

   Copy to Clipboard Toggle word wrap

   oc get KafkaMirrorMaker2

2. 查找要从 KafkaMirrorMaker2 自定义资源重启的 Kafka MirrorMaker 2 连接器的名称：

   Copy to Clipboard Toggle word wrap

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

3. 通过在 OpenShift 中注解 KafkaMirrorMaker2 资源，使用连接器的名称来重启连接器：

   Copy to Clipboard Toggle word wrap

   oc annotate KafkaMirrorMaker2 <mirrormaker_cluster_name> "strimzi.io/restart-connector=<mirrormaker_connector_name>"

   在本例中，my-mirror-maker-2 集群中的连接器 my-connector 被重启：

   Copy to Clipboard Toggle word wrap

   oc annotate KafkaMirrorMaker2 my-mirror-maker-2 "strimzi.io/restart-connector=my-connector"

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

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

流程

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

   Copy to Clipboard Toggle word wrap

   oc get KafkaMirrorMaker2

2. 查找连接器的名称和要从 KafkaMirrorMaker2 自定义资源重启的任务 ID：

   Copy to Clipboard Toggle word wrap

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

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

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

   Copy to Clipboard Toggle word wrap

   oc annotate KafkaMirrorMaker2 <mirrormaker_cluster_name> "strimzi.io/restart-connector-task=<mirrormaker_connector_name>:<task_id>"

   在本例中，在 my-mirror-maker-2 集群中连接器 my-connector 的任务 0 被重启：

   Copy to Clipboard Toggle word wrap

   oc annotate KafkaMirrorMaker2 my-mirror-maker-2 "strimzi.io/restart-connector-task=my-connector:0"

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

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