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

手順

1. Kafka Connect を OpenShift クラスターにデプロイします。3 つ以上のブローカーで構成される Kafka クラスターの場合は、examples/connect/kafka-connect.yaml ファイルを使用します。3 つ未満のブローカーで構成される Kafka クラスターの場合は、examples/connect/kafka-connect-single-node-kafka.yaml ファイルを使用します。

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

2. Kafka Connect が正常にデプロイされたことを確認します。

   oc get deployments

手順

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

   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
     config: 5
       file: "/opt/kafka/LICENSE" 6
       topic: my-topic 7
       # ...

   1

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

   2

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

   3

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

   4

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

   5

   キーと値のペアとしてのコネクター設定。

   6

   このサンプルソースコネクター設定では、/opt/kafka/LICENSE ファイルからデータが読み取られます。

   7

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

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

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

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

   touch examples/connect/sink-connector.yaml

4. 以下の 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

   1

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

   2

   キーと値のペアとしてのコネクター設定。

   3

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

   4

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

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

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

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

   oc get kctr --selector strimzi.io/cluster=MY-CONNECT-CLUSTER -o name
   
   my-source-connector
   my-sink-connector

   MY-CONNECT-CLUSTER を Kafka Connect クラスターに置き換えます。

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

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

手順

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

   oc get KafkaConnector

2. コネクターを再起動するには、OpenShift で KafkaConnector リソースにアノテーションを付けます。以下は oc annotate を使用した例になります。

   oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart=true

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

   アノテーションが調整プロセスで検出されれば、Kafka コネクターは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

手順

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

   oc get KafkaConnector

2. KafkaConnector カスタムリソースから再起動するタスクの ID を検索します。タスク ID は 0 から始まる負の値ではない整数です。

   oc describe KafkaConnector KAFKACONNECTOR-NAME

3. コネクタータスクを再起動するには、OpenShift で KafkaConnector リソースにアノテーションを付けます。たとえば、oc annotate を使用してタスク 0 を再起動します。

   oc annotate KafkaConnector KAFKACONNECTOR-NAME strimzi.io/restart-task=0

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

   アノテーションが調整プロセスで検出されれば、Kafka コネクタータスクは再起動されます。Kafka Connect が再起動リクエストを受け入れると、アノテーションは KafkaConnector カスタムリソースから削除されます。

手順

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

   MirrorMaker の場合

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

   MirrorMaker 2.0 の場合

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

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

   oc get deployments

手順

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

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

2. Kafka Bridge が正常にデプロイされたことを確認します。

   oc get deployments

1. エディターで Kafka リソースを更新します。

   oc edit kafka KAFKA-CONFIG-FILE

2. kafka-metrics.yaml の設定例を、ユーザーの Kafka リソース定義にコピーします。
3. ファイルを保存し、更新したリソースが調整されるのを待ちます。

手順

1. ユーザー定義プロジェクトのモニタリングが有効であることを確認します。

   oc get pods -n openshift-user-workload-monitoring

   有効であると、モニタリングコンポーネントの Pod が返されます。以下に例を示します。

   NAME                                   READY   STATUS    RESTARTS   AGE
   prometheus-operator-5cc59f9bc6-kgcq8   1/1     Running   0          25s
   prometheus-user-workload-0             5/5     Running   1          14s
   prometheus-user-workload-1             5/5     Running   1          14s
   thanos-ruler-user-workload-0           3/3     Running   0          14s
   thanos-ruler-user-workload-1           3/3     Running   0          14s

   Pod が返されなければ、ユーザー定義プロジェクトのモニタリングは無効になっています。「OpenShift 4 での Kafka メトリクスおよびダッシュボードの表示」 の前提条件を参照してください。

2. 複数の PodMonitor リソースが examples/metrics/prometheus-install/strimzi-pod-monitor.yamlで定義されています。

   PodMonitor リソースごとに、spec.namespaceSelector.matchNames プロパティーを編集します。

   apiVersion: monitoring.coreos.com/v1
   kind: PodMonitor
   metadata:
     name: cluster-operator-metrics
     labels:
       app: strimzi
   spec:
     selector:
       matchLabels:
         strimzi.io/kind: cluster-operator
     namespaceSelector:
       matchNames:
         - PROJECT-NAME 1
     podMetricsEndpoints:
     - path: /metrics
       port: http
   # ...

   1

   メトリクスをスクレープする Pod が実行されているプロジェクト (例: Kafka)。

3. strimzi-pod-monitor.yaml ファイルを、Kafka クラスターが稼働しているプロジェクトにデプロイします。

   oc apply -f strimzi-pod-monitor.yaml -n MY-PROJECT

4. Prometheus ルールのサンプルを同じプロジェクトにデプロイします。

   oc apply -f prometheus-rules.yaml -n MY-PROJECT

手順

1. Grafana の ServiceAccount を作成します。ここでは、リソースの名前は grafana-serviceaccount です。

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: grafana-serviceaccount
     labels:
       app: strimzi

2. ServiceAccount を Kafka クラスターが含まれるプロジェクトにデプロイします。

   oc apply -f GRAFANA-SERVICEACCOUNT -n MY-PROJECT

3. cluster-monitoring-view ロールを Grafana ServiceAccount に割り当てる ClusterRoleBinding リソースを作成します。ここでは、リソースの名前は grafana-cluster-monitoring-binding です。

   apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRoleBinding
   metadata:
     name: grafana-cluster-monitoring-binding
     labels:
       app: strimzi
   subjects:
     - kind: ServiceAccount
       name: grafana-serviceaccount
       namespace: MY-PROJECT 1
   roleRef:
     kind: ClusterRole
     name: cluster-monitoring-view
     apiGroup: rbac.authorization.k8s.io

   1

   プロジェクトの名前。

4. ClusterRoleBinding を Kafka クラスターが含まれるプロジェクトにデプロイします。

   oc apply -f GRAFANA-CLUSTER-MONITORING-BINDING -n MY-PROJECT

手順

1. Grafana ServiceAccount のアクセストークンを取得します。

   oc serviceaccounts get-token grafana-serviceaccount -n MY-PROJECT

   次のステップで使用するアクセストークンをコピーします。

2. Grafana の Thanos Querier 設定が含まれる datasource.yaml ファイルを作成します。

   以下に示すように、アクセストークンを httpHeaderValue1 プロパティーに貼り付けます。

   apiVersion: 1
   
   datasources:
   - name: Prometheus
     type: prometheus
     url: https://thanos-querier.openshift-monitoring.svc.cluster.local:9091
     access: proxy
     basicAuth: false
     withCredentials: false
     isDefault: true
     jsonData:
       timeInterval: 5s
       tlsSkipVerify: true
       httpHeaderName1: "Authorization"
     secureJsonData:
       httpHeaderValue1: "Bearer ${GRAFANA-ACCESS-TOKEN}" 1
     editable: true

   1

   GRAFANA-ACCESS-TOKEN: Grafana ServiceAccount のアクセストークンの値

3. grafana-config ファイルから datasource.yaml という名前の Config Map を作成します。

   oc create configmap grafana-config --from-file=datasource.yaml -n MY-PROJECT

4. Deployment および Service で構成される Grafana アプリケーションを作成します。

   grafana-config Config Map はデータソース設定のボリュームとしてマウントされます。

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     replicas: 1
     selector:
       matchLabels:
         name: grafana
     template:
       metadata:
         labels:
           name: grafana
       spec:
         serviceAccountName: grafana-serviceaccount
         containers:
         - name: grafana
           image: grafana/grafana:6.3.0
           ports:
           - name: grafana
             containerPort: 3000
             protocol: TCP
           volumeMounts:
           - name: grafana-data
             mountPath: /var/lib/grafana
           - name: grafana-logs
             mountPath: /var/log/grafana
           - name: grafana-config
             mountPath: /etc/grafana/provisioning/datasources/datasource.yaml
             readOnly: true
             subPath: datasource.yaml
           readinessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 5
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 15
             periodSeconds: 20
         volumes:
         - name: grafana-data
           emptyDir: {}
         - name: grafana-logs
           emptyDir: {}
         - name: grafana-config
           configMap:
             name: grafana-config
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     ports:
     - name: grafana
       port: 3000
       targetPort: 3000
       protocol: TCP
     selector:
       name: grafana
     type: ClusterIP

5. Grafana アプリケーションを、Kafka クラスターが含まれるプロジェクトにデプロイします。

   oc apply -f GRAFANA-APPLICATION -n MY-PROJECT

手順

1. Grafana サービスへのルートの詳細を取得します。以下に例を示します。

   oc get routes
   
   NAME               HOST/PORT                         PATH  SERVICES
   MY-GRAFANA-ROUTE   MY-GRAFANA-ROUTE-amq-streams.net        grafana

2. Web ブラウザーで、Route ホストおよびポートの URL を使用して Grafana ログイン画面にアクセスします。

3. ユーザー名とパスワードを入力し、続いて Log In をクリックします。

   デフォルトの Grafana ユーザー名およびパスワードは、どちらも admin です。初回ログイン後に、パスワードを変更できます。

4. Configuration > Data Sources で、Prometheus データソースが作成済みであることを確認します。データソースは 「Prometheus データソースを使用した Grafana のデプロイ」 に作成されています。
5. Dashboards > Manage をクリックしてから Import をクリックします。
6. examples/metrics/grafana-dashboards で、インポートするダッシュボードの JSON をコピーします。
7. JSON をテキストボックスに貼り付け、Load をクリックします。
8. 他の Grafana ダッシュボードのサンプルに、ステップ 1 -7 を繰り返します。

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     # ...
     kafkaExporter:
       image: my-registry.io/my-org/my-exporter-cluster: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. Grafana ユーザーインターフェースにアクセスします。

2. Strimzi Kafka Exporter ダッシュボードを選択します。

   メトリクスデータが収集されると、Kafka Exporter のチャートにデータが反映されます。

   AMQ Streams Kafka Exporter

   以下のメトリクスを表示します。

   • トピックの数
   • パーティションの数
   • レプリカの数
   • In-Sync レプリカの数
   • 複製の数が最低数未満であるパーティションの数
   • 最小の In-Sync レプリカ数にあるパーティション
   • 最小の In-Sync レプリカ数未満のパーティション
   • 優先ノードにないパーティション
   • 毎秒のトピックからのメッセージ
   • 毎秒消費されるトピックからのメッセージ
   • コンシューマーグループごとに毎分消費されるメッセージ
   • コンシューマーグループごとのラグ
   • パーティションの数
   • 最新のオフセット
   • 最も古いオフセット

手順

1. convert-file コマンドおよび適切なフラグで、API 変換ツールを実行します。

   例 1: YAML ファイルを変換し、出力を表示しますが、ファイルは変更されません。

   bin/api-conversion.sh convert-file --file input.yaml

   例 2: YAML ファイルを変換し、変更を元のソースファイルに書き込みます。

   bin/api-conversion.sh convert-file --file input.yaml --in-place

   例 3: YAML ファイルを変換し、変更を新しい出力ファイルに書き込みます。

   bin/api-conversion.sh convert-file --file input.yaml --output output.yaml

2. 変換した設定ファイルを使用して、カスタムリソースを更新します。

   oc apply -f CONVERTED-CONFIG-FILE

3. カスタムリソースが変換されたことを確認します。

   oc get KIND CUSTOM-RESOURCE-NAME -o yaml

手順

1. convert-resource コマンドおよび適切なフラグで、API 変換ツールを実行します。

   例 1: 現在の namespace のすべての AMQ Streams リソースを変換します。

   bin/api-conversion.sh convert-resource

   例 2: すべての namespace のすべての AMQ Streams リソースを変換します。

   bin/api-conversion.sh convert-resource --all-namespaces

   例 3: my-kafka namespace のすべての AMQ Streams リソースを変換します。

   bin/api-conversion.sh convert-resource --namespace my-kafka

   例 4: すべての namespace の Kafka リソースのみを変換します。

   bin/api-conversion.sh convert-resource --all-namespaces --kind Kafka

   例 5: すべての namespace の Kafka および Kafka Connect リソースを変換します。

   bin/api-conversion.sh convert-resource --all-namespaces --kind Kafka --kind KafkaConnect

   例 6: my-kafka namespace の my-cluster という名前の Kafka カスタムリソースを変換します。

   bin/api-conversion.sh convert-resource --kind Kafka --namespace my-kafka --name my-cluster

2. カスタムリソースが変換されたことを確認します。

   oc get KIND CUSTOM-RESOURCE-NAME -o yaml

手順

1. v1beta2 を使用するようにカスタムリソースを変換していない場合は、そのように変換します。

   これは、API 変換ツールを使用して、以下のいずれかの方法で行います。

   • 「API 変換ツールを使用したカスタムリソース設定ファイルの変換」

   • 「API 変換ツールを使用したカスタムリソースの直接変換」

     また、手動で変更することもできます。

2. crd-upgrade コマンドを使用して、API 変換ツールを実行します。

   bin/api-conversion.sh crd-upgrade

3. CRD がアップグレードされ、v1beta2 がストレージバージョンであることを確認します。

   たとえば、Kafka トピック CRD の場合は次のとおりです。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: CustomResourceDefinition
   metadata:
     name: kafkatopics.kafka.strimzi.io
     #...
   spec:
     group: kafka.strimzi.io
     #...
     versions:
     - name: v1beta2
       served: true
       storage: true
       #...
   status:
     #...
     storedVersions:
     - v1beta2

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. .spec.kafka.listener を新しい汎用リスナー形式に更新していない場合は、「リスナーの汎用リスナー設定への更新」 の説明に従って更新します。

   警告

   API バージョン v1beta2 では、古いリスナーの形式はサポートされません。

3. affinity が .spec.kafka.affinity に存在する場合は .spec.kafka.template.pod.affinity に移動します。
4. tolerations が .spec.kafka.tolerations に存在する場合は .spec.kafka.template.pod.tolerations に移動します。
5. .spec.kafka.template.tlsSidecarContainer が存在する場合は削除します。
6. .spec.kafka.tlsSidecarContainer が存在する場合は削除します。

7. 以下のポリシー設定のいずれかが存在する場合:

   • .spec.kafka.template.externalBootstrapService.externalTrafficPolicy

   • .spec.kafka.template.perPodService.externalTrafficPolicy

     1. type: loadbalancer および type: nodeport リスナー両方の設定を .spec.kafka.listeners[].configuration.externalTrafficPolicy に移動します。
     2. .spec.kafka.template.externalBootstrapService.externalTrafficPolicy または .spec.kafka.template.perPodService.externalTrafficPolicy を削除します。

8. 以下の loadbalancer リスナー設定のいずれかが存在する場合:

   • .spec.kafka.template.externalBootstrapService.loadBalancerSourceRanges

   • .spec.kafka.template.perPodService.loadBalancerSourceRanges

     1. type: loadbalancer リスナーの設定を .spec.kafka.listeners[].configuration.loadBalancerSourceRanges に移動します。
     2. .spec.kafka.template.externalBootstrapService.loadBalancerSourceRanges または .spec.kafka.template.perPodService.loadBalancerSourceRanges を削除します。

9. type: external ロギングが .spec.kafka.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

10. .spec.kafka.metrics フィールドを使用してメトリクスを有効にする場合:

    1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.kafka.metrics フィールドの内容と一致している必要があります。

       kind: ConfigMap
       apiVersion: v1
       metadata:
         name: kafka-metrics
         labels:
           app: strimzi
       data:
         kafka-metrics-config.yaml: |
             <YAML>

    2. ConfigMap およびキーを示す .spec.kafka.metricsConfig プロパティーを追加します。

       metricsConfig:
         type: jmxPrometheusExporter
         valueFrom:
           configMapKeyRef:
             name: kafka-metrics
             key: kafka-metrics-config.yaml

    3. 古い .spec.kafka.metrics フィールドを削除します。
11. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. affinity が .spec.zookeeper.affinity に存在する場合は .spec.zookeeper.template.pod.affinity に移動します。
3. tolerations が .spec.zookeeper.tolerations に存在する場合は .spec.zookeeper.template.pod.tolerations に移動します。
4. .spec.zookeeper.template.tlsSidecarContainer が存在する場合は削除します。
5. .spec.zookeeper.tlsSidecarContainer が存在する場合は削除します。

6. type: external ロギングが .spec.kafka.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

7. .spec.zookeeper.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.zookeeper.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-metrics
        labels:
          app: strimzi
      data:
        zookeeper-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.zookeeper.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-metrics
            key: zookeeper-metrics-config.yaml

   3. 古い .spec.zookeeper.metrics フィールドを削除します。
8. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. Kafka.spec.topicOperator を使用する場合:

   1. affinity を .spec.topicOperator.affinity から .spec.entityOperator.template.pod.affinity に移動します。
   2. tolerations を .spec.topicOperator.tolerations から .spec.entityOperator.template.pod.tolerations に移動します。
   3. .spec.topicOperator.tlsSidecar を .spec.entityOperator.tlsSidecar に移動します。
   4. affinity、tolerations、および tlsSidecar の移動後に、.spec.topicOperator の残りの設定を .spec.entityOperator.topicOperator に移動します。

3. type: external ロギングが .spec.topicOperator.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j2.properties

   注記

   このステップは、Entity Operator のアップグレード の一部として完了することもできます。

4. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. affinity を .spec.entityOperator.affinity から .spec.entityOperator.template.pod.affinity に移動します。
3. tolerations を .spec.entityOperator.tolerations から .spec.entityOperator.template.pod.tolerations に移動します。

4. type: external ロギングが .spec.entityOperator.userOperator.logging または .spec.entityOperator.topicOperator.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j2.properties

5. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. type: external ロギングが .spec.cruiseControl.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j2.properties

3. .spec.cruiseControl.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.cruiseControl.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-metrics
        labels:
          app: strimzi
      data:
        cruise-control-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.cruiseControl.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-metrics
            key: cruise-control-metrics-config.yaml

   3. 古い .spec.cruiseControl.metrics フィールドを削除します。
4. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで Kafka カスタムリソースを更新します。

   oc edit kafka KAFKA-CLUSTER

2. Kafka カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

3. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaConnect カスタムリソースを更新します。

   oc edit kafkaconnect KAFKA-CONNECT-CLUSTER

2. 以下があるか確認します。

   KafkaConnect.spec.affinity

   KafkaConnect.spec.tolerations

   あれば以下に変更します。

   KafkaConnect.spec.template.pod.affinity

   KafkaConnect.spec.template.pod.tolerations

   たとえば、以下の場合を考えてみましょう。

   spec:
     # ...
     affinity:
       # ...
     tolerations:
       # ...

   これを以下に変更します。

   spec:
     # ...
     template:
       pod:
         affinity:
           # ...
         tolerations:
           # ...

3. type: external ロギングが .spec.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

4. .spec.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-connect-metrics
        labels:
          app: strimzi
      data:
        connect-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-connect-metrics
            key: connect-metrics-config.yaml

   3. 古い .spec.metrics フィールドを削除します。

5. KafkaConnect カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

6. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaConnectS2I カスタムリソースを更新します。

   oc edit kafkaconnects2i S2I-CLUSTER

2. 以下があるか確認します。

   KafkaConnectS2I.spec.affinity

   KafkaConnectS2I.spec.tolerations

   あれば以下に変更します。

   KafkaConnectS2I.spec.template.pod.affinity

   KafkaConnectS2I.spec.template.pod.tolerations

   たとえば、以下の場合を考えてみましょう。

   spec:
     # ...
     affinity:
       # ...
     tolerations:
       # ...

   これを以下に変更します。

   spec:
     # ...
     template:
       pod:
         affinity:
           # ...
         tolerations:
           # ...

3. type: external ロギングが .spec.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

4. .spec.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-connect-s2i-metrics
        labels:
          app: strimzi
      data:
        connect-s2i-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-connect-s2i-metrics
            key: connect-s2i-metrics-config.yaml

   3. 古い .spec.metrics フィールドを削除します。

5. KafkaConnectS2I カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

6. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaMirrorMaker カスタムリソースを更新します。

   oc edit kafkamirrormaker MIRROR-MAKER

2. 以下があるか確認します。

   KafkaMirrorMaker.spec.affinity

   KafkaMirrorMaker.spec.tolerations

   あれば以下に変更します。

   KafkaMirrorMaker.spec.template.pod.affinity

   KafkaMirrorMaker.spec.template.pod.tolerations

   たとえば、以下の場合を考えてみましょう。

   spec:
     # ...
     affinity:
       # ...
     tolerations:
       # ...

   これを以下に変更します。

   spec:
     # ...
     template:
       pod:
         affinity:
           # ...
         tolerations:
           # ...

3. type: external ロギングが .spec.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

4. .spec.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-mm-metrics
        labels:
          app: strimzi
      data:
        mm-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-mm-metrics
            key: mm-metrics-config.yaml

   3. 古い .spec.metrics フィールドを削除します。

5. KafkaMirrorMaker カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

6. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaMirrorMaker2 カスタムリソースを更新します。

   oc edit kafkamirrormaker2 MIRROR-MAKER-2

2. affinity が .spec.affinity に存在する場合は .spec.template.pod.affinity に移動します。
3. tolerations が .spec.tolerations に存在する場合は .spec.template.pod.tolerations に移動します。

4. type: external ロギングが .spec.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j.properties

5. .spec.metrics フィールドを使用してメトリクスを有効にする場合:

   1. JMX Prometheus エクスポーターの YAML 設定をキーの下に保存する新しい ConfigMap を作成します。YAML は .spec.metrics フィールドの内容と一致している必要があります。

      kind: ConfigMap
      apiVersion: v1
      metadata:
        name: kafka-mm2-metrics
        labels:
          app: strimzi
      data:
        mm2-metrics-config.yaml: |
            <YAML>

   2. ConfigMap およびキーを示す .spec.metricsConfig プロパティーを追加します。

      metricsConfig:
        type: jmxPrometheusExporter
        valueFrom:
          configMapKeyRef:
            name: kafka-mm2-metrics
            key: mm2-metrics-config.yaml

   3. 古い .spec.metrics フィールドを削除します。

6. KafkaMirrorMaker2 カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1alpha1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

7. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaBridge カスタムリソースを更新します。

   oc edit kafkabridge KAFKA-BRIDGE

2. type: external ロギングが KafkaBridge.spec.logging で設定されている場合:

   ロギング設定が含まれる ConfigMap の name を置き換えます。

   logging:
     type: external
     name: my-config-map

   valueFrom.configMapKeyRef フィールドで、ログ保存される ConfigMap の name および key の両方を指定します。

   logging:
     type: external
     valueFrom:
       configMapKeyRef:
         name: my-config-map
         key: log4j2.properties

3. KafkaBridge カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1alpha1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

4. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaUser カスタムリソースを更新します。

   oc edit kafkauser KAFKA-USER

2. KafkaUser カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

3. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaTopic カスタムリソースを更新します。

   oc edit kafkatopic KAFKA-TOPIC

2. KafkaTopic カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1beta1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

3. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaConnector カスタムリソースを更新します。

   oc edit kafkaconnector KAFKA-CONNECTOR

2. KafkaConnector カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1alpha1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

3. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

1. エディターで KafkaRebalance カスタムリソースを更新します。

   oc edit kafkarebalance KAFKA-REBALANCE

2. KafkaRebalance カスタムリソースの apiVersion を v1beta2 に更新します。

   以下の行を見つけます。

   apiVersion: kafka.strimzi.io/v1alpha1

   この行を以下の行に変更します。

   apiVersion: kafka.strimzi.io/v1beta2

3. ファイルを保存し、エディターを終了して更新したカスタムリソースが調整されるのを待ちます。

手順

1. Kafka クラスター設定を更新します。

   oc edit kafka KAFKA-CONFIGURATION-FILE

2. Kafka.spec.kafka.version を変更して、以前のバージョンを指定します。

   たとえば、Kafka 2.7.0 から 2.6.0 へのダウングレードは以下のようになります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0 1
       config:
         log.message.format.version: "2.6" 2
         inter.broker.protocol.version: "2.6" 3
         # ...

   1

   Kafka のバージョンが以前のバージョンに変更されます。

   2

   メッセージ形式のバージョンは変更されません。

   3

   ブローカー間のプロトコルバージョンは変更されません。

   注記

   log.message.format.version および inter.broker.protocol.version の値は、浮動小数点数として解釈されないように文字列にする必要があります。

3. Kafka バージョンのイメージが Cluster Operator の STRIMZI_KAFKA_IMAGES に定義されているイメージとは異なる場合は、Kafka.spec.kafka.image を更新します。

   「Kafka バージョンおよびイメージマッピング」 を参照してください。

4. エディターを保存して終了し、ローリングアップデートの完了を待ちます。

   更新をログで確認するか、または Pod 状態の遷移を監視して確認します。

   oc logs -f CLUSTER-OPERATOR-POD-NAME | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"

   oc get pod -w

   Cluster Operator のログで INFO レベルのメッセージを確認します。

   Reconciliation #NUM(watch) Kafka(NAMESPACE/NAME): Kafka version downgrade from FROM-VERSION to TO-VERSION, phase 1 of 1 completed

5. すべてのクライアントアプリケーション (コンシューマー) をダウングレードして、以前のバージョンのクライアントバイナリーを使用します。

   これで、Kafka クラスターおよびクライアントは以前の Kafka バージョンを使用するようになります。

6. トピックメタデータの保存に ZooKeeper を使用する 0.22 よりも前のバージョンの AMQ Streams に戻す場合は、Kafka クラスターから内部トピックストアのトピックを削除します。

   oc run kafka-admin -ti --image=registry.redhat.io/amq7/amq-streams-kafka-27-rhel7:1.7.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete