9.7. Kafka MirrorMaker 2 の設定

手順

1. 2 つの Kafka リソースを設定します。1 つはソース Kafka クラスターを保護するためのもので、もう 1 つはターゲット Kafka クラスターを保護するためのものです。

   認証用のリスナー設定を追加し、認可を有効にすることができます。

   この例の場合、内部リスナーは mTLS 暗号化と認証を使用して Kafka クラスター用に設定されています。Kafka の simple 認可が有効になっています。

   TLS 暗号化と mTLS 認証を使用したソース Kafka クラスター設定の例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.7.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.7"
       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 クラスター設定の例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.7.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.7"
       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. 別の namespace で Kafka リソースを作成または更新します。

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

   Cluster Operator はリスナーを作成し、クラスターおよびクライアント認証局 (CA) 証明書を設定して Kafka クラスター内で認証を有効にします。

   証明書は、シークレット <cluster_name>-cluster-ca-cert に作成されます。

3. 2 つの KafkaUser リソースを設定します。1 つはソース Kafka クラスターのユーザー用で、もう 1 つはターゲット Kafka クラスターのユーザー用です。

   1. 対応するソースおよびターゲットの Kafka クラスターと同じ認証および認可タイプを設定します。たとえば、ソース Kafka クラスターの Kafka 設定で tls 認証と simple 認可タイプを使用した場合は、KafkaUser 設定でも同じものを使用します。
   2. MirrorMaker 2 に必要な ACL を設定して、ソースおよびターゲットの Kafka クラスターでの操作を許可します。

   mTLS 認証のソースユーザー設定の例

   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 認証のターゲットユーザー設定の例

   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 クラスター用に作成した各 namespace で、KafkaUser リソースを作成または更新します。

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

   User Operator はクライアント (MirrorMaker) に対応するユーザーを作成すると共に、選択した認証タイプに基づいて、クライアント認証に使用されるセキュリティークレデンシャルを作成します。

   User Operator は、KafkaUser リソースと同じ名前の新しいシークレットを作成します。シークレットには、mTLS 認証用の秘密鍵と公開鍵が含まれています。公開鍵は、クライアント CA によって署名されたユーザー証明書に含まれます。

5. ソースおよびターゲットの Kafka クラスターに接続するための認証の詳細を使用して KafkaMirrorMaker2 リソースを設定します。

   TLS 暗号化と mTLS 認証を使用した MirrorMaker 2 設定の例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.7.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
               certificate: ca.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
               certificate: ca.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 証明書。それらが別の namespace にある場合は、Kafka クラスターの namespace からクラスターシークレットをコピーします。

   2

   TLS mechanism を使用してソース Kafka クラスターにアクセスするためのユーザー認証。

   3

   ターゲット Kafka クラスターの TLS 証明書。

   4

   ターゲット Kafka クラスターにアクセスするためのユーザー認証。

6. ターゲット Kafka クラスターと同じ namespace で KafkaMirrorMaker2 リソースを作成または更新します。

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

手順

1. 再起動する MirrorMaker 2 コネクターを制御する KafkaMirrorMaker2 カスタムリソースの名前を見つけます。

   oc get KafkaMirrorMaker2

2. KafkaMirrorMaker2 リソースを編集して、コネクターを停止または一時停止します。

   MirrorMaker 2 コネクターを停止するための設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.7.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 カスタムリソースの名前を見つけます。

   oc get KafkaMirrorMaker2

2. KafkaMirrorMaker2 カスタムリソースから再起動する Kafka MirrorMaker 2 コネクターの名前を見つけます。

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

3. コネクターの名前を使用して、OpenShift の KafkaMirrorMaker2 リソースにアノテーションを付けてコネクターを再起動します。

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

   この例では、my-mirror-maker-2 クラスター内の my-connector コネクターが再起動されます。

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

4. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

   アノテーションが調整プロセスによって検出されているかぎり、MirrorMaker 2 コネクターは再起動されます。MirrorMaker 2 がリクエストを受け入れると、アノテーションは KafkaMirrorMaker2 カスタムリソースから削除されます。

手順

1. 再起動する MirrorMaker 2 コネクタータスクを制御する KafkaMirrorMaker2 カスタムリソースの名前を見つけます。

   oc get KafkaMirrorMaker2

2. KafkaMirrorMaker2 カスタムリソースから、コネクターの名前と再起動するタスクの ID を検索します。

   oc describe KafkaMirrorMaker2 <mirrormaker_cluster_name>

   タスク ID は 0 から始まる負の値ではない整数です。

3. 名前と ID を使用して、OpenShift の KafkaMirrorMaker2 リソースにアノテーションを付けてコネクタータスクを再開します。

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

   この例では、my-mirror-maker-2 クラスター内の my-connector コネクターのタスク 0 が再起動されます。

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

4. 次の調整が発生するまで待ちます (デフォルトでは 2 分ごとです)。

   アノテーションが調整プロセスによって検出されているかぎり、MirrorMaker 2 コネクタータスクが再起動されます。MirrorMaker 2 がリクエストを受け入れると、アノテーションは KafkaMirrorMaker2 カスタムリソースから削除されます。