Red Hat Summit Red Hat Summitサポートコンソール開発者トライアルの開始お問い合わせ

2.4.8. Kafka MirrorMaker 2.0 デプロイメントの保護

この手順は、MirrorMaker2.0 のデプロイメントを保護するために必要な設定の概要を説明しています。

ソース Kafka クラスターとターゲット Kafka クラスターには別々の設定が必要です。また、MirrorMaker がソースおよびターゲットの Kafka クラスターに接続するために必要な認証情報を提供するために、個別のユーザー設定が必要です。

Kafka クラスターの場合、OpenShift クラスター内のセキュア接続用の内部リスナーと、OpenShift クラスター外の接続用の外部リスナーを指定します。

認証および許可メカニズムを設定できます。ソースおよびターゲットの Kafka クラスターに実装されているセキュリティーオプションは、MirrorMaker 2.0 に実装されているセキュリティーオプションと互換性がある必要があります。

クラスターとユーザー認証情報を作成したら、セキュアな接続のために MirrorMaker 設定でそれらを指定します。

注記

この手順では、Cluster Operator によって生成された証明書が使用されますが、独自の証明書をインストール してそれらを置き換えることができます。外部認証局によって管理される Kafka リスナー証明書を使用するようにリスナーを設定 することもできます。

作業を開始する前の注意事項

この手順を開始する前に、AMQ Streams が提供する 設定ファイルの例 を確認してください。TLS または SCRAM-SHA-512 認証を使用して MirrorMaker 2.0 のデプロイメントを保護する場合の例が含まれています。例では、OpenShift クラスター内で接続するための内部リスナーを指定しています。

例には、ソースおよびターゲットの Kafka クラスターでの操作を可能にするために MirrorMaker 2.0 が必要とする ACL も示されています。

前提条件

  • AMQ Streams が実行されている
  • ソースクラスターとターゲットクラスターの namespace が分離されている

この手順では、ソースとターゲットの Kafka クラスターが別々の namespace にインストールされていることを前提としています。Topic Operator を使用する場合は、これを行う必要があります。Topoic Operator は、指定された namespace 内の単一クラスターのみをモニタリングします。

クラスターを namespace に分割することにより、クラスターシークレットをコピーして、namespace の外部からアクセスできるようにする必要があります。MirrorMaker 設定でシークレットを参照する必要があります。

手順

  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: {}
    Copy to Clipboard Toggle word wrap

    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: {}
    Copy to Clipboard Toggle word wrap

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

    oc apply -f <kafka_configuration_file> -n <namespace>
    Copy to Clipboard Toggle word wrap

    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
    Copy to Clipboard Toggle word wrap

    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
    Copy to Clipboard Toggle word wrap

    注記

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

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

    oc apply -f <kafka_user_configuration_file> -n <namespace>
    Copy to Clipboard Toggle word wrap

    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: ".*"
    Copy to Clipboard Toggle word wrap

    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>
    Copy to Clipboard Toggle word wrap
トップに戻る
Red Hat logoGithubredditYoutubeTwitter

詳細情報

試用、購入および販売

コミュニティー

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

Red Hat をお使いのお客様が、信頼できるコンテンツが含まれている製品やサービスを活用することで、イノベーションを行い、目標を達成できるようにします。 最新の更新を見る.

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

Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。このような変更は、段階的に実施される予定です。詳細情報: Red Hat ブログ.

会社概要

Red Hat は、企業がコアとなるデータセンターからネットワークエッジに至るまで、各種プラットフォームや環境全体で作業を簡素化できるように、強化されたソリューションを提供しています。

Theme

© 2025 Red Hat