AMQ Streams on OpenShift のデプロイおよびアップグレード

手順

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

   Copy to Clipboard Toggle word wrap

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

   sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml

   Copy to Clipboard Toggle word wrap

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

   oc create -f install/cluster-operator -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

   oc get deployments -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml

   Copy to Clipboard Toggle word wrap

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/amq7/amq-streams-rhel8-operator:2.2.2
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

   Copy to Clipboard Toggle word wrap

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/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>

   Copy to Clipboard Toggle word wrap

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

   oc create -f install/cluster-operator -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

   oc get deployments -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   sed -i '' 's/namespace: .*/namespace: <my_cluster_operator_namespace>/' install/cluster-operator/*RoleBinding*.yaml

   Copy to Clipboard Toggle word wrap

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/amq7/amq-streams-rhel8-operator:2.2.2
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   <my_cluster_operator_namespace> は、Cluster Operator をインストールする namespace に置き換えます。

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

   oc create -f install/cluster-operator -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

   oc get deployments -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

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

手順

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

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

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

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

     Copy to Clipboard Toggle word wrap

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

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

     Copy to Clipboard Toggle word wrap

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

   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 Toggle word wrap

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

   デフォルトのデプロイメントでは、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 Toggle word wrap

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

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

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

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

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

   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 Toggle word wrap

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

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

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

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

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

   oc get pods -n <my_cluster_operator_namespace>

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

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

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

手順

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"

   Copy to Clipboard Toggle word wrap

   1

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

   2

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

   3

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

   4

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

   5

   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

   (オプション) 相互 TLS 認証用のキーストア証明書を生成します。これを false に設定すると、TLS を使用した 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 環境変数で TLS を有効にした場合、Kafka クラスターへの接続を認証するために使用されるキーストアおよびトラストストアを指定します。

   TLS 設定の例

   # ....
   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 Toggle word wrap

   1

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

   2

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

   3

   キーストアには、TLS クライアント認証の秘密鍵が含まれます。

   4

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

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

   oc create -f install/topic-operator

   Copy to Clipboard Toggle word wrap

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

   oc get deployments

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

   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_LOG_LEVEL 

   7

   
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 

   8

   
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 

   9

   
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 

   10

   
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 

   11

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

   12

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

   13

   
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 

   14

   
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 

   15

   
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'

   Copy to Clipboard Toggle word wrap

   1

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

   2

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

   3

   TLS クライアント認証に対して新しいユーザー証明書に署名する認証局の公開鍵 (ca.crt) の値が含まれる OpenShift Secret。

   4

   TLS クライアント認証に対して新しいユーザー証明書に署名する認証局の秘密鍵 (ca.key) の値が含まれる OpenShift Secret。

   5

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

   6

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

   7

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

   8

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

   9

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

   10

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

   11

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

   12

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

   13

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

   14

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

   15

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

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

   TLS 設定の例

   # ....
   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 Toggle word wrap

   1

   TLS クライアント認証に対して Kafka ブローカー証明書に署名する認証局の公開鍵 (ca.crt) の値が含まれる OpenShift Secret。

   2

   Kafka クラスターに対する TLS 認証の秘密鍵と証明書が含まれるキーストア (entity-operator.p12) が含まれる OpenShift Secret。Secret には、キーストアにアクセスするためのパスワード (entity-operator.password) も含まれる必要があります。

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

   oc create -f install/user-operator

   Copy to Clipboard Toggle word wrap

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

   oc get deployments

   Copy to Clipboard Toggle word wrap

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

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

   Copy to Clipboard Toggle word wrap

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

手順

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: debezium-postgres-connector
           artifacts:
             - type: tgz
               url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.3.1.Final/debezium-connector-postgres-1.3.1.Final-plugin.tar.gz
               sha512sum: 962a12151bdf9a5a30627eebac739955a4fd95a08d373b86bdcea2b4d0c27dd6e1edd5cb548045e115e33a9e69b1b2a352bee24df035a0447cb820077af00c03
         - name: camel-telegram
           artifacts:
             - type: tgz
               url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.7.0/camel-telegram-kafka-connector-0.7.0-package.tar.gz
               sha512sum: a9b1ac63e3284bea7836d7d24d84208c49cdf5600070e6bd1535de654f6920b74ad950d51733e8020bf4187870699819f54ef5859c7846ee4081507f48873479
     #...

   Copy to Clipboard Toggle word wrap

   1

   Kafka Connect クラスターの仕様。

   2

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

   3

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

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

   $ oc apply -f KAFKA-CONNECT-CONFIG-FILE

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

手順

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

   FROM registry.redhat.io/amq7/amq-streams-kafka-32-rhel8:2.2.2
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   Copy to Clipboard Toggle word wrap

   プラグインファイルの例

   $ 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

   Copy to Clipboard Toggle word wrap

   注記

   この例では、MongoDB、MySQL、および PostgreSQL 用の Debezium コネクターを使用します。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 Toggle word wrap

     1

     Kafka Connect クラスターの仕様。

     2

     Pod の Docker イメージ。

     3

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

     または

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