OpenShift でのAMQ Streams の設定

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       replicas: 3 1
       version: 3.1.0 2
       logging: 3
         type: inline
         loggers:
           kafka.root.logger.level: "INFO"
       resources: 4
         requests:
           memory: 64Gi
           cpu: "8"
         limits:
           memory: 64Gi
           cpu: "12"
       readinessProbe: 5
         initialDelaySeconds: 15
         timeoutSeconds: 5
       livenessProbe:
         initialDelaySeconds: 15
         timeoutSeconds: 5
       jvmOptions: 6
         -Xms: 8192m
         -Xmx: 8192m
       image: my-org/my-image:latest 7
       listeners: 8
         - name: plain 9
           port: 9092 10
           type: internal 11
           tls: false 12
           configuration:
             useServiceDnsDomain: true 13
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication: 14
             type: tls
         - name: external 15
           port: 9094
           type: route
           tls: true
           configuration:
             brokerCertChainAndKey: 16
               secretName: my-secret
               certificate: my-certificate.crt
               key: my-key.key
       authorization: 17
         type: simple
       config: 18
         auto.create.topics.enable: "false"
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 2
         default.replication.factor: 3
         min.insync.replicas: 2
         inter.broker.protocol.version: "3.1"
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 19
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
       storage: 20
         type: persistent-claim 21
         size: 10000Gi 22
       rack: 23
         topologyKey: topology.kubernetes.io/zone
       metricsConfig: 24
         type: jmxPrometheusExporter
         valueFrom:
           configMapKeyRef: 25
             name: my-config-map
             key: my-key
       # ...
     zookeeper: 26
       replicas: 3 27
       logging: 28
         type: inline
         loggers:
           zookeeper.root.logger: "INFO"
       resources:
         requests:
           memory: 8Gi
           cpu: "2"
         limits:
           memory: 8Gi
           cpu: "2"
       jvmOptions:
         -Xms: 4096m
         -Xmx: 4096m
       storage:
         type: persistent-claim
         size: 1000Gi
       metricsConfig:
         # ...
     entityOperator: 29
       tlsSidecar: 30
         resources:
           requests:
             cpu: 200m
             memory: 64Mi
           limits:
             cpu: 500m
             memory: 128Mi
       topicOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 31
           type: inline
           loggers:
             rootLogger.level: "INFO"
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
       userOperator:
         watchedNamespace: my-topic-namespace
         reconciliationIntervalSeconds: 60
         logging: 32
           type: inline
           loggers:
             rootLogger.level: INFO
         resources:
           requests:
             memory: 512Mi
             cpu: "1"
           limits:
             memory: 512Mi
             cpu: "1"
     kafkaExporter: 33
       # ...
     cruiseControl: 34
       # ...
       tlsSidecar: 35
       # ...

   1

   レプリカノードの数。クラスターにトピックがすでに定義されている場合は、クラスターをスケーリング できます。

   2

   Kafka バージョン。アップグレード手順 に従うと、サポート対象のバージョンに変更できます。

   3

   指定された Kafka loggers and log levels が ConfigMap を介して直接的 (inline) または間接的 (external) に追加されます。カスタム ConfigMap は、log4j.properties キー下に配置する必要があります。Kafka kafka.root.logger.level ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   4

   サポートされているリソース (現在は cpu と memory)の予約と、消費可能な最大リソースを指定するための制限を要求します。

   5

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。

   6

   Kafka を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   7

   高度なオプション: コンテナーイメージの設定。これは特別な状況でのみ推奨されます。

   8

   リスナーは、ブートストラップアドレスでクライアントが Kafka クラスターに接続する方法を設定します。リスナーは、OpenShift クラスター内部または外部からの接続の 内部 または 外部 リスナーとして設定 されます。

   9

   リスナーを識別するための名前。Kafka クラスター内で一意である必要があります。

   10

   Kafka 内でリスナーによって使用されるポート番号。ポート番号は指定の Kafka クラスター内で一意である必要があります。許可されるポート番号は 9092 以上ですが、すでに Prometheus および JMX によって使用されているポート 9404 および 9999 以外になります。リスナーのタイプによっては、ポート番号は Kafka クライアントに接続するポート番号と同じではない場合があります。

   11

   internal として、または external リスナーに対して指定されるリスナータイプ (route、loadbalancer、nodeport、または ingress)。

   12

   各リスナーの TLS 暗号化を有効にします。デフォルトは false です。route リスナーに TLS 暗号化は必要ありません。

   13

   クラスターサービスサフィックス (通常は cluster.local) を含む完全修飾 DNS 名が割り当てられているかどうかを定義します。

   14

   相互 TLS、SCRAM-SHA-512、またはトークンベース OAuth 2.0 として指定 されるリスナー認証メカニズム。

   15

   外部リスナー設定は 、route、loadbalancer、または nodeportからなど、Kafka クラスターが外部の OpenShift に公開される方法を指定 します。

   16

   外部の認証局によって管理される Kafka リスナー証明書 の任意設定。brokerCertChainAndKey は、サーバー証明書および秘密鍵が含まれる Secret を指定します。TLS による暗号化が有効な任意のリスナーで Kafka リスナー証明書を設定できます。

   17

   承認は Kafka ブローカーで簡易、OAUTH2.0、または OPA 承認を有効化 します。簡易承認では、AclAuthorizer Kafka プラグインが使用されます。

   18

   config はブローカーの設定を指定します。AMQ Streams によって直接管理されないプロパティーに限り、標準 Apache Kafka 設定の提供が可能 です。

   19

   特定の 暗号スイート または TLS バージョンを有効化するために TLS 暗号化が有効になっている SSL プロパティー。

   20

   Storage は ephemeral、persistent-claim、jbod のいずれかに設定されています。

   21

   永続ボリュームのストレージサイズは拡張可能 で、さらに JBOD ストレージへのボリューム追加が可能 です。

   22

   永続ストレージには、動的ボリュームプロビジョニングのためのストレージ id や class など、追加の設定オプション があります。

   23

   ラックアウェアネス (Rack awareness) は、異なるラック全体でレプリカを分散するために設定されます。topologykey はクラスターノードのラベルと一致する必要があります。

   24

   Prometheus メトリクス は有効になっています。この例では、メトリクスは Prometheus JMX Exporter (デフォルトのメトリクスエクスポーター) に対して設定されます。

   25

   Prometheus JMX Exporter 経由でメトリクスを Grafana ダッシュボードにエクスポートする Prometheus ルール。Prometheus JMX Exporter の設定が含まれる ConfigMap を参照することで有効になります。metricsConfig.valueFrom.configMapKeyRef.key 配下に空のファイルが含まれる ConfigMap の参照を使用して、追加設定なしでメトリクスを有効にできます。

   26

   Kafka 設定と似たプロパティーが含まれる、ZooKeeper 固有の設定。

   27

   ZooKeeper ノードの数。通常、ZooKeeper クラスターまたはアンサンブルは、一般的に 3、5、7 個の奇数個のノードで実行されます。効果的なクォーラムを維持するには、過半数のノードが利用可能である必要があります。ZooKeeper クラスターでクォーラムを失うと、クライアントへの応答が停止し、Kafka ブローカーが機能しなくなります。AMQ Streams では、ZooKeeper クラスターの安定性および高可用性が重要になります。

   28

   指定された ZooKeeper ロガーおよびログレベル。

   29

   Topic Operator および User Operator の設定を指定 する Entity Operator 設定。

   30

   Entity Operator の TLS サイドカー設定。Entity Operator は、ZooKeeper とのセキュアな通信に TLS サイドカーを使用します。

   31

   指定された Topic Operator ロガーおよびログレベル。この例では、inline ロギングを使用します。

   32

   指定された User Operator ロガーおよびログレベル。

   33

   Kafka Exporter の設定。Kafka Exporter は、特にコンシューマーラグデータなどのメトリクスデータを Kafka ブローカーから抽出する任意のコンポーネントです。

   34

   Kafka クラスターのリバランス に使用される Cruise Control の任意設定。

   35

   Cruise Conrol の TLS サイドカーの設定。Cruise Control は、ZooKeeper とのセキュアな通信に TLS サイドカーを使用します。

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

   oc apply -f <kafka_configuration_file>

手順

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

   以下はその例です。

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

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

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

手順

1. 削除する Pod の名前を見つけます。

   Kafka ブローカー Pod の名前は <cluster-name>-kafka-<index> です。ここで、<index> はゼロで始まり、レプリカの合計数から 1 を引いた数で終了します。例: my-cluster-kafka-0

2. OpenShift で Pod リソースにアノテーションを付けます。

   oc annotate を使用します。

   oc annotate pod cluster-name-kafka-index strimzi.io/delete-pod-and-pvc=true

3. 基盤となる永続ボリューム要求 (Persistent Volume Claim) でアノテーションが付けられた Pod が削除され、再作成されるときに、次の調整の実行を待ちます。

手順

1. 削除する Pod の名前を見つけます。

   ZooKeeper Pod の名前は <cluster-name>-zookeeper-<index> です。ここで、<index> はゼロで始まり、レプリカの合計数から 1 を引いた数で終了します。例: my-cluster-zookeeper-0

2. OpenShift で Pod リソースにアノテーションを付けます。

   oc annotate を使用します。

   oc annotate pod cluster-name-zookeeper-index strimzi.io/delete-pod-and-pvc=true

3. 基盤となる永続ボリューム要求 (Persistent Volume Claim) でアノテーションが付けられた Pod が削除され、再作成されるときに、次の調整の実行を待ちます。

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect 1
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 2
   spec:
     replicas: 3 3
     authentication: 4
       type: tls
       certificateAndKey:
         certificate: source.crt
         key: source.key
         secretName: my-user-source
     bootstrapServers: my-cluster-kafka-bootstrap:9092 5
     tls: 6
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     config: 7
       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
     build: 8
       output: 9
         type: docker
         image: my-registry.io/my-org/my-connect-cluster:latest
         pushSecret: my-registry-credentials
       plugins: 10
         - 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
     externalConfiguration: 11
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     resources: 12
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 13
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 14
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metricsConfig: 15
       type: jmxPrometheusExporter
       valueFrom:
         configMapKeyRef:
           name: my-config-map
           key: my-key
     jvmOptions: 16
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 17
     rack:
       topologyKey: topology.kubernetes.io/zone 18
     template: 19
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 20
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   KafkaConnect を使用します。

   2

   Kafka Connect クラスターの KafkaConnectors を有効にします。

   3

   タスクを実行するワーカーの レプリカノード数。

   4

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-256/SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、Kafka Connect クラスターの認証。デフォルトでは、Kafka Connect はプレーンテキスト接続を使用して Kafka ブローカーに接続します。

   5

   Kafka Connect クラスターに接続するためのブートストラップサーバー。

   6

   クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   7

   ワーカーの Kafka Connect 設定 (コネクターではない)。標準の Apache Kafka 設定が提供されることがありますが、AMQ Streams によって直接管理されないプロパティーに限定されます。

   8

   コネクタープラグインで自動的にコンテナーイメージをビルドするためのビルド設定プロパティー。

   9

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

   10

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

   11

   ここで示す環境変数や、ボリュームを使用した Kafka コネクターの外部設定。設定プロバイダープラグイン を使用して、外部ソースから設定値を読み込む こともできます。

   12

   サポートされているリソース (現在は cpu と memory) の予約と、消費可能な最大リソースを指定するための制限を要求します。

   13

   指定された Kafka loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Connect log4j.rootLogger ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   14

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するための ヘルスチェック。

   15

   Prometheus メトリクス。この例では、Prometheus JMX エクスポーターの設定が含まれる ConfigMap を参照して有効になります。metricsConfig.valueFrom.configMapKeyRef.key 配下に空のファイルが含まれる ConfigMap の参照を使用して、追加設定なしでメトリクスを有効にできます。

   16

   Kafka Connect を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   17

   高度なオプション:コンテナーイメージの設定。これは特別な状況でのみ推奨されます。

   18

   ラックアウェアネス (Rack awareness) は、異なるラック全体でレプリカを分散するために設定されます。topologykey はクラスターノードのラベルと一致する必要があります。

   19

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   20

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

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

   oc apply -f KAFKA-CONNECT-CONFIG-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/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     # ...
     authorization:
       type: simple
       acls:
         # access to offset.storage.topic
         - resource:
             type: topic
             name: connect-cluster-offsets
             patternType: literal
           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 KAFKA-USER-CONFIG-FILE

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     replicas: 3 1
     consumer:
       bootstrapServers: my-source-cluster-kafka-bootstrap:9092 2
       groupId: "my-group" 3
       numStreams: 2 4
       offsetCommitInterval: 120000 5
       tls: 6
         trustedCertificates:
         - secretName: my-source-cluster-ca-cert
           certificate: ca.crt
       authentication: 7
         type: tls
         certificateAndKey:
           secretName: my-source-secret
           certificate: public.crt
           key: private.key
       config: 8
         max.poll.records: 100
         receive.buffer.bytes: 32768
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 9
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 10
     producer:
       bootstrapServers: my-target-cluster-kafka-bootstrap:9092
       abortOnSendFailure: false 11
       tls:
         trustedCertificates:
         - secretName: my-target-cluster-ca-cert
           certificate: ca.crt
       authentication:
         type: tls
         certificateAndKey:
           secretName: my-target-secret
           certificate: public.crt
           key: private.key
       config:
         compression.type: gzip
         batch.size: 8192
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 12
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 13
     include: "my-topic|other-topic" 14
     resources: 15
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 16
       type: inline
       loggers:
         mirrormaker.root.logger: "INFO"
     readinessProbe: 17
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metricsConfig: 18
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          name: my-config-map
          key: my-key
     jvmOptions: 19
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 20
     template: 21
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 22
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing: 23
       type: jaeger

   1

   レプリカノードの数。

   2

   コンシューマーおよびプロデューサーの ブートストラップサーバー。

   3

   コンシューマーのグループ ID。

   4

   コンシューマーストリームの数。

   5

   オフセットの自動コミット間隔 (ミリ秒単位)。

   6

   コンシューマーまたはプロデューサーの TLS 証明書が X.509 形式で保存される、キー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   7

   OAuth ベアラートークン または SASL ベースの SCRAM-SHA-256/SCRAM-SHA-512 または PLAIN メカニズムを使用するか、ここで紹介されているような TLS メカニズム を使用したコンシューマーまたはプロデューサーの認証。

   8

   コンシューマー および プロデューサー の Kafka 設定オプション。

   9

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   10

   HTTPS に設定することで、ホスト名の検証が有効 になります。空の文字列を指定すると検証が無効になります。

   11

   abortOnSendFailure プロパティー が true に設定されている場合、メッセージの送信に失敗した後、Kafka MirrorMaker は終了し、コンテナは再起動します。

   12

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   13

   HTTPS に設定することで、ホスト名の検証が有効 になります。空の文字列を指定すると検証が無効になります。

   14

   ソースからターゲット Kafka クラスターにミラーリングされた 含まれるトピック。

   15

   サポートされているリソース (現在は cpu と memory) の予約と、消費可能な最大リソースを指定するための制限を要求します。

   16

   指定された ロガーおよびログレベル が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。MirrorMaker には mirrormaker.root.logger と呼ばれる単一のロガーがあります。ログレベルは INFO、ERROR、WARN、TRACE、DEBUG、FATAL、または OFF に設定できます。

   17

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するための ヘルスチェック。

   18

   Prometheus メトリクス。この例では、Prometheus JMX エクスポーターの設定が含まれる ConfigMap を参照して有効になります。metricsConfig.valueFrom.configMapKeyRef.key 配下に空のファイルが含まれる ConfigMap の参照を使用して、追加設定なしでメトリクスを有効にできます。

   19

   Kafka MirrorMaker を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   20

   高度なオプション:コンテナーイメージの設定。これは特別な状況でのみ推奨されます。

   21

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   22

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

   23

   Jaeger の分散トレーシングは有効になっています。

   警告

   abortOnSendFailure プロパティーが false に設定されると、プロデューサーはトピックの次のメッセージを送信しようとします。失敗したメッセージは再送されないため、元のメッセージが失われる可能性があります。

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

   oc apply -f <your-file>

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker2
   spec:
     version: 3.1.0 1
     replicas: 3 2
     connectCluster: "my-cluster-target" 3
     clusters: 4
     - alias: "my-cluster-source" 5
       authentication: 6
         certificateAndKey:
           certificate: source.crt
           key: source.key
           secretName: my-user-source
         type: tls
       bootstrapServers: my-cluster-source-kafka-bootstrap:9092 7
       tls: 8
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-source-cluster-ca-cert
     - alias: "my-cluster-target" 9
       authentication: 10
         certificateAndKey:
           certificate: target.crt
           key: target.key
           secretName: my-user-target
         type: tls
       bootstrapServers: my-cluster-target-kafka-bootstrap:9092 11
       config: 12
         config.storage.replication.factor: 1
         offset.storage.replication.factor: 1
         status.storage.replication.factor: 1
         ssl.cipher.suites: "TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384" 13
         ssl.enabled.protocols: "TLSv1.2"
         ssl.protocol: "TLSv1.2"
         ssl.endpoint.identification.algorithm: HTTPS 14
       tls: 15
         trustedCertificates:
         - certificate: ca.crt
           secretName: my-cluster-target-cluster-ca-cert
     mirrors: 16
     - sourceCluster: "my-cluster-source" 17
       targetCluster: "my-cluster-target" 18
       sourceConnector: 19
         tasksMax: 10 20
         config:
           replication.factor: 1 21
           offset-syncs.topic.replication.factor: 1 22
           sync.topic.acls.enabled: "false" 23
           refresh.topics.interval.seconds: 60 24
           replication.policy.separator: "" 25
           replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy" 26
       heartbeatConnector: 27
         config:
           heartbeats.topic.replication.factor: 1 28
       checkpointConnector: 29
         config:
           checkpoints.topic.replication.factor: 1 30
           refresh.groups.interval.seconds: 600 31
           sync.group.offsets.enabled: true 32
           sync.group.offsets.interval.seconds: 60 33
           emit.checkpoints.interval.seconds: 60 34
           replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
       topicsPattern: ".*" 35
       groupsPattern: "group1|group2|group3" 36
     resources: 37
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 38
       type: inline
       loggers:
         connect.root.logger.level: "INFO"
     readinessProbe: 39
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     jvmOptions: 40
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 41
     template: 42
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 43
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger 44
     externalConfiguration: 45
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey

   1

   常に同じになる Kafka Connect と Mirror Maker 2.0 の バージョン。

   2

   タスクを実行するワーカーの レプリカノード数。

   3

   Kafka Connect の Kafka クラスターエイリアス。ターゲット Kafka クラスターを指定する必要があります。Kafka クラスターは、その内部トピックのために Kafka Connect によって使用されます。

   4

   同期される Kafka クラスターの 指定。

   5

   ソースの Kafka クラスターの クラスターエイリアス。

   6

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-256/SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、ソースクラスターの認証。

   7

   ソース Kafka クラスターに接続するための ブートストラップサーバー。

   8

   ソース Kafka クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   9

   ターゲット Kafka クラスターの クラスターエイリアス。

   10

   ターゲット Kafka クラスターの認証は、ソース Kafka クラスターと同様に設定されます。

   11

   ターゲット Kafka クラスターに接続するための ブートストラップサーバー。

   12

   Kafka Connect の設定。標準の Apache Kafka 設定が提供されることがありますが、AMQ Streams によって直接管理されないプロパティーに限定されます。

   13

   TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。

   14

   HTTPS に設定することで、ホスト名の検証が有効になります。空の文字列を指定すると検証が無効になります。

   15

   ターゲット Kafka クラスターの TLS による暗号化は、ソース Kafka クラスターと同様に設定されます。

   16

   MirrorMaker 2.0 コネクター。

   17

   MirrorMaker 2.0 コネクターによって使用されるソースクラスターの クラスターエイリアス。

   18

   MirrorMaker 2.0 コネクターによって使用されるターゲットクラスターの クラスターエイリアス。

   19

   リモートトピックを作成する MirrorSourceConnector の設定。デフォルトの設定オプションは config によって上書きされます。

   20

   コネクターによる作成が可能なタスクの最大数。タスクは、データのレプリケーションを処理し、並行して実行されます。インフラストラクチャーが処理のオーバーヘッドをサポートする場合、この値を大きくするとスループットが向上されます。Kafka Connect は、クラスターのメンバー間でタスクを分散します。ワーカーよりも多くのタスクがある場合は、ワーカーには複数のタスクが割り当てられます。シンクコネクターでは、消費される各トピックパーティションに 1 つタスクがあるようになることを目指します。ソースコネクターでは、並行して実行できるタスクの数は外部システムによって異なる場合もあります。並列処理を実現できない場合、コネクターは最大数より少ないタスクを作成します。

   21

   ターゲットクラスターで作成されるミラーリングされたトピックのレプリケーション係数。

   22

   ソースおよびターゲットクラスターのオフセットをマップする MirrorSourceConnector offset-syncs 内部トピックのレプリケーション係数。

   23

   ACL ルールの同期 が有効になっていると、同期されたトピックに ACL が適用されます。デフォルトは true です。この機能は User Operator と互換性がありません。User Operator を使用している場合は、このプロパティーを false に設定します。

   24

   新規トピックのチェック頻度を変更する任意設定。デフォルトでは 10 分毎にチェックされます。

   25

   リモートトピック名の変更に使用する区切り文字を定義します。

   26

   リモートトピック名の自動変更をオーバーライドするポリシーを追加します。その名前の前にソースクラスターの名前を追加する代わりに、トピックが元の名前を保持します。このオプションの設定は、active/passive バックアップおよびデータ移行に役立ちます。トピックオフセットの同期を設定するには、このプロパティーも checkpointConnector.config に設定する必要があります。

   27

   接続チェックを実行する MirrorHeartbeatConnector の設定。デフォルトの設定オプションは config によって上書きされます。

   28

   ターゲットクラスターで作成されたハートビートトピックのレプリケーション係数。

   29

   オフセットを追跡する MirrorCheckpointConnector の設定。デフォルトの設定オプションは config によって上書きされます。

   30

   ターゲットクラスターで作成されたチェックポイントトピックのレプリケーション係数。

   31

   新規コンシューマーグループのチェック頻度を変更する任意設定。デフォルトでは 10 分毎にチェックされます。

   32

   コンシューマーグループのオフセットを同期する任意設定。これは、active/passive 設定でのリカバリーに便利です。同期はデフォルトでは有効になっていません。

   33

   コンシューマーグループオフセットの同期が有効な場合は、同期の頻度を調整できます。

   34

   オフセット追跡のチェック頻度を調整します。オフセット同期の頻度を変更する場合、これらのチェックの頻度も調整する必要がある場合があります。

   35

   正規表現パターンとして定義された ソースクラスターからのトピックレプリケーション。ここで、すべてのトピックを要求します。

   36

   正規表現パターンとして定義された ソースクラスターからのコンシューマーグループレプリケーション。ここで、3 つのコンシューマーグループを名前で要求します。コンマ区切りリストを使用できます。

   37

   サポートされているリソース (現在は cpu と memory) の予約と、消費可能な最大リソースを指定するための制限を要求します。

   38

   指定された Kafka loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Connect log4j.rootLogger ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   39

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するための ヘルスチェック。

   40

   Kafka MirrorMaker を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   41

   高度なオプション:コンテナーイメージの設定。これは特別な状況でのみ推奨されます。

   42

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   43

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

   44

   Jaeger の分散トレーシングは有効になっています。

   45

   環境変数として Kafka MirrorMaker にマウントされた OpenShift Secret の 外部設定。設定プロバイダープラグイン を使用して、外部ソースから設定値を読み込む こともできます。

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

   oc apply -f MIRRORMAKER-CONFIGURATION-FILE

手順

1. 2 つの Kafka リソースを設定します。1 つはソース Kafka クラスターを保護するためのもので、もう 1 つはターゲット Kafka クラスターを保護するためのものです。

   認証用のリスナー設定を追加し、認可を有効にすることができます。

   この例の場合、内部リスナーは TLS 暗号化と認証を使用して Kafka クラスター用に設定されています。Kafka の simple 認証が有効になっています。

   TLS 認証を使用したソース Kafka クラスターの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.1.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.1"
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   TLS 認証を使用したターゲット Kafka クラスターの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.1.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.1"
       storage:
         type: jbod
         volumes:
           - id: 0
             type: persistent-claim
             size: 100Gi
             deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 別の namespace で Kafka リソースを作成または更新します。

   oc apply -f <kafka_configuration_file> -n <namespace>

   Cluster Operator はリスナーを作成し、クラスターおよびクライアント認証局 (CA) 証明書を設定して Kafka クラスター内で認証を有効にします。

   証明書は、シークレット <cluster_name>-cluster-ca-cert に作成されます。

3. 2 つの KafkaUser リソースを設定します。1 つはソース Kafka クラスターのユーザー用で、もう 1 つはターゲット Kafka クラスターのユーザー用です。

   1. 対応するソースおよびターゲットの Kafka クラスターと同じ認証および認可タイプを設定します。たとえば、ソース Kafka クラスターの Kafka 設定で tls 認証と simple 認可タイプを使用した場合は、KafkaUser 設定でも同じものを使用します。

   2. ソースおよびターゲットの Kafka クラスターでの操作を可能にするために MirrorMaker 2.0 が必要とする ACL を設定します。

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

   TLS クライアント認証のソースユーザーの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-source-user
     labels:
       strimzi.io/cluster: my-source-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # MirrorSourceConnector
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Create
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: DescribeConfigs
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: DescribeConfigs
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operation: Read

   TLS クライアント認証のターゲットユーザーの設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-target-user
     labels:
       strimzi.io/cluster: my-target-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # Underlying Kafka Connect internal topics to store configuration, offsets, or status
         - resource:
             type: group
             name: mirrormaker2-cluster
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operation: Create
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Describe
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: DescribeConfigs
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operation: Create
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Create
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Alter
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: AlterConfigs
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Create
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operation: Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Read
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operation: Describe
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operation: Read
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operation: Create
         - resource:
             type: topic
             name: heartbeats
           operation: Describe
         - resource:
             type: topic
             name: heartbeats
           operation: Write

   注記

   type を tls-external に設定することにより、User Operator の外部で発行された証明書を使用できます。詳細については、User authentication を参照してください。

4. ソースおよびターゲットの Kafka クラスター用に作成した各 namespace で、KafkaUser リソースを作成または更新します。

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   User Operator はクライアント (MirrorMaker) に対応するユーザーを作成すると共に、選択した認証タイプに基づいて、クライアント認証に使用されるセキュリティークレデンシャルを作成します。

   User Operator は、KafkaUser リソースと同じ名前の新しいシークレットを作成します。シークレットには、TLS クライアント認証の秘密鍵と公開鍵が含まれます。公開鍵は、クライアント認証局 (CA) によって署名されたユーザー証明書に含まれます。

5. ソースおよびターゲットの Kafka クラスターに接続するための認証の詳細を使用して KafkaMirrorMaker2 リソースを設定します。

   TLS 認証を使用した MirrorMaker 2.0 の設定例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.1.0
     replicas: 1
     connectCluster: "my-target-cluster"
     clusters:
       - alias: "my-source-cluster"
         bootstrapServers: my-source-cluster-kafka-bootstrap:9093
         tls: 1
           trustedCertificates:
             - secretName: my-source-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 2
           type: tls
           certificateAndKey:
             secretName: my-source-user
             certificate: user.crt
             key: user.key
       - alias: "my-target-cluster"
         bootstrapServers: my-target-cluster-kafka-bootstrap:9093
         tls: 3
           trustedCertificates:
             - secretName: my-target-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 4
           type: tls
           certificateAndKey:
             secretName: my-target-user
             certificate: user.crt
             key: user.key
         config:
           # -1 means it will use the default replication factor configured in the broker
           config.storage.replication.factor: -1
           offset.storage.replication.factor: -1
           status.storage.replication.factor: -1
     mirrors:
       - sourceCluster: "my-source-cluster"
         targetCluster: "my-target-cluster"
         sourceConnector:
           config:
             replication.factor: 1
             offset-syncs.topic.replication.factor: 1
             sync.topic.acls.enabled: "false"
         heartbeatConnector:
           config:
             heartbeats.topic.replication.factor: 1
         checkpointConnector:
           config:
             checkpoints.topic.replication.factor: 1
             sync.group.offsets.enabled: "true"
         topicsPattern: ".*"
         groupsPattern: ".*"

   1

   ソース Kafka クラスターの TLS 証明書。それらが別の namespace にある場合は、Kafka クラスターの namespace からクラスターシークレットをコピーします。

   2

   TLS mechanism を使用してソース Kafka クラスターにアクセスするためのユーザー認証。

   3

   ターゲット Kafka クラスターの TLS 証明書。

   4

   ターゲット Kafka クラスターにアクセスするためのユーザー認証。

6. ターゲット Kafka クラスターと同じ namespace で KafkaMirrorMaker2 リソースを作成または更新します。

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>

手順

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

   oc get KafkaMirrorMaker2

2. KafkaMirrorMaker2 カスタムリソースから再起動される Kafka MirrorMaker 2.0 コネクターの名前を見つけます。

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

3. コネクターを再起動するには、OpenShift で KafkaMirrorMaker2 リソースにアノテーションを付けます。この例では、oc annotate は my-source->my-target.MirrorSourceConnector という名前のコネクターを再起動します。

   oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector=my-source->my-target.MirrorSourceConnector"

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

   アノテーションが調整プロセスで検出されれば、Kafka MirrorMaker 2.0 コネクターは再起動されます。再起動要求が許可されると、アノテーションは KafkaMirrorMaker2 カスタムリソースから削除されます。

手順

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

   oc get KafkaMirrorMaker2

2. Kafka MirrorMaker 2.0 コネクターの名前と、KafkaMirrorMaker2 カスタムリソースから再起動されるタスクの ID を検索します。タスク ID は 0 から始まる負の値ではない整数です。

   oc describe KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME

3. コネクタータスクを再起動するには、OpenShift で KafkaMirrorMaker2 リソースにアノテーションを付けます。この例では、oc annotate は my-source->my-target.MirrorSourceConnector という名前のコネクタのタスク 0 を再起動します。

   oc annotate KafkaMirrorMaker2 KAFKAMIRRORMAKER-2-NAME "strimzi.io/restart-connector-task=my-source->my-target.MirrorSourceConnector:0"

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

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

手順

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

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

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     replicas: 3 1
     bootstrapServers: <cluster_name>-cluster-kafka-bootstrap:9092 2
     tls: 3
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     authentication: 4
       type: tls
       certificateAndKey:
         secretName: my-secret
         certificate: public.crt
         key: private.key
     http: 5
       port: 8080
       cors: 6
         allowedOrigins: "https://strimzi.io"
         allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
     consumer: 7
       config:
         auto.offset.reset: earliest
     producer: 8
       config:
         delivery.timeout.ms: 300000
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         logger.bridge.level: "INFO"
         # enabling DEBUG just for send operation
         logger.send.name: "http.openapi.operation.send"
         logger.send.level: "DEBUG"
     jvmOptions: 11
       "-Xmx": "1g"
       "-Xms": "1g"
     readinessProbe: 12
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     image: my-org/my-image:latest 13
     template: 14
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       bridgeContainer: 15
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   レプリカノードの数。

   2

   ターゲット Kafka クラスターに接続するための ブートストラップサーバー。Kafka クラスターの名前は <cluster_name> を使用します。

   3

   ソース Kafka クラスターの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。複数の証明書が同じシークレットに保存されている場合は、複数回リストできます。

   4

   OAuth ベアラートークン、SASL ベースの SCRAM-SHA-256/SCRAM-SHA-512 またはPLAIN メカニズムを使用し、ここで示された TLS メカニズム を使用する、Kafka Bridge クラスターの認証。デフォルトでは、Kafka Bridge は認証なしで Kafka ブローカーに接続します。

   5

   Kafka ブローカーへの HTTP アクセス。

   6

   選択されたリソースおよびアクセスメソッドを指定する CORS アクセス。要求に別の HTTP ヘッダーを追加して、Kafka クラスターへのアクセスが許可されるオリジンが記述されます。

   7

   コンシューマー設定 オプション。

   8

   プロデューサー設定 オプション。

   9

   サポートされているリソース (現在は cpu と memory) の予約と、消費可能な最大リソースを指定するための制限を要求します。

   10

   指定された Kafka Bridge loggers and log levels が ConfigMap を介して直接的に (inline) または間接的に (external) に追加されます。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。Kafka Bridge ロガーでは、ログレベルを INFO、ERROR、WARN、TRACE、DEBUG、FATAL または OFF に設定できます。

   11

   Kafka Bridge を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。

   12

   コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するための ヘルスチェック。

   13

   オプション:コンテナーイメージの設定。これは特別な状況でのみ推奨されます。

   14

   テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。

   15

   環境変数は、Jaeger を使用した分散トレーシングにも設定 されます。

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

   oc apply -f KAFKA-BRIDGE-CONFIG-FILE

手順

1. クラスターデプロイメントを指定するリソースの affinity プロパティーを編集します。ワーカーノードが Kafka ブローカーまたは ZooKeeper ノードで共有されないようにするには、strimzi.io/name ラベルを使用します。topologyKey を kubernetes.io/hostname に設定して、選択した Pod が同じホスト名のノードでスケジュールされないように指定します。これにより、同じワーカーノードを単一の Kafka ブローカーと単一の ZooKeeper ノードで共有できます。以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/name
                         operator: In
                         values:
                           - CLUSTER-NAME-kafka
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/name
                         operator: In
                         values:
                           - CLUSTER-NAME-zookeeper
                   topologyKey: "kubernetes.io/hostname"
       # ...

   CLUSTER-NAME は、Kafka カスタムリソースの名前です。

2. Kafka ブローカーと ZooKeeper ノードが同じワーカーノードを共有しないようにする場合は、strimzi.io/cluster ラベルを使用します。以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/cluster
                         operator: In
                         values:
                           - CLUSTER-NAME
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: strimzi.io/cluster
                         operator: In
                         values:
                           - CLUSTER-NAME
                   topologyKey: "kubernetes.io/hostname"
       # ...

   CLUSTER-NAME は、Kafka カスタムリソースの名前です。

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

   oc apply -f <kafka_configuration_file>

手順

1. クラスターデプロイメントを指定するリソースの affinity プロパティーを編集します。ラベルを使用して、同じノードでスケジュールすべきでない Pod を指定します。topologyKey を kubernetes.io/hostname に設定し、選択した Pod が同じホスト名のノードでスケジュールされてはならないことを指定する必要があります。以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             podAntiAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 - labelSelector:
                     matchExpressions:
                       - key: application
                         operator: In
                         values:
                           - postgresql
                           - mongodb
                   topologyKey: "kubernetes.io/hostname"
       # ...
     zookeeper:
       # ...

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

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

   oc apply -f <kafka_configuration_file>

手順

1. AMQ Streams コンポーネントをスケジュールする必要のあるノードにラベルを付けます。

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

   oc label node NAME-OF-NODE node-type=fast-network

   または、既存のラベルによっては再利用が可能です。

2. クラスターデプロイメントを指定するリソースの affinity プロパティーを編集します。以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           affinity:
             nodeAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 nodeSelectorTerms:
                   - matchExpressions:
                     - key: node-type
                       operator: In
                       values:
                       - fast-network
       # ...
     zookeeper:
       # ...

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

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

   oc apply -f <kafka_configuration_file>

手順

1. 専用ノードとして使用するノードを選択します。
2. これらのノードにスケジュールされているワークロードがないことを確認します。

3. 選択したノードにテイントを設定します。

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

   oc adm taint node NAME-OF-NODE dedicated=Kafka:NoSchedule

4. さらに、選択したノードにラベルも追加します。

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

   oc label node NAME-OF-NODE dedicated=Kafka

5. クラスターデプロイメントを指定するリソースの affinity および tolerations プロパティーを編集します。

   以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       template:
         pod:
           tolerations:
             - key: "dedicated"
               operator: "Equal"
               value: "Kafka"
               effect: "NoSchedule"
           affinity:
             nodeAffinity:
               requiredDuringSchedulingIgnoredDuringExecution:
                 nodeSelectorTerms:
                 - matchExpressions:
                   - key: dedicated
                     operator: In
                     values:
                     - Kafka
       # ...
     zookeeper:
       # ...

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

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

   oc apply -f <kafka_configuration_file>

手順

1. ConfigMap を作成します。

   ConfigMap を YAML ファイルとして作成するか、プロパティーファイルから Config Map を作成します。

   Kafka のルートロガー定義が含まれる ConfigMap の例:

   kind: ConfigMap
   apiVersion: v1
   metadata:
     name: logging-configmap
   data:
     log4j.properties:
       kafka.root.logger.level="INFO"

   プロパティーファイルを使用している場合は、コマンドラインでファイルを指定します。

   oc create configmap logging-configmap --from-file=log4j.properties

   プロパティーファイルではロギング設定が定義されます。

   # Define the logger
   kafka.root.logger.level="INFO"
   # ...

2. リソースの spec にexternal ロギングを定義し、logging.valueFrom.configMapKeyRef.name に ConfigMap の名前を、logging.valueFrom.configMapKeyRef.key にこの ConfigMap のキーを設定します。

   spec:
     # ...
     logging:
       type: external
       valueFrom:
         configMapKeyRef:
           name: logging-configmap
           key: log4j.properties

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

   oc apply -f <kafka_configuration_file>

手順

1. 050-ConfigMap-strimzi-cluster-operator.yaml ファイルを更新して、フィルタープロパティーを ConfigMap に追加します。

   この例では、フィルタープロパティーは my-kafka-cluster Kafka クラスターのログのみを返します。

   kind: ConfigMap
   apiVersion: v1
   metadata:
     name: strimzi-cluster-operator
   data:
     log4j2.properties:
       #...
       appender.console.filter.filter1.type=MarkerFilter
       appender.console.filter.filter1.onMatch=ACCEPT
       appender.console.filter.filter1.onMismatch=DENY
       appender.console.filter.filter1.marker=Kafka(my-namespace/my-kafka-cluster)

   または、ConfigMap を直接編集することもできます。

   oc edit configmap strimzi-cluster-operator

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

   oc create -f install/cluster-operator/050-ConfigMap-strimzi-cluster-operator.yaml

手順

1. ConfigMap を作成します。

   ConfigMap を YAML ファイルとして作成するか、プロパティーファイルから Config Map を作成します。

   この例では、フィルタープロパティーは my-topic トピックに対してのみログを返します。

   kind: ConfigMap
   apiVersion: v1
   metadata:
     name: logging-configmap
   data:
     log4j2.properties:
       rootLogger.level="INFO"
       appender.console.filter.filter1.type=MarkerFilter
       appender.console.filter.filter1.onMatch=ACCEPT
       appender.console.filter.filter1.onMismatch=DENY
       appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)

   プロパティーファイルを使用している場合は、コマンドラインでファイルを指定します。

   oc create configmap logging-configmap --from-file=log4j2.properties

   プロパティーファイルではロギング設定が定義されます。

   # Define the logger
   rootLogger.level="INFO"
   # Set the filters
   appender.console.filter.filter1.type=MarkerFilter
   appender.console.filter.filter1.onMatch=ACCEPT
   appender.console.filter.filter1.onMismatch=DENY
   appender.console.filter.filter1.marker=KafkaTopic(my-namespace/my-topic)
   # ...

2. リソースの spec にexternal ロギングを定義し、logging.valueFrom.configMapKeyRef.name に ConfigMap の名前を、logging.valueFrom.configMapKeyRef.key にこの ConfigMap のキーを設定します。

   Topic Operatorについては、Kafka リソースの topicOperator 設定でロギングを指定します。

   spec:
     # ...
     entityOperator:
       topicOperator:
         logging:
           type: external
           valueFrom:
             configMapKeyRef:
               name: logging-configmap
               key: log4j2.properties

3. Cluster Operator をデプロイして変更を適用します。

1. 相互 TLS 認証 (TLS による暗号化が有効なリスナーのみ)
2. SCRAM-SHA-512 認証
3. OAuth 2.0 のトークンベースの認証
4. カスタム認証

手順

1. Kafka リソースを設定します。

   1. 承認には authorization プロパティーを設定します。

   2. listeners プロパティーを設定し、認証でリスナーを作成します。

      以下はその例です。

      apiVersion: kafka.strimzi.io/v1beta2
      kind: Kafka
      spec:
        kafka:
          # ...
          authorization: 1
            type: simple
            superUsers: 2
              - CN=client_1
              - user_2
              - CN=client_3
          listeners:
            - name: tls
              port: 9093
              type: internal
              tls: true
              authentication:
                type: tls 3
          # ...
        zookeeper:
          # ...

      1

      Authorizationは、AclAuthorizer Kafka プラグインを使用して、Kafka ブローカーでのsimpleな承認を可能にします。

      2

      Kafka へのアクセスを制限されないユーザープリンシパルのリスト。CN は、TLS による認証が使用される場合のクライアント証明書のコモンネームです。

      3

      リスナーの認証メカニズムは各リスナーに対して設定でき、相互 TLS、SCRAM-SHA-512、またはトークンベース OAuth 2.0 として指定 できます。

      外部リスナーを設定している場合、設定は選択した接続のメカニズムによって異なります。

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

   oc apply -f <kafka_configuration_file>

   Kafka クラスターは、TLS 認証を使用する Kafka ブローカーリスナーと共に設定されます。

   Kafka ブローカー Pod ごとにサービスが作成されます。

   サービスが作成され、Kafka クラスターに接続するための ブートストラップアドレス として機能します。

   kafka ブローカーのアイデンティティーを検証するクラスター CA 証明書もシークレット <cluster_name>-cluster-ca-cert に作成されます。

手順

1. KafkaUser リソースを設定します。

   以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-user
     labels:
       strimzi.io/cluster: my-cluster
   spec:
     authentication: 1
       type: tls
     authorization:
       type: simple 2
       acls:
         - resource:
             type: topic
             name: my-topic
             patternType: literal
           operation: Read
         - resource:
             type: topic
             name: my-topic
             patternType: literal
           operation: Describe
         - resource:
             type: group
             name: my-group
             patternType: literal
           operation: Read

   1

   相互 tls または scram-sha-512 として定義されたユーザー認証メカニズム。

   2

   ACL ルールのリストが必要な簡易承認。

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

   oc apply -f USER-CONFIG-FILE

   KafkaUser リソースと同じ名前の Secret と共に、ユーザーが作成されます。Secret には、TLS クライアント認証の秘密鍵と公開鍵が含まれます。

手順

1. Kafka リソースを開きます。

2. networkPolicyPeers プロパティーで、Kafka クラスターへのアクセスが許可されるアプリケーション Pod または namespace を定義します。

   以下は、ラベル app が kafka-client に設定されているアプリケーションからの接続のみを許可するよう tls リスナーを設定する例になります。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
           networkPolicyPeers:
             - podSelector:
                 matchLabels:
                   app: kafka-client
       # ...
     zookeeper:
       # ...

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

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

   oc apply -f your-file

1. 承認サーバーをデプロイし、AMQ Streams と統合するためにそのデプロイメントを設定します。
2. OAuth 2.0 を使用するよう設定された Kafka ブローカーリスナーで Kafka クラスターをデプロイまたは更新します。
3. OAuth 2.0 を使用するように Java ベースの Kafka クライアントを更新します。
4. OAuth 2.0 を使用するように Kafka コンポーネントクライアントを更新します。

手順

1. Red Hat Single Sign-On の Admin Console にアクセスするか、Red Hat Single Sign-On の Admin CLI を使用して、OAuth 2.0 認証の設定時に作成した Kafka ブローカークライアントの Authorization Services を有効にします。
2. 承認サービスを使用して、クライアントのリソース、承認スコープ、ポリシー、およびパーミッションを定義します。
3. ロールとグループをユーザーとクライアントに割り当てて、パーミッションをユーザーとクライアントにバインドします。

4. エディターで Kafka リソースの Kafka ブローカー設定 (Kafka.spec.kafka) を更新して、Kafka ブローカーで Red Hat Single Sign-On による承認が使用されるように設定します。

   oc edit kafka my-cluster

5. Kafka ブローカーの kafka 設定を指定して、keycloak による承認を使用し、承認サーバーと Red Hat Single Sign-On の Authorization Services にアクセスできるようにします。

   以下はその例です。

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       authorization:
         type: keycloak 1
         tokenEndpointUri: <https://<auth-server-address>/auth/realms/external/protocol/openid-connect/token> 2
         clientId: kafka 3
         delegateToKafkaAcls: false 4
         disableTlsHostnameVerification: false 5
         superUsers: 6
         - CN=fred
         - sam
         - CN=edward
         tlsTrustedCertificates: 7
         - secretName: oauth-server-cert
           certificate: ca.crt
         grantsRefreshPeriodSeconds: 60 8
         grantsRefreshPoolSize: 5 9
         connectTimeoutSeconds: 60 10
         readTimeoutSeconds: 60 11
       #...

   1

   タイプ keycloak によって Red Hat Single Sign-On の承認が有効になります。

   2

   Red Hat Single Sign-On トークンエンドポイントの URI。実稼働環境の場合は、常に https:// urls を使用してください。トークンベースの oauth 認証を設定する場合、jwksEndpointUri をローカル JWT 検証の URI として指定します。tokenEndpointUri URI のホスト名は同じである必要があります。

   3

   承認サービスが有効になっている Red Hat Single Sign-On の OAuth 2.0 クライアント定義のクライアント ID。通常、kafka が ID として使用されます。

   4

   (オプション) Red Hat Single Sign-On Authorization Services ポリシーでアクセスが拒否された場合、Kafka AclAuthorizer に権限を委譲します。デフォルトは false です。

   5

   (任意設定): TLS ホスト名の検証を無効にします。デフォルトは false です。

   6

   (任意設定): 指定の スーパーユーザー。

   7

   (任意設定): 承認サーバーへの TLS 接続用の信用できる証明書。

   8

   (任意設定): 連続する付与 (Grants) 更新実行の間隔。これは、アクティブなセッションが Red Hat Single Sign-On でユーザーのパーミッション変更を検出する最大時間です。デフォルト値は 60 です。

   9

   (任意設定): アクティブなセッションの付与 (Grants) の更新 (並行して) に使用するスレッドの数。デフォルト値は 5 です。

   10

   （オプション）: Red Hat Single Sign-On トークンエンドポイントへの接続時のタイムアウト（秒単位）。デフォルト値は 60 です。

   11

   （オプション）: Red Hat Single Sign-On トークンエンドポイントへの接続時の読み取りタイムアウト（秒単位）。デフォルト値は 60 です。

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

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

   oc logs -f ${POD_NAME} -c kafka
   oc get pod -w

   ローリングアップデートによって、ブローカーが OAuth 2.0 承認を使用するように設定されます。

8. クライアントまたは特定のロールを持つユーザーとして Kafka ブローカーにアクセスして、設定したパーミッションを検証し、必要なアクセス権限があり、付与されるべきでないアクセス権限がないことを確認します。