3.4. Kafka MirrorMaker の設定

本章では、Kafka クラスター間でデータを複製するために AMQ Streams クラスターで Kafka MirrorMaker デプロイメントを設定する方法を説明します。

AMQ Streams では、MirrorMaker または MirrorMaker 2.0 を使用できます。MirrorMaker 2.0 は最新バージョンで、Kafka クラスター間でより効率的にデータをミラーリングする方法を提供します。

MirrorMaker を使用している場合は、KafkaMirrorMaker リソースを設定します。

以下の手順は、リソースの設定方法を示しています。

Kafka MirrorMaker の設定

サポート対象のプロパティーも以下で詳細に説明されています。

Kafka MirrorMaker 設定プロパティー

KafkaMirrorMaker リソースの完全なスキーマは、KafkaMirrorMaker schema のスキーマ参照に記載されています。

注記

KafkaMirrorMaker リソースに適用されるラベルは、Kafka MirrorMaker を設定する OpenShift リソースにも適用されます。そのため、必要に応じてリソースにラベルが適用されるため便利です。

3.4.1. Kafka MirrorMaker の設定

KafkaMirrorMaker リソースのプロパティーを使用して、Kafka MirrorMaker デプロイメントを設定します。

TLS または SASL 認証を使用して、プロデューサーおよびコンシューマーのアクセス制御を設定できます。この手順では、コンシューマーおよびプロデューサー側で TLS による暗号化および認証を使用する設定を説明します。

前提条件

AMQ Streams および Kafka がデプロイされている必要があります。
ソースおよびターゲットの Kafka クラスターが使用できる必要があります。

手順

KafkaMirrorMaker リソースの spec プロパティーを編集します。
設定可能なプロパティーは以下の例のとおりです。
```
apiVersion: kafka.strimzi.io/v1beta1
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"
  producer:
    bootstrapServers: my-target-cluster-kafka-bootstrap:9092
    abortOnSendFailure: false 10
    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" 11
      ssl.enabled.protocols: "TLSv1.2"
      ssl.protocol: "TLSv1.2"
  whitelist: "my-topic|other-topic" 12
  resources: 13
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 14
    type: inline
    loggers:
      mirrormaker.root.logger: "INFO"
  readinessProbe: 15
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metrics: 16
    lowercaseOutputName: true
    rules:
      - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*><>Count"
        name: "kafka_server_$1_$2_total"
      - pattern: "kafka.server<type=(.+), name=(.+)PerSec\\w*,
        topic=(.+)><>Count"
        name: "kafka_server_$1_$2_total"
        labels:
          topic: "$3"
  jvmOptions: 17
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest 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"
  tracing: 21
    type: jaeger
```
1
レプリカノードの数。
2
コンシューマーおよびプロデューサーのブートストラップサーバー。
3
コンシューマーのグループ ID。
4
コンシューマーストリームの数。
5
オフセットの自動コミット間隔 (ミリ秒単位)。
6
コンシューマーまたはプロデューサーの TLS 証明書が X.509 形式で保存されるキー名のある TLS による暗号化。詳細は、KafkaMirrorMakerTls のスキーマ参照を参照してください。
7
OAuth ベアラートークン、SASL ベースの SCRAM-SHA-512 または PLAIN メカニズムを使用し、ここで示された TLS メカニズムを使用する、コンシューマーおよびプロデューサーの認証。
8
コンシューマーおよびプロデューサーの Kafka 設定オプション。
9
TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。
10
true に設定された場合、Kafka MirrorMaker が終了し、メッセージの送信失敗後にコンテナーが再起動します。
11
TLS バージョンの特定の 暗号スイート と実行される外部リスナーの SSL プロパティー。
12
ソースからターゲット Kafka クラスターにミラーリングされたトピック。
13
現在 cpu および memory である、サポートされるリソースの予約を要求し、消費可能な最大リソースを指定を制限します。
14
ConfigMap より直接的 (inline) または間接的 (external) に追加されたロガーおよびログレベルを指定します。カスタム ConfigMap は、log4j.properties または log4j2.properties キー下に配置する必要があります。MirrorMaker には mirrormaker.root.logger と呼ばれる単一のロガーがあります。ログレベルは INFO、ERROR、WARN、TRACE、DEBUG、FATAL、または OFF に設定できます。
15
コンテナーを再起動するタイミング (liveness) およびコンテナーがトラフィックを許可できるタイミング (readiness) を把握するためのヘルスチェック。
16
Prometheus メトリクス。この例では、Prometheus JMX エクスポーターの設定で有効になっています。metrics: {} を使用すると追加設定なしでメトリクスを有効にすることができます。
17
Kafka MirrorMaker を実行している仮想マシン (VM) のパフォーマンスを最適化するための JVM 設定オプション。
18
高度な任意手順: 特別な場合のみ推奨されるコンテナーイメージの設定。
19
テンプレートのカスタマイズ。ここでは、Pod は非アフィニティーでスケジュールされるため、Pod は同じホスト名のノードではスケジュールされません。
20
環境変数は、Jaeger を使用した分散トレーシングにも設定されます。
21
Jaeger の分散トレーシングは有効になっています。
警告
abortOnSendFailure プロパティーが false に設定されると、プロデューサーはトピックの次のメッセージを送信しようとします。失敗したメッセージは再送されないため、元のメッセージが失われる可能性があります。
リソースを作成または更新します。
```
oc apply -f <your-file>
```

3.4. Kafka MirrorMaker の設定

3.4.1. Kafka MirrorMaker の設定

詳細情報

試用、購入および販売

コミュニティー

Red Hat ドキュメントについて

多様性を受け入れるオープンソースの強化

会社概要

Red Hat legal and privacy links

Red Hat legal and privacy links