手順

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 apply -f install/cluster-operator -n my-cluster-operator-namespace

3. Cluster Operator が正常にデプロイされたことを確認します。

   oc get deployments

手順

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/050-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/amq7/amq-streams-rhel7-operator:1.5.0
           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 apply -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n watched-namespace
   oc apply -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n watched-namespace
   oc apply -f install/cluster-operator/032-RoleBinding-strimzi-cluster-operator-topic-operator-delegation.yaml -n watched-namespace

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

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

5. Cluster Operator が正常にデプロイされたことを確認します。

   oc get deployments

手順

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/050-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/amq7/amq-streams-rhel7-operator:1.5.0
           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-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-topic-operator-delegation --clusterrole=strimzi-topic-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

   my-cluster-operator-namespace は、Cluster Operator をインストールする namespace に置き換えます。

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

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

5. Cluster Operator が正常にデプロイされたことを確認します。

   oc get deployments

手順

1. OpenShift 4 Web コンソールで、Operators > OperatorHub をクリックします。

2. Streaming & Messaging カテゴリーの AMQ Streams Operator を検索または閲覧します。

   
3. AMQ Streams タイルをクリックし、右側のサイドバーで Install をクリックします。

4. Create Operator Subscription 画面で、以下のインストールおよび更新オプションから選択します。

   • Installation Mode: AMQ Streams Operator をクラスターのすべての (プロジェクト) namespace にインストール (デフォルト) するか、特定の (プロジェクト) namespace インストールするかを選択します。namespace を使用して関数を分離することが推奨されます。特定の namespace を Kafka クラスターおよびその他の AMQ Streams コンポーネントの専用とすることが推奨されます。
   • Approval Strategy: デフォルトでは、OLM (Operator Lifecycle Manager) によって、AMQ Streams Operator が自動的に最新の AMQ Streams バージョンにアップグレードされます。今後のアップグレードを手動で承認する場合は、Manual を選択します。詳細は、OpenShift ドキュメントのOperatorを参照してください。

5. Subscribe をクリックすると、AMQ Streams Operator が OpenShift クラスターにインストールされます。

   AMQ Streams Operator によって、Cluster Operator、CRD、およびロールベースアクセス制御 (RBAC) リソースは選択された namespace またはすべての namespace にデプロイされます。

6. Installed Operators 画面で、インストールの進捗を確認します。AMQ Streams Operator は、ステータスが InstallSucceeded に変更されると使用できます。

   

手順

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

   開発またはテストでは、一時クラスターの使用が適している可能性があります。永続クラスターはどのような状況でも使用することができます。

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

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

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

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

2. Kafka クラスターが正常にデプロイされたことを確認します。

   oc get deployments

手順

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

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

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

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

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

   次のように oc apply を使用します。

   oc apply -f <your-file>

手順

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

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

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

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

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

   oc apply -f <your-file>

手順

1. 以下を設定し、install/topic-operator/05-Deployment-strimzi-topic-operator.yaml ファイルの Deployment.spec.template.spec.containers[0].env プロパティーを編集します。

   1. STRIMZI_KAFKA_BOOTSTRAP_SERVERS。 hostname:‍port ペアのコンマ区切りリストで Kafka クラスターのブートストラップブローカーを指定します。
   2. STRIMZI_ZOOKEEPER_CONNECT。hostname:‍port ペアのコンマ区切りリストで ZooKeeper ノードを指定します。これは、Kafka クラスターが使用する ZooKeeper クラスターと同じである必要があります。
   3. STRIMZI_NAMESPACE。Operator が KafkaTopic リソースを監視する OpenShift namespace。
   4. STRIMZI_RESOURCE_LABELS。Operator によって管理される KafkaTopic リソースを識別するために使用されるラベルセレクター。
   5. STRIMZI_FULL_RECONCILIATION_INTERVAL_MS。定期的な調整の間隔 (秒単位) を指定します。
   6. STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS。Kafka からトピックメタデータを取得するための試行回数を指定します。各試行の間隔は、指数バックオフとして定義されます。パーティションまたはレプリカの数によって、トピックの作成に時間がかかる可能性がある場合は、この値を増やすことを検討してください。デフォルトは 6 です。
   7. STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS。ZooKeeper セッションのタイムアウト (ミリ秒単位)。例: 10000。デフォルトは 20000 (20 秒) です。
   8. STRIMZI_TOPICS_PATH。Topic Operator がそのメタデータを保存する Zookeeper ノードパス。デフォルトは /strimzi/topics です。
   9. STRIMZI_TLS_ENABLED。Kafka ブローカーとの通信を暗号化するために、TLS サポートを有効にします。デフォルトは true です。
   10. STRIMZI_TRUSTSTORE_LOCATION。TLS ベースの通信を有効にするための証明書が含まれるトラストストアへのパス。TLS が STRIMZI_TLS_ENABLED によって有効化された場合のみ必須です。
   11. STRIMZI_TRUSTSTORE_PASSWORD。STRIMZI_TRUSTSTORE_LOCATION で定義されるトラストストアにアクセスするためのパスワード。TLS が STRIMZI_TLS_ENABLED によって有効化された場合のみ必須です。
   12. STRIMZI_KEYSTORE_LOCATION。TLS ベースの通信を有効にするための秘密鍵が含まれるキーストアへのパス。TLS が STRIMZI_TLS_ENABLED によって有効化された場合のみ必須です。
   13. STRIMZI_KEYSTORE_PASSWORD。STRIMZI_KEYSTORE_LOCATION で定義されるキーストアにアクセスするためのパスワード。TLS が STRIMZI_TLS_ENABLED によって有効化された場合のみ必須です。
   14. STRIMZI_LOG_LEVEL。ロギングメッセージの出力レベル。値は、ERROR、WARNING、INFO、DEBUG、および TRACE に設定できます。デフォルトは INFO です。
   15. STRIMZI_JAVA_OPTS (任意)。Topic Operator を実行する JVM に使用される Java オプション。例: -Xmx=512M -Xms=256M
   16. STRIMZI_JAVA_SYSTEM_PROPERTIES (任意)。Topic Operator に設定される -D オプションをリストします。例: -Djavax.net.debug=verbose -DpropertyName=value.

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

   oc apply -f install/topic-operator

3. Topic Operator が正常にデプロイされていることを確認します。

   oc describe deployment strimzi-topic-operator

   Replicas: エントリーに 1 available が表示されれば、Topic Operator はデプロイされています。

   注記

   OpenShift への接続が低速な場合やイメージがこれまでダウンロードされたことがない場合は、デプロイメントに遅延が発生することがあります。

手順

1. 以下を設定し、install/user-operator/05-Deployment-strimzi-user-operator.yaml ファイルの Deployment.spec.template.spec.containers[0].env プロパティーを編集します。

   1. STRIMZI_KAFKA_BOOTSTRAP_SERVERS。 hostname:‍port ペアのコンマ区切りリストで Kafka ブローカーを指定します。
   2. STRIMZI_ZOOKEEPER_CONNECT。hostname:‍port ペアのコンマ区切りリストで ZooKeeper ノードを指定します。これは、Kafka クラスターが使用する ZooKeeper クラスターと同じである必要があります。TLS 暗号化で ZooKeeper ノードに接続することはサポートされません。
   3. STRIMZI_NAMESPACE。Operator が KafkaUser リソースを監視する OpenShift namespace。
   4. STRIMZI_LABELS。Operator によって管理される KafkaUser リソースを識別するために使用されるラベルセレクター。
   5. STRIMZI_FULL_RECONCILIATION_INTERVAL_MS。定期的な調整の間隔 (秒単位) を指定します。
   6. STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS。ZooKeeper セッションのタイムアウト (ミリ秒単位)。例: 10000。デフォルトは 20000 (20 秒) です。
   7. STRIMZI_CA_CERT_NAME。TLS クライアント認証に対して新しいユーザー証明書を署名するための認証局の公開鍵が含まれる OpenShift Secret を示します。Secret の ca.crt キーに、認証局の公開鍵が含まれている必要があります。
   8. STRIMZI_CA_KEY_NAME。TLS クライアント認証に対して新しいユーザー証明書を署名するための認証局の秘密鍵が含まれる OpenShift Secret を示します。Secret の ca.crt キーに、認証局の秘密鍵が含まれている必要があります。
   9. STRIMZI_CLUSTER_CA_CERT_SECRET_NAME。TLS ベースの通信を有効にするために Kafka ブローカーの証明書の署名に使用される認証局の秘密鍵が含まれる OpenShift Secret を示します。Secret の ca.crt キーに、認証局の公開鍵が含まれている必要があります。この環境変数の設定は任意で、Kafka クラスターとの通信が TLS ベースである場合のみ設定する必要があります。
   10. STRIMZI_EO_KEY_SECRET_NAME。Kafka クラスターに対する TLS クライアント認証の秘密鍵と関連する証明書が含まれる OpenShift Secret を示します。Secret の entity-operator.p12 キーに、秘密鍵と証明書が含まれるキーストアが含まれ、 entity-operator.password キーに関連するパスワードが含まれる必要があります。この環境変数の設定は任意で、Kafka クラスターとの通信が TLS ベースで、TLS のクライアント認証が必要な場合のみ設定する必要があります。
   11. STRIMZI_CA_VALIDITY。認証局の有効期限。デフォルトは 365 日です。
   12. STRIMZI_CA_RENEWAL。認証局の更新期限。
   13. STRIMZI_LOG_LEVEL。ロギングメッセージの出力レベル。値は、ERROR、WARNING、INFO、DEBUG、および TRACE に設定できます。デフォルトは INFO です。
   14. STRIMZI_GC_LOG_ENABLED。ガベージコレクション (GC) ロギングを有効にします。デフォルトは true です。デフォルトでは、古い証明書が期限切れになる前の証明書の更新期間は 30 日です。
   15. STRIMZI_JAVA_OPTS (任意)。User Operator を実行する JVM に使用される Java オプション。例: -Xmx=512M -Xms=256M
   16. STRIMZI_JAVA_SYSTEM_PROPERTIES (任意)。User Operator に設定される -D オプションをリストします。例: -Djavax.net.debug=verbose -DpropertyName=value.

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

   oc apply -f install/user-operator

3. User Operator が正常にデプロイされていることを確認します。

   oc describe deployment strimzi-user-operator

   Replicas: エントリーに 1 available が表示されれば、User Operator はデプロイされています。

   注記

   OpenShift への接続が低速な場合やイメージがこれまでダウンロードされたことがない場合は、デプロイメントに遅延が発生することがあります。

手順

1. registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0 をベースイメージとして使用して、新規の Dockerfile を作成します。

   FROM registry.redhat.io/amq7/amq-streams-kafka-25-rhel7:1.5.0
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   プラグインファイルの例

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-3.4.2.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-3.4.2.jar
   │   ├── mongodb-driver-core-3.4.2.jar
   │   └── README.md
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-0.13.0.jar
   │   ├── mysql-connector-java-5.1.40.jar
   │   ├── README.md
   │   └── wkb-1.0.2.jar
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-0.7.1.jar
       ├── debezium-core-0.7.1.jar
       ├── LICENSE.txt
       ├── postgresql-42.0.0.jar
       ├── protobuf-java-2.6.1.jar
       └── README.md

2. コンテナーイメージをビルドします。
3. カスタムイメージをコンテナーレジストリーにプッシュします。

4. 新しいコンテナーイメージを示します。

   以下のいずれかを行います。

   • KafkaConnect カスタムリソースの KafkaConnect.spec.image プロパティーを編集します。

     設定された場合、このプロパティーによって Cluster Operator の STRIMZI_KAFKA_CONNECT_IMAGES 変数がオーバーライドされます。

     apiVersion: kafka.strimzi.io/v1beta1
     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 ワーカー (コネクターではない) の設定。

     または

   • install/cluster-operator/050-Deployment-strimzi-cluster-operator.yaml ファイルの STRIMZI_KAFKA_CONNECT_IMAGES 変数を編集して新しいコンテナーイメージを示すようにした後、Cluster Operator を再インストールします。

手順

1. コマンドラインで oc apply コマンドを使用し、Kafka Connect の S2I クラスターを作成およびデプロイします。

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

2. Kafka Connect プラグインでディレクトリーを作成します。

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-3.4.2.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-3.4.2.jar
   │   ├── mongodb-driver-core-3.4.2.jar
   │   └── README.md
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-0.13.0.jar
   │   ├── mysql-connector-java-5.1.40.jar
   │   ├── README.md
   │   └── wkb-1.0.2.jar
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-0.7.1.jar
       ├── debezium-core-0.7.1.jar
       ├── LICENSE.txt
       ├── postgresql-42.0.0.jar
       ├── protobuf-java-2.6.1.jar
       └── README.md

3. oc start-build コマンドで、準備したディレクトリーを使用してイメージの新しいビルドを開始します。

   oc start-build my-connect-cluster-connect --from-dir ./my-plugins/

   注記

   ビルドの名前は、デプロイされた Kafka Connect クラスターと同じになります。

4. ビルドが完了したら、Kafka Connect のデプロイメントによって新しいイメージが自動的に使用されます。

手順

1. Kafka リソースで、Kafka クラスター、ZooKeeper クラスター、またはその両方に割り当てられた永続ボリュームのサイズを増やします。

   • Kafka クラスターに割り当てられたボリュームサイズを増やすには、spec.kafka.storage プロパティーを編集します。

   • ZooKeeper クラスターに割り当てたボリュームサイズを増やすには、spec.zookeeper.storage プロパティーを編集します。

     たとえば、ボリュームサイズを 1000Gi から 2000Gi に増やすには、以下のように編集します。

     apiVersion: kafka.strimzi.io/v1beta1
     kind: Kafka
     metadata:
       name: my-cluster
     spec:
       kafka:
         # ...
         storage:
           type: persistent-claim
           size: 2000Gi
           class: my-storage-class
         # ...
       zookeeper:
         # ...

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

   次のように oc apply を使用します。

   oc apply -f your-file

   OpenShift では、Cluster Operator からの要求に応じて、選択された永続ボリュームの容量が増やされます。サイズ変更が完了すると、サイズ変更された永続ボリュームを使用するすべての Pod が Cluster Operator によって再起動されます。これは自動的に行われます。

手順

1. Kafka リソースの spec.kafka.storage.volumes プロパティーを編集します。新しいボリュームを volumes アレイに追加します。たとえば、id が 2 の新しいボリュームを追加します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 1
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
         - id: 2
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

3. 新しいトピックを作成するか、既存のパーティションを新しいディスクに再度割り当てます。

手順

1. 削除するディスクからすべてのパーティションを再度割り当てます。削除するディスクに割り当てられたままになっているパーティションのデータは削除される可能性があります。

2. Kafka リソースの spec.kafka.storage.volumes プロパティーを編集します。volumes アレイから 1 つまたは複数のボリュームを削除します。たとえば、ID が 1 と 2 のボリュームを削除します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの replicas プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       replicas: 3
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.kafka.config プロパティーで、Kafka 設定を 1 つまたは複数入力します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       config:
         default.replication.factor: 3
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 1
       # ...
     zookeeper:
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. Kafka.spec.kafka リソースの listeners プロパティーを編集します。

   認証のないプレーン (暗号化されていない) リスナーの設定例:

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         plain: {}
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.kafka.listeners プロパティーで、認証を有効にするリスナーに authentication フィールドを追加します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         tls:
           authentication:
             type: tls
       # ...
     zookeeper:
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. Kafka.spec.kafka リソースの authorization プロパティーを追加または編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       authorization:
         type: simple
         superUsers:
           - CN=fred
           - sam
           - CN=edward
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.zookeeper.replicas プロパティーで、複製された ZooKeeper サーバーの数を入力します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       replicas: 3
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. クラスターデプロイメントを指定する Kafka リソースが含まれる YAML 設定ファイルを開きます。

2. Kafka リソースの spec.zookeeper.config プロパティーで、1 つまたは複数の ZooKeeper 設定を指定します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       config:
         autopurge.snapRetainCount: 3
         autopurge.purgeInterval: 1
       # ...

3. 新しい設定を適用してリソースを作成または更新します。

   次のように oc apply を使用します。

   oc apply -f kafka.yaml

   kafka.yaml は、設定するリソースの YAML 設定ファイルに置き換えます (例: kafka-persistent.yaml)。

手順

1. OpenShift コンソールを使用してターミナルを開くか、CLI から exec コマンドを実行します。

   以下に例を示します。

   oc exec -it my-cluster-kafka-0 -- bin/kafka-topics.sh --list --zookeeper localhost:2181

   必ず localhost:2181 を使用してください。

   ZooKeeper に対して Kafka コマンドを実行できるようになりました。

手順

1. Kafka リソースの entityOperator プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     entityOperator:
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
       userOperator:
         watchedNamespace: my-user-namespace
         reconciliationIntervalSeconds: 60

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定するリソースの resources プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. ノードがデプロイされたゾーンやラックを表すノードラベルについては、OpenShift 管理者に相談します。

2. ラベルをトポロジーキーとして使用し、Kafka リソースの rack プロパティーを編集します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       rack:
         topologyKey: failure-domain.beta.kubernetes.io/zone
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、または KafkaConnectS2I の livenessProbe または readinessProbe プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、または KafkaConnectS2I リソースの metrics プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、KafkaConnectS2I、KafkaMirrorMaker、または KafkaBridge リソースの jvmOptions プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       jvmOptions:
         "-Xmx": "8g"
         "-Xms": "8g"
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、または KafkaConnectS2I リソースの image プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       image: my-org/my-image:latest
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの tlsSidecar プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       tlsSidecar:
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka リソースの KafkaExporter プロパティーを編集します。

   設定可能なプロパティーは以下の例のとおりです。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     # ...
     kafkaExporter:
       image: my-org/my-image:latest 1
       groupRegex: ".*" 2
       topicRegex: ".*" 3
       resources: 4
         requests:
           cpu: 200m
           memory: 64Mi
         limits:
           cpu: 500m
           memory: 128Mi
       logging: debug 5
       enableSaramaLogging: true 6
       template: 7
         pod:
           metadata:
             labels:
               label1: value1
           imagePullSecrets:
             - name: my-docker-credentials
           securityContext:
             runAsUser: 1000001
             fsGroup: 0
           terminationGracePeriodSeconds: 120
       readinessProbe: 8
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe: 9
         initialDelaySeconds: 15
         timeoutSeconds: 5
   # ...

   1

   高度な任意手順: 特別な場合のみ推奨される コンテナーイメージの設定。

   2

   メトリクスに含まれるコンシューマーグループを指定する正規表現。

   3

   メトリクスに含まれるトピックを指定する正規表現。

   4

   予約する CPU およびメモリーリソース。

   5

   指定の重大度 (debug、info、warn、error、fatal) 以上でメッセージをログに記録するためのログ設定。

   6

   Sarama ロギングを有効にするブール値 (Kafka Exporter によって使用される Go クライアントライブラリー)。

   7

   デプロイメントテンプレートおよび Pod のカスタマイズ。

   8

   Healthcheck の readiness プローブ。

   9

   Healthcheck の liveness プローブ。

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

   oc apply -f kafka.yaml

手順

1. 移動するトピックを一覧表示する topics.json という名前の JSON ファイルを準備します。これには、以下の構造が必要です。

   {
     "version": 1,
     "topics": [
       <TopicObjects>
     ]
   }

   ここで <TopicObjects> は、以下のようなコンマ区切りのオブジェクトリストになります。

   {
     "topic": <TopicName>
   }

   たとえば、topic-a と topic-b のすべてのパーティションを再割り当てするには、以下のような topics.json ファイルを準備する必要があります。

   {
     "version": 1,
     "topics": [
       { "topic": "topic-a"},
       { "topic": "topic-b"}
     ]
   }

2. topics.json ファイルをブローカー Pod の 1 つにコピーします。

   cat topics.json | oc exec -c kafka <BrokerPod> -i -- \
     /bin/bash -c \
     'cat > /tmp/topics.json'

3. kafka-reassign-partitions.sh コマンドを使用して、再割り当て JSON を生成します。

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list <BrokerList> \
     --generate

   たとえば、 topic-a および topic-b のすべてのパーティションをブローカー 4 および 7 に移動する場合は、以下を実行し m す。

   oc exec <BrokerPod> -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --topics-to-move-json-file /tmp/topics.json \
     --broker-list 4,7 \
     --generate

手順

1. kafka.spec.kafka.replicas 設定オプションを増やして、新しいブローカーを必要なだけ追加します。
2. 新しいブローカー Pod が起動したことを確認します。

3. 後でコマンドを実行するブローカー Pod に reassignment.json ファイルをコピーします。

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   以下は例になります。

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

4. 同じブローカー Pod から kafka-reassign-partitions.sh コマンドラインツールを使用して、パーティションの再割り当てを実行します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   レプリケーションをスロットルで調整する場合、--throttle とブローカー間のスロットル率 (バイト/秒単位) を渡すこともできます。以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   このコマンドは、2 つの再割り当て JSON オブジェクトを出力します。最初の JSON オブジェクトには、移動されたパーティションの現在の割り当てが記録されます。後で再割り当てを元に戻す必要がある場合に備え、この値をローカルファイル (Pod のファイル以外) に保存します。2 つ目の JSON オブジェクトは、再割り当て JSON ファイルに渡した目的の再割り当てです。

5. 再割り当ての最中にスロットルを変更する必要がある場合は、同じコマンドラインに別のスロットル率を指定して実行します。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

6. ブローカー Pod のいずれかから kafka-reassign-partitions.sh コマンドラインツールを使用して、再割り当てが完了したかどうかを定期的に確認します。これは先ほどの手順と同じコマンドですが、--execute オプションの代わりに --verify オプションを使用します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

7. --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。割り当てを元のブローカーに戻すために JSON ファイルを保存した場合は、ここでそのファイルを削除できます。

手順

1. 後でコマンドを実行するブローカー Pod に reassignment.json ファイルをコピーします。

   cat reassignment.json | \
     oc exec broker-pod -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

   以下は例になります。

   cat reassignment.json | \
     oc exec my-cluster-kafka-0 -c kafka -i -- /bin/bash -c \
     'cat > /tmp/reassignment.json'

2. 同じブローカー Pod から kafka-reassign-partitions.sh コマンドラインツールを使用して、パーティションの再割り当てを実行します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --execute

   レプリケーションをスロットルで調整する場合、--throttle とブローカー間のスロットル率 (バイト/秒単位) を渡すこともできます。以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 5000000 \
     --execute

   このコマンドは、2 つの再割り当て JSON オブジェクトを出力します。最初の JSON オブジェクトには、移動されたパーティションの現在の割り当てが記録されます。後で再割り当てを元に戻す必要がある場合に備え、この値をローカルファイル (Pod のファイル以外) に保存します。2 つ目の JSON オブジェクトは、再割り当て JSON ファイルに渡した目的の再割り当てです。

3. 再割り当ての最中にスロットルを変更する必要がある場合は、同じコマンドラインに別のスロットル率を指定して実行します。以下は例になります。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --throttle 10000000 \
     --execute

4. ブローカー Pod のいずれかから kafka-reassign-partitions.sh コマンドラインツールを使用して、再割り当てが完了したかどうかを定期的に確認します。これは先ほどの手順と同じコマンドですが、--execute オプションの代わりに --verify オプションを使用します。

   oc exec broker-pod -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

   以下に例を示します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 \
     --reassignment-json-file /tmp/reassignment.json \
     --verify

5. --verify コマンドによって、移動した各パーティションが正常に完了したことが報告されると、再割り当ては終了します。この最終的な --verify によって、結果的に再割り当てスロットルも削除されます。割り当てを元のブローカーに戻すために JSON ファイルを保存した場合は、ここでそのファイルを削除できます。

6. すべてのパーティションの再割り当てが終了すると、削除されるブローカーはクラスター内のいずれのパーティションにも対応しないはずです。これは、ブローカーのデータログディレクトリーにライブパーティションのログが含まれていないことを確認すると検証できます。ブローカーのログディレクトリーに、拡張正規表現 [a-zA-Z0-9.-]+\.[a-z0-9]+-delete$ と一致しないディレクトリーが含まれる場合、ブローカーにはライブパーティションがあるため、停止してはなりません。

   これを確認するには、以下のコマンドを実行します。

   oc exec my-cluster-kafka-0 -c kafka -it -- \
     /bin/bash -c \
     "ls -l /var/lib/kafka/kafka-log_<N>_ | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'"

   N は削除された Pod(s) の数に置き換えます。

   上記のコマンドによって出力が生成される場合、ブローカーにはライブパーティションがあります。この場合、再割り当てが終了していないか、再割り当て JSON ファイルが適切ではありません。

7. ブローカーにライブパーティションがないことが確認できたら、Kafka リソースの Kafka.spec.kafka.replicas を編集できます。これにより、StatefulSet がスケールダウンされ、番号が最も大きいブローカー Pod(s) が削除されます。

手順

1. Kafka リソースの maintenanceTimeWindows プロパティー を追加または編集します。たとえば、0800 から 1059 までと、1400 から 1559 までのメンテナーンスを可能にするには、以下のように maintenanceTimeWindows を設定します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     maintenanceTimeWindows:
       - "* * 8-10 * * ?"
       - "* * 14-15 * * ?"

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. KafkaConnect または KafkaConnectS2I リソースの replicas プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnectS2I
   metadata:
     name: my-cluster
   spec:
     # ...
     replicas: 3
     # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. KafkaConnect または KafkaConnectS2I リソースの bootstrapServers プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-cluster
   spec:
     # ...
     bootstrapServers: my-cluster-kafka-bootstrap:9092
     # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. (任意手順): 認証で使用される TLS 証明書が存在しない場合はファイルで準備し、Secret を作成します。

   注記

   Kafka クラスターの Cluster Operator によって作成されるシークレットは直接使用されることがあります。

   oc create を使用すると作成できます。

   oc create secret generic my-secret --from-file=my-file.crt

2. KafkaConnect または KafkaConnectS2I リソースの tls プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     tls:
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
     # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. (任意手順): 認証に使用される鍵が存在しない場合はファイルで準備し、Secret を作成します。

   注記

   User Operator によって作成されたシークレットを使用できます。

   oc create を使用すると作成できます。

   oc create secret generic my-secret --from-file=my-public.crt --from-file=my-private.key

2. KafkaConnect または KafkaConnectS2I リソースの authentication プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: my-public.crt
         key: my-private.key
     # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. (任意手順): 認証で使用されるパスワードがない場合はファイルで準備し、Secret を作成します。

   注記

   User Operator によって作成されたシークレットを使用できます。

   oc create を使用すると作成できます。

   echo -n '<password>' > <my-password.txt>
   oc create secret generic <my-secret> --from-file=<my-password.txt>

2. KafkaConnect または KafkaConnectS2I リソースの authentication プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     authentication:
       type: scram-sha-512
       username: <my-username>
       passwordSecret:
         secretName: <my-secret>
         password: <my-password.txt>
     # ...

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

   OpenShift では oc apply を使用してこれを行うことができます。

   oc apply -f your-file

手順

1. KafkaConnect または KafkaConnectS2I リソースの config プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect
   metadata:
     name: my-connect
   spec:
     # ...
     config:
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

3. Kafka Connect の承認が有効である場合、Kafka Connect ユーザーを設定し、Kafka Connect のコンシューマーグループおよびトピックへのアクセスを有効にします。

手順

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/v1beta1
   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
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           operation: Read
           host: "*"
         # access to status.storage.topic
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-status
             patternType: literal
           operation: Read
           host: "*"
         # access to config.storage.topic
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Write
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Create
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Describe
           host: "*"
         - resource:
             type: topic
             name: connect-cluster-configs
             patternType: literal
           operation: Read
           host: "*"
         # consumer group
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operation: Read
           host: "*"

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

   oc apply -f your-file

手順

1. クラスターデプロイメントを指定するリソースの resources プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     kafka:
       # ...
       resources:
         requests:
           cpu: "8"
           memory: 64Gi
         limits:
           cpu: "12"
           memory: 128Gi
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、または KafkaConnectS2I の livenessProbe または readinessProbe プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       readinessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       # ...
     zookeeper:
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file

手順

1. Kafka、KafkaConnect、または KafkaConnectS2I リソースの metrics プロパティーを編集します。以下に例を示します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
       metrics:
         lowercaseOutputName: true
       # ...

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

   oc apply を使用して、これを行うことができます。

   oc apply -f your-file