OpenShift 上の Streams for Apache Kafka のデプロイと管理

手順

1. Cluster Operator がインストールされる namespace を使用するように、Streams for Apache Kafka インストールファイルを編集します。

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

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

手順

1. Cluster Operator がインストールされる namespace を使用するように、Streams for Apache Kafka インストールファイルを編集します。

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

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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-rhel9-operator:2.7.0
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

   Copy to Clipboard

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>

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

手順

1. Cluster Operator がインストールされる namespace を使用するように、Streams for Apache Kafka インストールファイルを編集します。

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

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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-rhel9-operator:2.7.0
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

   Copy to Clipboard

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

   Copy to Clipboard

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

手順

1. KRaft ベースの Kafka クラスターをデプロイします。

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

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

     Copy to Clipboard

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

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

     Copy to Clipboard

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

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

     Copy to Clipboard

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

     oc apply -f examples/kafka/kafka-with-node-pools.yaml

     Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

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

   Copy to Clipboard

   • 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. ZooKeeper ベースの Kafka クラスターをデプロイします。

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

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

     Copy to Clipboard

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

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

     Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

   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

   Copy to Clipboard

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

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

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

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

手順

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

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

   Copy to Clipboard

2. EntityTopicOperatorSpec スキーマリファレンス で説明されているプロパティーを使用して、Topic Operator の spec を設定します。

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

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

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

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

   Copy to Clipboard

   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: {}

   Copy to Clipboard

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

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

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

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

手順

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

   以下に例を示します。

   oc exec -ti my-cluster-zookeeper-0 -- bin/zookeeper-shell.sh localhost:12181 ls /

   Copy to Clipboard

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

手順

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

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

   Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

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

   Copy to Clipboard

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

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

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

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

手順

1. .spec.build.output でコンテナーレジストリーを、.spec.build.plugins で追加のコネクターを指定して、KafkaConnect カスタムリソースを設定します。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 

   1

   
     #...
     build:
       output: 

   2

   
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 

   3

   
         - name: connector-1
           artifacts:
             - type: tgz
               url: <url_to_download_connector_1_artifact>
               sha512sum: <SHA-512_checksum_of_connector_1_artifact>
         - name: connector-2
           artifacts:
             - type: jar
               url: <url_to_download_connector_2_artifact>
               sha512sum: <SHA-512_checksum_of_connector_2_artifact>
     #...

   Copy to Clipboard

   1

   Kafka Connect クラスターの仕様。

   2

   (必須) 新しいイメージがプッシュされるコンテナーレジストリーの設定。

   3

   (必須) 新しいコンテナーイメージに追加するコネクタープラグインとそれらのアーティファクトのリスト。各プラグインは、1 つ以上の artifact を使用して設定する必要があります。

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

   $ oc apply -f <kafka_connect_configuration_file>

   Copy to Clipboard
3. 新しいコンテナーイメージがビルドされ、Kafka Connect クラスターがデプロイされるまで待ちます。
4. Kafka Connect REST API または KafkaConnector カスタムリソースを使用して、追加したコネクタープラグインを使用します。

手順

1. ベースイメージとして registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 を使用して、新しい Dockerfile を作成します。

   FROM registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   Copy to Clipboard

   プラグインファイルの例

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   Copy to Clipboard

   COPY コマンドは、コンテナーイメージにコピーするプラグインファイルを指します。

   この例では、Debezium コネクター (MongoDB、MySQL、および PostgreSQL) のプラグインを追加しますが、簡潔にするためにすべてのファイルがリストされているわけではありません。Kafka Connect で実行されている Debezium は、他の Kafka Connect タスクと同じように表示されます。

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

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

   次のいずれかの方法でイメージを指定できます。

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

     設定されている場合、このプロパティーは Cluster Operator の STRIMZI_KAFKA_CONNECT_IMAGES 環境変数をオーバーライドします。

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 

     1

     
       #...
       image: my-new-container-image 

     2

     
       config: 

     3

     
         #...

     Copy to Clipboard

     1

     Kafka Connect クラスターの仕様。

     2

     Kafka Connect Pod の Docker イメージ。

     3

     Kafka Connect ワーカー (コネクターではない) の設定。

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

手順

1. 次のいずれかの方法で、FileStreamSourceConnector および FileStreamSinkConnector プラグインを Kafka Connect に追加します。

   • プラグインを使用して新しいコンテナーイメージを自動的にビルドするように Kafka Connect を設定する
   • ベースの Kafka Connect イメージから Docker イメージを作成します (手動または継続的インテグレーションを使用)

2. Kafka Connect 設定で strimzi.io/use-connector-resources annotation を true に設定します。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true"
   spec:
       # ...

   Copy to Clipboard

   KafkaConnector リソースを有効にすると、Cluster Operator はそれらを監視します。

3. examples/connect/source-connector.yaml ファイルを編集します。

   KafkaConnector ソースコネクター設定の例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector  

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster 

   2

   
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 

   3

   
     tasksMax: 2 

   4

   
     autoRestart: 

   5

   
       enabled: true
     config: 

   6

   
       file: "/opt/kafka/LICENSE" 

   7

   
       topic: my-topic 

   8

   
       # ...

   Copy to Clipboard

   1

   コネクターの名前として使用される KafkaConnector リソースの名前。OpenShift リソースで有効な名前を使用します。

   2

   コネクターインスタンスを作成する Kafka Connect クラスターの名前。コネクターは、リンク先の Kafka Connect クラスターと同じ namespace にデプロイする必要があります。

   3

   コネクタークラスのフルネーム。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。

   4

   コネクターが作成できる Kafka Connect タスクの最大数。

   5

   失敗したコネクターとタスクの自動再起動を有効にします。デフォルトでは、再起動の回数は無制限ですが、maxRestarts プロパティーを使用して自動再起動の最大回数を設定できます。

   6

   キーと値のペア形式の コネクター設定。

   7

   外部データファイルの場所。この例では、/opt/kafka/LICENSE ファイルから読み取るように FileStreamSourceConnector を設定しています。

   8

   ソースデータのパブリッシュ先となる Kafka トピック。

4. OpenShift クラスターでソース KafkaConnector を作成します。

   oc apply -f examples/connect/source-connector.yaml

   Copy to Clipboard

5. examples/connect/sink-connector.yaml ファイルを作成します。

   touch examples/connect/sink-connector.yaml

   Copy to Clipboard

6. 以下の YAML を sink-connector.yaml ファイルに貼り付けます。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 

   1

   
     tasksMax: 2
     config: 

   2

   
       file: "/tmp/my-file" 

   3

   
       topics: my-topic 

   4

   Copy to Clipboard

   1

   コネクタークラスのフルネームまたはエイリアス。これは、Kafka Connect クラスターによって使用されているイメージに存在するはずです。

   2

   キーと値のペア形式の コネクター設定。

   3

   ソースデータのパブリッシュ先となる一時ファイル。

   4

   ソースデータの読み取り元となる Kafka トピック。

7. OpenShift クラスターにシンク KafkaConnector を作成します。

   oc apply -f examples/connect/sink-connector.yaml

   Copy to Clipboard

8. コネクターリソースが作成されたことを確認します。

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   Copy to Clipboard

   <my_connect_cluster> を Kafka Connect クラスターの名前に置き換えます。

9. コンテナーで、kafka -console-consumer.sh を実行して、ソースコネクターによってトピックに書き込まれたメッセージを読み取ります。

   oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   Copy to Clipboard

   <my_kafka_cluster> を Kafka クラスターの名前に置き換えます。

1. 設定で KafkaConnector リソースをデプロイし、コネクターインスタンスを作成します。
2. strimzi.io/use-connector-resources アノテーションを true に設定して、Kafka Connect 設定で KafkaConnector リソースを有効にします。

手順

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

   MirrorMaker の場合

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

   Copy to Clipboard

   MirrorMaker 2 の場合:

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

   Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

   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

   Copy to Clipboard

   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

   Copy to Clipboard

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard

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

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

   Copy to Clipboard

   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>

   Copy to Clipboard

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

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

   Copy to Clipboard
   注記

   ローカルマシンのポート 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_FULL_RECONCILIATION_INTERVAL_MS 

   4

   
                 value: "120000"
               - name: STRIMZI_LOG_LEVEL 

   5

   
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 

   6

   
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 

   7

   
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 

   8

   
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 

   9

   
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 

   10

   
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 

   11

   
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 

   12

   
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 

   13

   
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 

   14

   
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 

   15

   
                 value: "SSL"
               - name: STRIMZI_USE_FINALIZERS
                 value: "false" 

   16

   Copy to Clipboard

   1

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

   2

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

   3

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

   4

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

   5

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

   6

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

   7

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

   8

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

   9

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

   10

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

   11

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

   12

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

   13

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

   14

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

   15

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

   16

   トピックの削除 を制御するためにファイナライザーを使用しない場合は、STRIMZI_USE_FINALIZERS を false に設定します。

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"
   # ...

   Copy to Clipboard

   1

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

   2

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

   3

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

   4

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

4. 変更を Deployment 設定に適用して、Topic Operator をデプロイします。

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

   oc get deployments

   Copy to Clipboard

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

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

   Copy to Clipboard

   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

   Copy to Clipboard

   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
   # ..."

   Copy to Clipboard

   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

   Copy to Clipboard

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

   oc get deployments

   Copy to Clipboard

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

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

   Copy to Clipboard

   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

   Copy to Clipboard

   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>

   Copy to Clipboard