OpenShift での AMQ Streams のデプロイと管理

手順

1. Cluster Operator のインストール先の namespace を使用するように、AMQ Streams インストールファイルを編集します。

   たとえば、この手順では Cluster Operator は my-cluster-operator-namespace という namespace にインストールされます。

   Linux の場合は、以下を使用します。

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS の場合は、以下を使用します。

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. Cluster Operator をデプロイします。

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

3. デプロイメントのステータスを確認します。

   oc get deployments -n my-cluster-operator-namespace

   デプロイメント名と準備状態が表示されている出力

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

手順

1. Cluster Operator のインストール先の namespace を使用するように、AMQ Streams インストールファイルを編集します。

   たとえば、この手順では Cluster Operator は my-cluster-operator-namespace という namespace にインストールされます。

   Linux の場合は、以下を使用します。

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS の場合は、以下を使用します。

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml ファイルを編集し、Cluster Operator が監視するすべての namespace のリストを STRIMZI_NAMESPACE 環境変数に追加します。

   たとえば、この手順では Cluster Operator は watched-namespace-1、watched-namespace-2、および watched-namespace-3 という namespace を監視します。

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.1
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

3. リストした各 namespace に RoleBindings をインストールします。

   この例では、コマンドの watched-namespace を前述のステップでリストした namespace に置き換えます。watched-namespace-1、watched-namespace-2、および watched-namespace-3 にも、繰り返し同様の操作を実行します。

   oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/023-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>

4. Cluster Operator をデプロイします。

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. デプロイメントのステータスを確認します。

   oc get deployments -n my-cluster-operator-namespace

   デプロイメント名と準備状態が表示されている出力

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

手順

1. Cluster Operator のインストール先の namespace を使用するように、AMQ Streams インストールファイルを編集します。

   たとえば、この手順では Cluster Operator は my-cluster-operator-namespace という namespace にインストールされます。

   Linux の場合は、以下を使用します。

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS の場合は、以下を使用します。

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml ファイルを編集し、STRIMZI_NAMESPACE 環境変数の値を * に設定します。

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         # ...
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.1
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

3. クラスター全体ですべての namespace にアクセスできる権限を Cluster Operator に付与する ClusterRoleBindings を作成します。

   oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-watched --clusterrole=strimzi-cluster-operator-watched --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

4. Cluster Operator を OpenShift クラスターにデプロイします。

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. デプロイメントのステータスを確認します。

   oc get deployments -n my-cluster-operator-namespace

   デプロイメント名と準備状態が表示されている出力

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

手順

1. 一時または永続クラスターを作成およびデプロイします。

   • 一時クラスターを作成およびデプロイするには、以下を実行します。

     oc apply -f examples/kafka/kafka-ephemeral.yaml

   • 永続クラスターを作成およびデプロイするには、以下を実行します。

     oc apply -f examples/kafka/kafka-persistent.yaml

2. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   Pod 名および readiness が表示される出力

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-kafka-0          1/1     Running   0
   my-cluster-kafka-1          1/1     Running   0
   my-cluster-kafka-2          1/1     Running   0
   my-cluster-zookeeper-0      1/1     Running   0
   my-cluster-zookeeper-1      1/1     Running   0
   my-cluster-zookeeper-2      1/1     Running   0

   my-cluster は Kafka クラスターの名前です。

   0 で始まる連続インデックス番号は、作成された各 Kafka および ZooKeeper Pod を識別します。

   デフォルトのデプロイメントでは、Entity Operator クラスター、3 つの Kafka Pod、および 3 つの ZooKeeper Pod を作成します。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. コマンドラインから KafkaNodePools フィーチャーゲートを有効にします。

   oc set env deployment/strimzi-cluster-operator STRIMZI_FEATURE_GATES="+KafkaNodePools"

   または、Cluster Operator Deployment を編集し、STRIMZI_FEATURE_GATES 環境変数を更新します。

   env
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

   これにより、クラスター Operator が更新されます。

   KRaft モードを使用する場合は、UseKRaft フィーチャーゲートも有効にします。

2. ノードプールを作成します。

   • 3 つのブローカーの 2 つのノードプールを備えた Kafka クラスターと ZooKeeper クラスターをデプロイするには、次の手順を実行します。

     oc apply -f examples/kafka/nodepools/kafka.yaml

   • デュアルロールノードを使用する単一ノードプールを備えた Kafka クラスターを KRaft モードでデプロイするには、次の手順を実行します。

     oc apply -f examples/kafka/nodepools/kafka-with-dual-role-kraft-nodes.yaml

   • ブローカーノードとコントローラーノードに個別のノードプールを使用して Kafka クラスターを KRaft モードでデプロイするには、次の手順を実行します。

     oc apply -f examples/kafka/nodepools/kafka-with-kraft.yaml

3. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   出力にはノードプール名と準備状況が表示されます

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-pool-a-kafka-0   1/1     Running   0
   my-cluster-pool-a-kafka-1   1/1     Running   0
   my-cluster-pool-a-kafka-4   1/1     Running   0

   • my-cluster は Kafka クラスターの名前です。

   • pool-a はノードプールの名前です。

     0 で始まる連続したインデックス番号は、作成された各 Kafka Pod を識別します。ZooKeeper を使用している場合は、ZooKeeper Pod も表示されます。

     READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

     デプロイに関する情報は、プール内のノードの ID のリストなど、KafkaNodePool リソースのステータスにも表示されます。

     注記

     ノード ID は、クラスター内のすべてのノードプールにわたって 0 (ゼロ) から順番に割り当てられます。これは、ノード ID が特定のノードプール内で連続して実行されない可能性があることを意味します。クラスター全体のノード ID のシーケンスにギャップがある場合、追加される次のノードにはギャップを埋める ID が割り当てられます。スケールダウンすると、プール内で最も大きな数のノード ID を持つノードが削除されます。

手順

1. Kafka リソースの entityOperator プロパティーを編集し、topicOperator が含まれるようにします。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. EntityTopicOperatorSpec スキーマ参照 で説明されているプロパティーを使用して、トピック Operator spec を設定します。

   すべてのプロパティーにデフォルト値を使用する場合は、空のオブジェクト ({}) を使用します。

3. リソースを作成または更新します。

   oc apply -f <kafka_configuration_file>

4. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   Pod 名と準備状況が表示される出力

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster は Kafka クラスターの名前です。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. Kafka リソースの entityOperator プロパティーを編集し、userOperator が含まれるようにします。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. EntityUserOperatorSpec スキーマ参照 に記載されているプロパティーを使用して、User Operator の spec を設定します。

   すべてのプロパティーにデフォルト値を使用する場合は、空のオブジェクト ({}) を使用します。

3. リソースを作成または更新します。

   oc apply -f <kafka_configuration_file>

4. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   Pod 名と準備状況が表示される出力

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster は Kafka クラスターの名前です。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. Kafka Connect を OpenShift クラスターにデプロイします。examples/connect/kafka-connect.yaml ファイルを使用して Kafka Connect をデプロイします。

   oc apply -f examples/connect/kafka-connect.yaml

2. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   デプロイメント名と準備状態が表示されている出力

   NAME                                 READY  STATUS   RESTARTS
   my-connect-cluster-connect-<pod_id>  1/1    Running  0

   my-connect-cluster は、Kafka Connect クラスターの名前です。

   Pod ID は、作成された各 Pod を識別します。

   デフォルトのデプロイでは、単一の Kafka Connect Pod を作成します。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. Kafka MirrorMaker を OpenShift クラスターにデプロイします。

   MirrorMaker の場合

   oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml

   MirrorMaker 2 の場合:

   oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml

2. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   デプロイメント名と準備状態が表示されている出力

   NAME                                    READY  STATUS   RESTARTS
   my-mirror-maker-mirror-maker-<pod_id>   1/1    Running  1
   my-mm2-cluster-mirrormaker2-<pod_id>    1/1    Running  1

   my-mirror-maker は、Kafka MirrorMaker クラスターの名前です。my-mm2-cluster は、Kafka MirrorMaker 2 クラスターの名前です。

   Pod ID は、作成された各 Pod を識別します。

   デフォルトのデプロイメントでは、単一の MirrorMaker または MirrorMaker 2 Pod をインストールします。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. Kafka Bridge を OpenShift クラスターにデプロイします。

   oc apply -f examples/bridge/kafka-bridge.yaml

2. デプロイメントのステータスを確認します。

   oc get pods -n <my_cluster_operator_namespace>

   デプロイメント名と準備状態が表示されている出力

   NAME                       READY  STATUS   RESTARTS
   my-bridge-bridge-<pod_id>  1/1    Running  0

   my-bridge は、Kafka Bridge クラスターの名前です。

   Pod ID は、作成された各 Pod を識別します。

   デフォルトのデプロイメントでは、単一の Kafka Bridge Pod をインストールします。

   READY は、Ready/expected 状態のレプリカ数を表示します。STATUS が Running と表示されれば、デプロイメントは成功です。

手順

1. OpenShift クラスターの Pod の名前をリストします。

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-<pod_id>

2. ポート 8080 で Kafka Bridge Pod に接続します。

   oc port-forward pod/my-bridge-bridge-<pod_id> 8080:8080 &

   注記

   ローカルマシンのポート 8080 がすでに使用中の場合は、代わりの HTTP ポート (8008 など) を使用します。

手順

1. install/topic-operator/05-Deployment-strimzi-topic-operator.yaml スタンドアロンデプロイメントファイルの env プロパティーを編集します。

   スタンドアロンの Topic Operator デプロイメント設定の例

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-topic-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-topic-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_RESOURCE_LABELS 3
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_ZOOKEEPER_CONNECT 4
                 value: my-cluster-zookeeper-client:2181
               - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5
                 value: "18000"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7
                 value: "6"
               - name: STRIMZI_LOG_LEVEL 8
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 9
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 10
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 12
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 13
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 14
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 15
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 16
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 17
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 18
                 value: "SSL"

   1

   KafkaTopic リソースを監視する Topic Operator の OpenShift namespace。Kafka クラスターの namespace を指定します。

   2

   Kafka クラスターのすべてのブローカーを検出し、接続するブートストラップブローカーアドレスのホストとポートのペア。サーバーがダウンした場合に備えて、コンマ区切りリストを使用して 2 つまたは 3 つのブローカーアドレスを指定します。

   3

   Topic Operator によって管理される KafkaTopic リソースを識別するラベル。これは、Kafka クラスターの名前である必要はありません。KafkaTopic リソースに割り当てられたラベルにすることができます。複数の Topic Operator をデプロイする場合、ラベルはそれぞれに一意である必要があります。つまり、Operator は同じリソースを管理できません。

   4

   (ZooKeeper) ZooKeeper クラスターに接続するためのアドレスのホストおよびポートのペア。これは、Kafka クラスターが使用する ZooKeeper クラスターと同じである必要があります。

   5

   (ZooKeeper) ZooKeeper セッションのタイムアウト (ミリ秒単位)。デフォルトは 18000 (18 秒) です。

   6

   定期的な調整の間隔 (秒単位)。デフォルトは 120000 (2 分) です。

   7

   Kafka からトピックメタデータの取得を試行する回数。各試行の間隔は、指数バックオフとして定義されます。パーティションまたはレプリカの数が原因で、トピックの作成に時間がかかる場合は、この値を大きくすることを検討してください。デフォルトの試行回数は 6 回です。

   8

   ロギングメッセージの出力レベル。レベルを、ERROR、WARNING、INFO、DEBUG、または TRACE に設定できます。

   9

   Kafka ブローカーとの暗号化された通信の TLS サポートを有効にします。

   10

   (任意) Topic Operator を実行する JVM に使用される Java オプション。

   11

   (任意) Topic Operator に設定されたデバッグ (-D) オプション。

   12

   (オプション)TLS が STRIMZI_TLS_ENABLED によって有効になっている場合、トラストストア証明書の生成を省略します。この環境変数が有効になっている場合、ブローカーは TLS 証明書に公的に信頼できる認証局を使用する必要があります。デフォルトは false です。

   13

   (オプション) mTLS 認証用のキーストア証明書を生成します。これを false に設定すると、mTLS を使用した Kafka ブローカーへのクライアント認証が無効になります。デフォルトは true です。

   14

   (オプション)Kafka ブローカーに接続するときにクライアント認証の SASL サポートを有効にします。デフォルトは false です。

   15

   (任意) クライアント認証用の SASL ユーザー名。SASL が STRIMZI_SASL_ENABLED によって有効化された場合のみ必須です。

   16

   (任意) クライアント認証用の SASL パスワード。SASL が STRIMZI_SASL_ENABLED によって有効化された場合のみ必須です。

   17

   (任意) クライアント認証用の SASL メカニズム。SASL が STRIMZI_SASL_ENABLED によって有効化された場合のみ必須です。この値は plain、scram-sha-256、または scram-sha-512 に設定できます。

   18

   (任意)Kafka ブローカーとの通信に使用されるセキュリティープロトコル。デフォルト値は PLAINTEXT です。値は PLAINTEXT、SSL、SASL_PLAINTEXT、または SASL_SSL に設定できます。

2. 公開認証局から証明書を使用している Kafka ブ ローカーに接続する場合は、STRIMZI_PUBLIC_CA を true に設定します。たとえば、Amazon AWS MSK サービスを使用している場合は、このプロパティーを true に設定します。

3. STRIMZI_TLS_ENABLED 環境変数で mTLS を有効にした場合は、Kafka クラスターへの接続認証に使用されるキーストアおよびトラストストアを指定します。

   mTLS 設定の例

   # ....
   env:
     - name: STRIMZI_TRUSTSTORE_LOCATION 1
       value: "/path/to/truststore.p12"
     - name: STRIMZI_TRUSTSTORE_PASSWORD 2
       value: "TRUSTSTORE-PASSWORD"
     - name: STRIMZI_KEYSTORE_LOCATION 3
       value: "/path/to/keystore.p12"
     - name: STRIMZI_KEYSTORE_PASSWORD 4
       value: "KEYSTORE-PASSWORD"
   # ...

   1

   トラストストアには、Kafka および ZooKeeper サーバー証明書の署名に使用される認証局の公開鍵が含まれます。

   2

   トラストストアにアクセスするためのパスワード。

   3

   キーストアには、mTLS 認証用の秘密鍵が含まれています。

   4

   キーストアにアクセスするためのパスワード。

4. Topic Operator をデプロイします。

   oc create -f install/topic-operator

5. デプロイメントのステータスを確認します。

   oc get deployments

   デプロイメント名と準備状態が表示されている出力

   NAME                    READY  UP-TO-DATE  AVAILABLE
   strimzi-topic-operator  1/1    1           1

   READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

手順

1. 以下の env プロパティーを install/user-operator/05-Deployment-strimzi-user-operator.yaml スタンドアロンデプロイメントファイルで編集します。

   スタンドアロン User Operator デプロイメント設定の例

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-user-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-user-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_CA_CERT_NAME 3
                 value: my-cluster-clients-ca-cert
               - name: STRIMZI_CA_KEY_NAME 4
                 value: my-cluster-clients-ca
               - name: STRIMZI_LABELS 5
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_WORK_QUEUE_SIZE 7
                 value: 10000
               - name: STRIMZI_CONTROLLER_THREAD_POOL_SIZE 8
                 value: 10
               - name: STRIMZI_USER_OPERATIONS_THREAD_POOL_SIZE 9
                 value: 4
               - name: STRIMZI_LOG_LEVEL 10
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 11
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 12
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 13
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 14
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 15
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 16
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 17
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 18
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 19
                 value: |
                   default.api.timeout.ms=120000
                   request.timeout.ms=60000

   1

   KafkaUser リソースを監視する User Operator の OpenShift namespace。指定できる namespace は 1 つだけです。

   2

   Kafka クラスターのすべてのブローカーを検出し、接続するブートストラップブローカーアドレスのホストとポートのペア。サーバーがダウンした場合に備えて、コンマ区切りリストを使用して 2 つまたは 3 つのブローカーアドレスを指定します。

   3

   mTLS 認証用の新しいユーザー証明書に署名する認証局の公開鍵 (ca.crt) 値を含む OpenShift Secret。

   4

   mTLS 認証用の新しいユーザー証明書に署名する CA の秘密鍵 (ca.key) 値を含む OpenShift Secret。

   5

   User Operator によって管理される KafkaUser リソースを識別するラベル。これは、Kafka クラスターの名前である必要はありません。KafkaUser リソースに割り当てられたラベルにすることができます。複数の User Operator をデプロイする場合、ラベルはそれぞれに一意である必要があります。つまり、Operator は同じリソースを管理できません。

   6

   定期的な調整の間隔 (秒単位)。デフォルトは 120000 (2 分) です。

   7

   コントローラーイベントキューのサイズ。キューのサイズは、User Operator が操作すると予想されるユーザーの最大数と少なくとも同じ大きさにする必要があります。デフォルトは 1024 です。

   8

   ユーザーを調整するためのワーカープールのサイズ。プールを大きくすると、より多くのリソースが必要になる可能性がありますが、より多くの KafkaUser リソースも処理されます。デフォルトは 50 です。

   9

   Kafka Admin API および OpenShift 操作のワーカープールのサイズ。プールを大きくすると、より多くのリソースが必要になる可能性がありますが、より多くの KafkaUser リソースも処理されます。デフォルトは 4 です。

   10

   ロギングメッセージの出力レベル。レベルを、ERROR、WARNING、INFO、DEBUG、または TRACE に設定できます。

   11

   ガベッジコレクション (GC) ロギングを有効にします。デフォルトは true です。

   12

   CA の有効期間。デフォルトは 365 日です。

   13

   CA の更新期間。更新期間は、現在の証明書の有効期日から逆算されます。デフォルトでは、古い証明書が期限切れになる前の証明書の更新期間は 30 日です。

   14

   (任意) User Operator を実行する JVM に使用される Java オプション。

   15

   (任意) User Operator に設定されたデバッグ (-D) オプション。

   16

   (オプション)User Operator によって作成される OpenShift シークレットの名前の接頭辞。

   17

   (任意)Kafka クラスターが Kafka Admin API を使用した承認 ACL ルールの管理をサポートするかどうかを示します。false に設定すると、User Operator は simple 承認 ACL ルールを持つすべてのリソースを拒否します。これは、Kafka クラスターログで不要な例外を回避するのに役立ちます。デフォルトは true です。

   18

   (オプション) 期限切れのユーザー証明書が更新されるメンテナンス時間枠を定義する Cron 式のセミコロンで区切られたリスト。

   19

   (オプション) プロパティー形式で User Operator が使用する Kafka Admin クライアントを設定するための設定オプション。

2. mTLS を使用して Kafka クラスターに接続する場合は、接続の認証に使用されるシークレットを指定します。それ以外の場合は、次のステップに進みます。

   mTLS 設定の例

   # ....
   env:
     - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1
       value: my-cluster-cluster-ca-cert
     - name: STRIMZI_EO_KEY_SECRET_NAME 2
       value: my-cluster-entity-operator-certs
   # ..."

   1

   Kafka ブローカー証明書に署名する CA の公開鍵 (ca.crt) 値を含む OpenShift Secret。

   2

   Kafka クラスターに対する mTLS 認証に使用される証明書の公開鍵 (entity-operator.crt) と秘密鍵 (entity-operator.key) を含む OpenShift Secret。

3. User Operator をデプロイします。

   oc create -f install/user-operator

4. デプロイメントのステータスを確認します。

   oc get deployments

   デプロイメント名と準備状態が表示されている出力

   NAME                   READY  UP-TO-DATE  AVAILABLE
   strimzi-user-operator  1/1    1           1

   READY は、Ready/expected 状態のレプリカ数を表示します。AVAILABLE 出力に 1 が表示されれば、デプロイメントは成功しています。

手順

1. Kafka リソースの config にプラグインのプロパティーを追加します。

   プラグインプロパティーは、この設定例のとおりです。

   Kafka Static Quota プラグインの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       config:
         client.quota.callback.class: io.strimzi.kafka.quotas.StaticQuotaCallback 1
         client.quota.callback.static.produce: 1000000 2
         client.quota.callback.static.fetch: 1000000 3
         client.quota.callback.static.storage.soft: 400000000000 4
         client.quota.callback.static.storage.hard: 500000000000 5
         client.quota.callback.static.storage.check-interval: 5 6

   1

   Kafka Static Quota プラグインを読み込みます。

   2

   プロデューサーのバイトレートしきい値を設定します。この例では 1 MBps です。

   3

   コンシューマーのバイトレートしきい値を設定します。この例では 1 MBps です。

   4

   ストレージのソフト制限の下限を設定します。この例では 400 GB です。

   5

   ストレージのハード制限の上限を設定します。この例では 500 GB です。

   6

   ストレージのチェックの間隔 (秒単位) を設定します。この例では 5 秒です。これを 0 に設定するとチェックを無効にできます。

2. リソースを更新します。

   oc apply -f <kafka_configuration_file>

手順

1. 次の例に示すように、スケールアップまたはスケールダウン時に使用する ID でノードプールにアノテーションを付けます。

   スケールアップ用の ID は、ノードプール pool-a に割り当てられます。

   スケールアップ用の ID の割り当て

   oc annotate kafkanodepool pool-a strimzi.io/next-node-ids="[0,1,2,10-20,30]"

   ノードを pool-a に追加するときに、この範囲内で使用可能な最小の ID が使用されます。

   スケールダウン用の ID はノードプール pool-b に割り当てられます。

   スケールダウン用の ID の割り当て

   oc annotate kafkanodepool pool-b strimzi.io/remove-node-ids="[60-50,9,8,7]"

   pool-b をスケールダウンすると、この範囲内で使用可能な最大の ID が削除されます。

2. ノードプールをスケーリングできるようになりました。

   詳細は、以下を参照してください。

   • 「(プレビュー) ノードプールへのノードの追加」
   • 「(プレビュー) ノードプールからのノードの削除」
   • 「(プレビュー) ノードプール間でのノードの移動」

   調整時に、アノテーションの形式が間違っている場合は警告が表示されます。

手順

1. ノードプールに新しいノードを作成します。

   たとえば、ノードプール pool-a には 3 つのレプリカがあります。レプリカの数を増やしてノードを追加します。

   oc scale kafkanodepool pool-a --replicas=4

2. デプロイメントのステータスを確認し、ノードプール内の Pod が作成され、ステータスが READY になるまで待ちます。

   oc get pods -n <my_cluster_operator_namespace>

   出力には、ノードプール内の 4 つの Kafka ノードが表示されます

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-a-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-2  1/1    Running  0
   my-cluster-pool-a-kafka-3  1/1    Running  0

3. ノードプール内のノードの数を増やした後、パーティションを再割り当てします。

   ノードプールをスケールアップした後、Cruise Control の add-brokers モードを使用して、パーティションレプリカを既存のブローカーから新しく追加したブローカーに移動できます。

手順

1. ノードプール内のノードの数を減らす前に、パーティションを再割り当てします。

   ノードプールをスケールダウンする前に、Cruise Control remove-brokers モードを使用して、削除されるブローカーからパーティションレプリカを移動できます。

2. 再割り当てプロセスが完了し、削除されるノードにライブパーティションがなくなったら、ノードプール内の Kafka ノードの数を減らします。

   たとえば、ノードプール pool-b には 4 つのレプリカがあります。レプリカの数を減らしてノードを削除します。

   oc scale kafkanodepool pool-a --replicas=3

   出力には、ノードプール内の 3 つの Kafka ノードが表示されます

   NAME                        READY   STATUS    RESTARTS
   my-cluster-pool-b-kafka-0   1/1     Running   0
   my-cluster-pool-b-kafka-1   1/1     Running   0
   my-cluster-pool-b-kafka-2   1/1     Running   0

手順

1. ターゲットノードプールに新しいノードを作成します。

   たとえば、ノードプール pool-a には 3 つのレプリカがあります。レプリカの数を増やしてノードを追加します。

   oc scale kafkanodepool pool-a --replicas=4

2. デプロイメントのステータスを確認し、ノードプール内の Pod が作成され、ステータスが READY になるまで待ちます。

   oc get pods -n <my_cluster_operator_namespace>

   出力には、ターゲットノードプール内の 4 つの Kafka ノードが表示されます

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-a-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-4  1/1    Running  0
   my-cluster-pool-a-kafka-5  1/1    Running  0

   ノード ID は、作成時にノードの名前に追加されます。ノード ID が 5 であるノード my-cluster-pool-a-kafka-5 を追加します。

3. パーティションを古いノードから新しいノードに再割り当てします。

   ソースノードプールをスケールダウンする前に、Cruise Control の remove-broker モードを使用して、削除されるブローカーからパーティションレプリカを移動できます。

4. 再割り当てプロセスが完了したら、ソースノードプール内の Kafka ノードの数を減らします。

   たとえば、ノードプール pool-b には 4 つのレプリカがあります。レプリカの数を減らしてノードを削除します。

   oc scale kafkanodepool pool-b --replicas=3

   プール内で最も数値の大きい ID を持つノードが削除されます。

   出力には、ソースノードプール内の 3 つの Kafka ノードが表示されます

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-b-kafka-2  1/1    Running  0
   my-cluster-pool-b-kafka-3  1/1    Running  0
   my-cluster-pool-b-kafka-6  1/1    Running  0

手順

1. 新しい KafkaNodePool リソースを作成します。

   1. リソースに kafka という名前を付けます。
   2. strimzi.io/cluster ラベルが既存の Kafka リソースを指すようにします。
   3. 現在の Kafka クラスターと一致するようにレプリカ数とストレージ設定を設定します。
   4. ロールを Broker に設定します。

   Kafka クラスターの移行で使用されるノードプールの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: kafka
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     replicas: 3
     roles:
       - broker
     storage:
       type: jbod
       volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false

2. KafkaNodePool リソースを適用します。

   oc apply -f <node_pool_configuration_file>

   このリソースを適用すると、Kafka がノードプールの使用に切り替わります。

   変更やローリングアップデートはなく、リソースは以前と同じです。

3. Cluster Operator 設定の STRIMZI_FEATURE_GATES 環境変数を更新して、+KafkaNodePools を含めます。

   env:
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

4. strimzi.io/node-pools:enabled アノテーションを使用して、Kafka リソースで KafkaNodePools フィーチャーゲートを有効にします。

   ZooKeeper を使用したクラスター内のノードプールの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     annotations:
       strimzi.io/node-pools: enabled
   spec:
     kafka:
       version: 3.5.0
       replicas: 3
     # ...
     storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false

5. Kafka リソースを適用します。

   oc apply -f <kafka_configuration_file>

手順

1. クラスターオペレータにプロキシー環境変数を追加するには、その Deployment 設定 (install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml) を更新します。

   Cluster Operator のプロキシー設定の例

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "HTTP_PROXY"
             value: "http://proxy.com" 1
           - name: "HTTPS_PROXY"
             value: "https://proxy.com" 2
           - name: "NO_PROXY"
             value: "internal.com, other.domain.com" 3
     # ...

   1

   プロキシーサーバーのアドレス。

   2

   プロキシーサーバーのセキュアなアドレス。

   3

   プロキシーサーバーの例外として直接アクセスされるサーバーのアドレス。URL はコンマで区切られます。

   または、Deployment を直接編集します。

   oc edit deployment strimzi-cluster-operator

2. Deployment を直接編集せずに YAML ファイルを更新する場合は、変更を適用します。

   oc create -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

手順

1. Cluster Operator で FIPS モードを無効にするには、Deployment 設定 (install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml) を更新し、FIPS_MODE 環境変数を追加します。

   Cluster Operator の FIPS 設定例

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "FIPS_MODE"
             value: "disabled" 1
     # ...

   1

   FIPS モードを無効にします。

   または、Deployment を直接編集します。

   oc edit deployment strimzi-cluster-operator

2. Deployment を直接編集せずに YAML ファイルを更新する場合は、変更を適用します。

   oc apply -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

手順

1. KafkaUser リソースの authorization プロパティーを編集し、アクセス権限をユーザーに付与します。

   以下の例では、literal の名前の値を使用して Kafka Connect トピックおよびコンシューマーグループにアクセス権限が設定されます。

   プロパティー	名前
   offset.storage.topic	connect-cluster-offsets
   status.storage.topic	connect-cluster-status
   config.storage.topic	connect-cluster-configs
   group	connect-cluster

   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: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
           host: "*"

2. リソースを作成または更新します。

   oc apply -f KAFKA-USER-CONFIG-FILE

手順

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.5.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.5"
       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.5.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.5"
       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 クラスターでの操作を許可します。

      ACL は、内部 MirrorMaker コネクター、および基盤となる Kafka Connect フレームワークによって使用されます。

   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:
         # Underlying Kafka Connect internal topics to store configuration, offsets, or status
         - resource:
             type: group
             name: mirrormaker2-cluster
           operations:
             - Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - 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.5.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>