RHEL での AMQ Streams の使用

手順

1. RHEL を FIPS モードでインストールします。

   詳細については、RHEL ドキュメント のセキュリティー強化に関する情報を参照してください。

2. AMQ Streams のインストールに進みます。

3. FIPS 準拠のアルゴリズムとプロトコルを使用するように AMQ Streams を設定します。

   使用する場合は、次の設定が準拠していることを確認してください。

   • SSL 暗号スイートと TLS バージョンは、JDK フレームワークでサポートされている必要があります。
   • SCRAM-SHA-512 パスワードの長さは少なくとも 32 文字である必要があります。

1. 新しい kafka ユーザーとグループを追加します。

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

2. amq-streams-<version>-bin.zip ファイルの内容を抽出して /opt/kafka ディレクトリーに移動します。

   unzip amq-streams-<version>-bin.zip -d /opt
   mv /opt/kafka*redhat* /opt/kafka

   この手順には管理者権限が必要です。

3. /opt/kafka ディレクトリーの所有権を kafka ユーザーに変更します。

   chown -R kafka:kafka /opt/kafka

4. ZooKeeper データを格納する /var/lib/zookeeper ディレクトリーを作成し、その所有権を kafka ユーザーに設定します。

   mkdir /var/lib/zookeeper
   chown -R kafka:kafka /var/lib/zookeeper

5. Kafka データを格納する /var/lib/kafka ディレクトリーを作成し、その所有権を kafka ユーザーに設定します。

   mkdir /var/lib/kafka
   chown -R kafka:kafka /var/lib/kafka

   Kafka のデフォルトの設定を単一ノードクラスターとして実行 できるようになりました。

   インストールを使用して、Kafka Connect などの他の Kafka コンポーネントを同じホストで実行することもできます。

   他のコンポーネントを実行するには、コンポーネント設定の bootstrap.servers プロパティーを使用して、Kafka ブローカーに接続するためのホスト名とポートを指定します。

   同じホスト上の単一の Kafka ブローカーを指すブートストラップサーバー設定の例

   bootstrap.servers=localhost:9092

   ただし、別のホストに Kafka コンポーネントをインストールして実行することを推奨します。

6. (オプション) Kafka コンポーネントを個別のホストにインストールします。

   1. インストールファイルを各ホストの /opt/kafka ディレクトリーに抽出します。
   2. /opt/kafka ディレクトリーの所有権を kafka ユーザーに変更します。

   3. Kafka ブローカーを実行しているホスト (またはマルチノード環境のホスト) にコンポーネントを接続する bootstrap.servers 設定を追加します。

      異なるホスト上の Kafka ブローカーを指すブートストラップサーバー設定の例

      bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092

      この設定は、Kafka Connect、MirrorMaker 2、および Kafka Bridge に使用できます。

クラスターの実行

1. ZooKeeper 設定ファイル /opt/kafka/config/zookeeper.properties を編集します。dataDir オプションを /var/lib/zookeeper/ に設定します。

   dataDir=/var/lib/zookeeper/

2. Kafka 設定ファイル /opt/kafka/config/server.properties を編集します。log.dirs オプションを /var/lib/kafka/ に設定します。

   log.dirs=/var/lib/kafka/

3. kafka ユーザーに切り替えます。

   su - kafka

4. ZooKeeper を起動します。

   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

5. ZooKeeper が実行されていることを確認します。

   jcmd | grep zookeeper

   戻り値:

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

6. Kafka を起動します。

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

7. Kafka が稼働していることを確認します。

   jcmd | grep kafka

   戻り値:

   number kafka.Kafka /opt/kafka/config/server.properties

手順

1. Kafka コンソールプロデューサーを起動し、メッセージを新しいトピックに送信するように設定します。

   /opt/kafka/bin/kafka-console-producer.sh --broker-list <bootstrap_address> --topic <topic-name>

   以下に例を示します。

   /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

2. コンソールに複数のメッセージを入力します。Enter を押して、各メッセージを新しいトピックに送信します。

   >message 1
   >message 2
   >message 3
   >message 4

   Kafka が新しいトピックを自動的に作成すると、トピックが存在しないという警告が表示される可能性があります。

   WARN Error while fetching metadata with correlation id 39 :
   {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

   この警告は、さらにメッセージを送信すると、再び表示されることはありません。

3. 新しいターミナルウィンドウで、Kafka コンソールコンシューマーを起動し、新しいトピックの最初からメッセージを読み取るように設定します。

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server <bootstrap_address> --topic <topic-name> --from-beginning

   以下に例を示します。

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

   受信メッセージがコンシューマーコンソールに表示されます。

4. プロデューサーコンソールに切り替え、追加のメッセージを送信します。それらがコンシューマーコンソールに表示されていることを確認します。
5. Ctrl+C を押して Kafka コンソールプロデューサーを停止し、コンシューマーを停止します。

手順

1. Kafka ブローカーを停止します。

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

2. Kafka ブローカーが停止していることを確認します。

   jcmd | grep kafka

3. ZooKeeper を停止します。

   su - kafka
   /opt/kafka/bin/zookeeper-server-stop.sh

クラスターの実行

1. /var/lib/zookeeper/ に myid ファイルを作成します。最初の ZooKeeper ノードに ID 1 を、2 番目の ZooKeeper ノードに 2 を、それぞれ入力します。

   su - kafka
   echo "<NodeID>" > /var/lib/zookeeper/myid

   以下に例を示します。

   su - kafka
   echo "1" > /var/lib/zookeeper/myid

2. 以下に対して、ZooKeeper 設定ファイル /opt/kafka/config/zookeeper.properties を編集します。

   • dataDir オプションを /var/lib/zookeeper/ に設定します。
   • initLimit および syncLimit オプションを設定します。
   • reconfigEnabled および standaloneEnabled オプションを設定します。

   • すべての ZooKeeper ノードのリストを追加します。このリストには、現在のノードも含まれている必要があります。

     5 つのメンバーを持つ ZooKeeper クラスターのノードの設定例

     tickTime=2000
     dataDir=/var/lib/zookeeper/
     initLimit=5
     syncLimit=2
     reconfigEnabled=true
     standaloneEnabled=false
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     
     server.1=172.17.0.1:2888:3888:participant;172.17.0.1:2181
     server.2=172.17.0.2:2888:3888:participant;172.17.0.2:2181
     server.3=172.17.0.3:2888:3888:participant;172.17.0.3:2181
     server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181
     server.5=172.17.0.5:2888:3888:participant;172.17.0.5:2181

3. デフォルトの設定ファイルで ZooKeeper を起動します。

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

4. ZooKeeper が稼働していることを確認します。

   jcmd | grep zookeeper

   戻り値:

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

5. クラスターのすべてのノードでこの手順を繰り返します。

6. ncat ユーティリティーを使用して、stat コマンドを各ノードに送信し、すべてのノードがクラスターのメンバーであることを確認します。

   ncat stat を使用してノードのステータスを確認します。

   echo stat | ncat localhost 2181

   stat のような 4 文字のコマンドを使用するには、zookeeper.properties で 4lw.commands.whitelist=* を指定する必要があります。

   この出力で、ノードが leader または follower のいずれかであることがわかります。

   ncat コマンドの出力例

   ZooKeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMT
   Clients:
    /0:0:0:0:0:0:0:1:59726[0](queued=0,recved=1,sent=0)
   
   Latency min/avg/max: 0/0/0
   Received: 2
   Sent: 1
   Connections: 1
   Outstanding: 0
   Zxid: 0x200000000
   Mode: follower
   Node count: 4

1. 以下のように、Kafka 設定ファイル /opt/kafka/config/server.properties を編集します。

   • 最初のブローカーの broker.id フィールドを 0 に、2 番目のブローカーを 1 に、それぞれ設定します。
   • zookeeper.connect オプションで ZooKeeper への接続の詳細を設定します。
   • Kafka リスナーを設定します。

   • logs.dir のディレクトリーに、コミットログが保存されるディレクトリーを設定します。

     以下は、Kafka ブローカーの設定例です。

     broker.id=0
     zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
     listeners=REPLICATION://:9091,PLAINTEXT://:9092
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     inter.broker.listener.name=REPLICATION
     log.dirs=/var/lib/kafka

     各 Kafka ブローカーが同じハードウェアで実行されている通常のインストールでは、broker.id 設定プロパティーのみがブローカー設定ごとに異なります。

2. デフォルトの設定ファイルで Kafka ブローカーを起動します。

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

3. Kafka ブローカーが稼働していることを確認します。

   jcmd | grep Kafka

   戻り値:

   number kafka.Kafka /opt/kafka/config/server.properties

4. ncat ユーティリティーを使用して dump コマンドを ZooKeeper ノードのいずれかに送信し、すべてのノードが Kafka クラスターのメンバーであることを確認します。

   ncat dump を使用して、ZooKeeper に登録されているすべての Kafka ブローカーを確認します。

   echo dump | ncat zoo1.my-domain.com 2181

   dump のような 4 文字のコマンドを使用するには、zookeeper.properties で 4lw.commands.whitelist=* を指定する必要があります。

   出力には、設定および起動したすべての Kafka ブローカーが含まれている必要があります。

   3 つのノードで設定される Kafka クラスターの ncat コマンドの出力例

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

1. Kafka ブローカーを停止します。

   /opt/kafka/bin/kafka-server-stop.sh

2. 完了後に再起動を必要とするブローカー設定に変更を加えます。

   詳細は、以下を参照してください。

   • Kafka の設定
   • Kafka ブローカーおよび ZooKeeper のアップグレード

3. Kafka ブローカーを再起動します。

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

4. Kafka が稼働していることを確認します。

   jcmd | grep kafka

   戻り値:

   number kafka.Kafka /opt/kafka/config/server.properties

5. ncat ユーティリティーを使用して dump コマンドを ZooKeeper ノードのいずれかに送信し、すべてのノードが Kafka クラスターのメンバーであることを確認します。

   ncat dump を使用して、ZooKeeper に登録されているすべての Kafka ブローカーを確認します。

   echo dump | ncat zoo1.my-domain.com 2181

   dump のような 4 文字のコマンドを使用するには、zookeeper.properties で 4lw.commands.whitelist=* を指定する必要があります。

   出力には、起動した Kafka ブローカーが含まれている必要があります。

   3 つのノードで設定される Kafka クラスターの ncat コマンドの出力例

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

6. ブローカで、レプリケーションが不十分なパーティションがゼロになるまで待ちます。コマンドラインから確認するか、メトリクスを使用できます。

   • --under-replicated-partitions パラメーターを指定して kafka-topics.sh コマンドを使用します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <bootstrap_address>  --describe --under-replicated-partitions

     以下に例を示します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions

     このコマンドは、クラスターでレプリケーションが不十分なパーティションのトピックのリストを表示します。

     レプリケーションが不十分なパーティションのトピック

     Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2
     Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1
     Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3
     # …

     In-Sync レプリカ (ISR) の数がレプリカの数より少ない場合、レプリケーションが不十分なパーティションが一覧表示されます。リストが返されない場合は、レプリケーションが不十分なパーティションはありません。

   • UnderReplicatedPartitions メトリックを使用します。

     kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

     このメトリックからは、レプリカが不十分なパーティションの数がわかります。数がゼロになるまで待機します。

     ヒント

     トピックにレプリケーションが不十分なパーティションがあると、Kafka Exporter を使用してアラートを作成します。

手順

1. kafka-storage ツールを使用して Kafka クラスターの ID を生成します。

   /opt/kafka/bin/kafka-storage.sh random-uuid

   このコマンドは ID を返します。KRaft モードでクラスター ID が必要です。

2. クラスターの各ブローカーに設定プロパティーファイルを作成します。

   ファイルは、Kafka で提供される例に基づくことができます。

   1. ロールを broker, `controller または broker, controller として指定します。

      たとえば、process.roles=broker, controller です。

   2. クラスター内の各ノードに 0 から始まる一意の node.id を指定します。

      たとえば、node.id=1 です。

   3. <node_id>@<hostname:port> の形式で controller.quorum.voters のリストを指定します。

      たとえば、controller.quorum.voters=1@localhost:9093 です。

3. Kafka クラスターの各ノードにログディレクトリーを設定します。

   /opt/kafka/bin/kafka-storage.sh format -t <uuid> -c /opt/kafka/config/kraft/server.properties

   戻り値:

   Formatting /tmp/kraft-combined-logs

   <uuid> を生成したクラスター ID に置き換えます。クラスター内の各ノードで、同じ ID を使用します。

   ブローカー用に作成したプロパティーファイルを使用してブローカー設定を適用します。

   server.properties 設定ファイルに指定されたデフォルトのログディレクトリーの場所は /tmp/kraft-combined-logs です。複数のログディレクトリーを設定するには、コンマ区切りリストを追加できます。

4. 各 Kafka ブローカーを起動します。

   /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties

5. Kafka が稼働していることを確認します。

   jcmd | grep kafka

   戻り値:

   number kafka.Kafka /opt/kafka/config/kraft/server.properties

手順

1. 環境変数設定プロバイダー JAR ファイルを Kafka libs ディレクトリーに追加します。

2. Kafka コンポーネントの設定プロパティーファイルで環境変数設定プロバイダーを初期化します。たとえば、Kafka のプロバイダーを初期化するには、設定を server.properties ファイルに追加します。

   環境変数の設定プロバイダーを有効にする設定

   config.providers=env
   config.providers.env.class=io.strimzi.kafka.EnvVarConfigProvider

3. プロパティーファイルに設定を追加して、環境変数からデータをロードします。

   環境変数からデータをロードするための設定

   option=${env:<MY_ENV_VAR_NAME>}

   MY_ENV_VAR_NAME などの大文字または大文字の環境変数の命名規則を使用します。

4. 変更を保存します。

5. Kafka コンポーネントを再起動します。

   マルチノードクラスターでブローカーを再起動する方法は、「Kafka ブローカーの正常なローリング再起動の実行」 を参照してください。

トピックの作成

1. kafka-topics.sh ユーティリティーを使用し以下の項目を指定して、トピックを作成します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --create オプション: 作成される新しいトピック。
   • --topic オプション: トピック名。
   • --partitions オプション: パーティション数。

   • --replication-factor オプション: トピックレプリケーション係数。

     また、--config オプションを使用して、デフォルトのトピック設定オプションの一部を上書きすることもできます。このオプションは複数回使用して、異なるオプションを上書きできます。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --create --topic <TopicName> --partitions <NumberOfPartitions> --replication-factor <ReplicationFactor> --config <Option1>=<Value1> --config <Option2>=<Value2>

     mytopic というトピックを作成するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic mytopic --partitions 50 --replication-factor 3 --config cleanup.policy=compact --config min.insync.replicas=2

2. kafka-topics.sh を使用して、トピックが存在することを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

   mytopic というトピックを記述するコマンドの例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

トピックの記述

1. kafka-topics.sh ユーティリティーを使用し以下の項目を指定して、トピックを説明します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --describe オプション: トピックを記述することを指定するために使用します。
   • --topic オプション: このオプションでトピック名を指定する必要があります。

   • --topic オプションを省略すると、利用可能なすべてのトピックを記述します。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

     mytopic というトピックを記述するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

     このコマンドは、このトピックに属するすべてのパーティションおよびレプリカをリスト表示します。また、すべてのトピック設定オプションも表示されます。

トピック設定の変更

1. kafka-configs.sh ツールを使用して、現在の設定を取得します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。

   • --describe オプション: 現在の設定を取得するために使用します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --describe

     mytopic という名前のトピックの設定を取得するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --describe

2. kafka-configs.sh ツールを使用して、現在の設定を変更します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。
   • --alter オプション: 現在の設定を変更するために使用します。

   • --add-config オプション: 追加または変更するオプションを指定します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --alter --add-config <Option>=<Value>

     mytopic という名前のトピックの設定を変更するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --add-config min.insync.replicas=1

3. kafka-configs.sh ツールを使用して、既存の設定オプションを削除します。

   • --bootstrap-server オプションで Kafka ブローカーのホストおよびポートを指定します。
   • --entity-type を topic として、--entity-name をトピックの名前に設定します。
   • --delete-config オプション: 既存の設定オプションを削除するために使用します。

   • --remove-config オプション: 削除するオプションを指定します。

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name <TopicName> --alter --delete-config <Option>

     mytopic という名前のトピックの設定を変更するコマンドの例

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --delete-config min.insync.replicas

トピックの削除

1. kafka-topics.sh ユーティリティーを使用してトピックを削除します。

   • --bootstrap-server における Kafka ブローカーのホストおよびポート。
   • --delete オプション: 既存のトピックを削除することを指定するために使用します。

   • --topic オプション: このオプションでトピック名を指定する必要があります。

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --delete --topic <TopicName>

     mytopic というトピックを作成するコマンドの例

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytopic

2. kafka-topics.sh を使用して、トピックが削除されたことを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

   すべてのトピックをリスト表示するコマンドの例

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

手順

1. テキストエディターでサンプルプロパティーファイルを開くか、新しいプロパティーファイルを作成し、ファイルを編集して接続情報と各 Kafka クラスターのレプリケーションフローを追加します。

   以下の例は、cluster-1 および cluster-2 の 2 つのクラスターを双方向に接続する設定を示しています。クラスター名は、clusters プロパティーで設定できます。

   MirrorMaker 2 の設定例

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<truststore_name>
   cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_
   cluster-1.ssl.keystore.password=<keystore_name>
   cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<truststore_name>
   cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=<keystore_name>
   cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-2->cluster-1.enabled=true 7
   cluster-1->cluster-2.topics=.* 8
   cluster-2->cluster-1.topics=topic-1, topic-2 9
   cluster-1->cluster-2.groups=.* 10
   cluster-2->cluster-1.groups=group-1, group-2 11
   
   replication.policy.separator=- 12
   sync.topic.acls.enabled=false 13
   refresh.topics.interval.seconds=60 14
   refresh.groups.interval.seconds=60 15

   1

   各 Kafka クラスターは、そのエイリアスで識別されます。

   2

   ブートストラップアドレス およびポート 443 を使用した、cluster-1 の接続情報。両方のクラスターはポート 443 を使用し、OpenShift Routes を使用して Kafka に接続します。

   3

   ssl. プロパティーは、cluster-1 の TLS 設定を定義します。

   4

   cluster-2 の接続情報です。

   5

   ssl. プロパティーは、cluster-2 の TLS 設定を定義します。

   6

   cluster-1 から cluster-2 へのレプリケーションフローが有効になっています。

   7

   cluster-2 から cluster-1. へのレプリケーションフローが有効になっています。

   8

   cluster-1 から cluster-2 へのすべてのトピックのレプリケーションです。ソースコネクターは指定のトピックをレプリケーションします。チェックポイントコネクターは、指定されたトピックのオフセットを追跡します。

   9

   cluster-2 から cluster-1 への特定のトピックのレプリケーション。

   10

   cluster-1 から cluster-2 へのすべてのコンシューマーグループのレプリケーション。チェックポイントコネクターは、指定されたコンシューマーグループをレプリケーションします。

   11

   cluster-2 から cluster-1 への特定のコンシューマーグループのレプリケーション。

   12

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

   13

   有効にすると、同期されたトピックに ACL が適用されます。デフォルトは false です。

   14

   新しいトピックの同期をチェックする間隔です。

   15

   新しいコンシューマーグループの同期をチェックする間隔です。

2. オプション: 必要に応じて、リモートトピックの名前の自動変更をオーバーライドするポリシーを追加します。その名前の前にソースクラスターの名前を追加する代わりに、トピックが元の名前を保持します。

   このオプションの設定は、active/passive バックアップおよびデータ移行に使用されます。

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

3. オプション: コンシューマーグループのオフセットを同期する場合は、設定を追加して同期を有効にし、管理します。

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

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

   2

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

   3

   オフセット追跡のチェック頻度を調整します。オフセット同期の頻度を変更する場合は、これらのチェックの頻度も調整することを推奨します。

4. ターゲットクラスターで ZooKeeper および Kafka を起動します。

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon \
   /opt/kafka/config/zookeeper.properties

   /opt/kafka/bin/kafka-server-start.sh -daemon \
   /opt/kafka/config/server.properties

5. プロパティーファイルで定義したクラスター接続設定およびレプリケーションポリシーで MirrorMaker を起動します。

   /opt/kafka/bin/connect-mirror-maker.sh \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker はクラスター間の接続を設定します。

6. ターゲットクラスターごとに、トピックがレプリケーションされていることを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

手順

1. MirrorMaker consumer.properties ファイルと producer.properties ファイルを編集して、MirrorMaker 2 機能をオフにします。

   以下に例を示します。

   replication.policy.class=org.apache.kafka.mirror.LegacyReplicationPolicy 1
   
   refresh.topics.enabled=false 2
   refresh.groups.enabled=false
   emit.checkpoints.enabled=false
   emit.heartbeats.enabled=false
   sync.topic.configs.enabled=false
   sync.topic.acls.enabled=false

   1

   MirrorMaker の以前のバージョンをエミュレートします。

   2

   MirrorMaker 2 の機能 (内部 チェックポイント および ハートビート トピックを含む) が無効になります

2. 変更を保存し、以前のバージョンの MirrorMaker で使用していたプロパティーファイルで MirrorMaker を再起動します。

   su - kafka /opt/kafka/bin/kafka-mirror-maker.sh \
   --consumer.config /opt/kafka/config/consumer.properties \
   --producer.config /opt/kafka/config/producer.properties \
   --num.streams=2

   consumer プロパティーはソースクラスターの設定を提供し、producer プロパティーはターゲットクラスターの設定を提供します。

   MirrorMaker はクラスター間の接続を設定します。

3. ターゲットクラスターで ZooKeeper および Kafka を起動します。

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

4. ターゲットクラスターの場合、トピックがレプリケーションされていることを確認します。

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

手順

1. kafka ユーザーに切り替えます。

   su - kafka

2. kafka-configs.sh ツールを使用して、ブローカーのブローカーロガーのリストを表示します。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --describe --entity-type broker-loggers --entity-name BROKER-ID

   たとえば、ブローカー 0 の場合:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0

   これにより、TRACE、DEBUG、INFO、WARN、ERROR、または FATAL の各ロガーのログレベルが返されます。

   以下に例を示します。

   #...
   kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={}
   kafka.log.TimeIndex=INFO sensitive=false synonyms={}

3. 1 つ以上のブローカーロガーのログレベルを変更します。--alter および --add-config オプションを使用して、各ロガーとそのレベルを二重引用符のコンマ区切りリストとして指定します。

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --alter --add-config "LOGGER-ONE=NEW-LEVEL,LOGGER-TWO=NEW-LEVEL" --entity-type broker-loggers --entity-name BROKER-ID

   たとえば、ブローカー 0 の場合:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0

   成功すると、以下が返されます。

   Completed updating config for broker: 0.

手順

1. kafka ユーザーに切り替えます。

   su - kafka

2. connect-log4j.properties ファイルで設定されているロガーの現在のログレベルを確認します。

   $ cat /opt/kafka/config/connect-log4j.properties
   
   # ...
   log4j.rootLogger=INFO, stdout, connectAppender
   # ...
   log4j.logger.org.apache.zookeeper=ERROR
   log4j.logger.org.reflections=ERROR

   curl コマンドを使用して、Kafka Connect API の admin/loggers エンドポイントからログレベルを確認します。

   curl -s http://localhost:8083/admin/loggers/ | jq
   
   {
     "org.apache.zookeeper": {
       "level": "ERROR"
     },
     "org.reflections": {
       "level": "ERROR"
     },
     "root": {
       "level": "INFO"
     }
   }

   jq は出力を JSON 形式で出力します。このリストには、標準の org および root レベルのロガーに加えて、ログレベルが変更された特定のロガーが表示されます。

   Kafka Connect の admin.listeners 設定に TLS (トランスポート層セキュリティー) 認証を設定する場合、ロガーエンドポイントのアドレスは、https://localhost:8083 など、https としてプロトコルを使用して admin.listeners に指定された値になります。

   特定のロガーのログレベルを取得することもできます。

   curl -s http://localhost:8083/admin/loggers/org.apache.kafka.connect.mirror.MirrorCheckpointConnector | jq
   
   {
     "level": "INFO"
   }

3. PUT メソッドを使用して、ロガーのログレベルを変更します。

   curl -Ss -X PUT -H 'Content-Type: application/json' -d '{"level": "TRACE"}' http://localhost:8083/admin/loggers/root
   
   {
     # ...
   
     "org.reflections": {
       "level": "TRACE"
     },
     "org.reflections.Reflections": {
       "level": "TRACE"
     },
     "root": {
       "level": "TRACE"
     }
   }

   root ロガーを変更すると、デフォルトでルートログレベルを使用していたロガーのログレベルも変更されます。

手順

1. Kafka 設定ファイル /opt/kafka/config/server.properties を編集します。

   プラグインプロパティーは、この設定例のとおりです。

   Kafka Static Quota プラグインの設定例

   # ...
   client.quota.callback.class=io.strimzi.kafka.quotas.StaticQuotaCallback 1
   client.quota.callback.static.produce=1000000 2
   client.quota.callback.static.fetch=1000000 3
   client.quota.callback.static.storage.soft=400000000000 4
   client.quota.callback.static.storage.hard=500000000000 5
   client.quota.callback.static.storage.check-interval=5 6
   # ...

   1

   Kafka Static Quota プラグインを読み込みます。

   2

   プロデューサーのバイトレートしきい値を設定します。この例では 1 MBps です。

   3

   コンシューマーのバイトレートしきい値を設定します。この例では 1 MBps です。

   4

   ストレージのソフト制限の下限を設定します。この例では 400 GB です。

   5

   ストレージのハード制限の上限を設定します。この例では 500 GB です。

   6

   ストレージのチェックの間隔 (秒単位) を設定します。この例では 5 秒です。これを 0 に設定するとチェックを無効にできます。

2. デフォルトの設定ファイルで Kafka ブローカーを起動します。

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

3. Kafka ブローカーが稼働していることを確認します。

   jcmd | grep Kafka

1. サービスプリンシパル
2. Kafka ブローカー (Kerberos ログインを使用するため)
3. ZooKeeper (Kerberos ログインを使用するため)
4. プロデューサーおよびコンシューマークライアント (Kerberos 認証を使用して Kafka にアクセスするため)

1. Kerberos KDC を使用してサービスプリンシパルと、プリンシパルキーを保存するキータブを作成します。

   以下に例を示します。

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     ZooKeeper サービスプリンシパルは、Kafka config/server.properties ファイルの zookeeper.connect 設定と同じホスト名にする必要があります。

     zookeeper.connect=node1.example.redhat.com:2181

     ホスト名が異なる場合は localhost が使用されるため、認証に失敗します。

2. ホストにディレクトリーを作成し、キータブファイルを追加します。

   以下に例を示します。

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. kafka ユーザーがディレクトリーにアクセスできることを確認します。

   chown kafka:kafka -R /opt/kafka/krb5

1. opt/kafka/config/jaas.conf ファイルを作成または変更して、ZooKeeper クライアントおよびサーバー操作をサポートします。

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 1
       storeKey=true 2
       useTicketCache=false 3
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 4
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 5
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   1

   true に設定し、キータブからプリンシパルキーを取得します。

   2

   true に設定し、プリンシパルキーを保存します。

   3

   true に設定し、チケットキャッシュから Ticket Granting Ticket (TGT) を取得します。

   4

   keyTab プロパティーは、Kerberos KDC からコピーされた keytab ファイルの場所を示します。その場所とファイルは、kafka ユーザーが読み取り可能になものにする必要があります。

   5

   principal プロパティーは、KDC ホストで作成された完全修飾プリンシパル名と一致するように設定され、その形式は SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME に従います。

2. opt/kafka/config/zookeeper.properties を編集して、更新された JAAS 設定を使用します。

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 1
   kerberos.removeHostFromPrincipal=false 2
   kerberos.removeRealmFromPrincipal=false 3
   quorum.auth.enableSasl=true 4
   quorum.auth.learnerRequireSasl=true 5
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 6
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 7
   quorum.cnxn.threads.size=20

   1

   ログイン更新の頻度をミリ秒単位で制御します。これは、チケットの更新間隔に合わせて調整できます。デフォルトは 1 時間です。

   2

   ホスト名がログインプリンシパル名の一部として使用されるかどうかを指定します。クラスターのすべてのノードで単一の keytab を使用する場合、これは true に設定されます。ただし、トラブルシューティングのために、各ブローカーホストに個別のキータブと完全修飾プリンシパルを生成することが推奨されます。

   3

   Kerberos ネゴシエーションのプリンシパル名からレルム名を削除するかどうかを制御します。この設定は、false にすることを推奨します。

   4

   ZooKeeper サーバーおよびクライアントの SASL 認証メカニズムを有効にします。

   5

   RequireSasl プロパティーは、マスター選出などのクォーラムイベントに SASL 認証を必要とするかどうかを制御します。

   6

   loginContext プロパティーは、指定されたコンポーネントの認証設定に使用される JAAS 設定のログインコンテキストの名前を識別します。loginContext 名は、opt/kafka/config/jaas.conf ファイルの関連セクションの名前に対応します。

   7

   識別に使用されるプリンシパル名を形成するために使用される命名規則を制御します。プレースホルダー _HOST は、実行時に server.1 プロパティーによって定義されたホスト名に自動的に解決されます。

3. JVM パラメーターで ZooKeeper を起動し、Kerberos ログイン設定を指定します。

   su - kafka
   export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   デフォルトのサービス名 (zookeeper) を使用していない場合は、-Dzookeeper.sasl.client.username=NAME パラメーターを使用して名前を追加します。

   注記

   /etc/krb5.conf を場所として使用している場合は、ZooKeeper、Kafka、Kafka プロデューサーおよびコンシューマーの起動時に -Djava.security.krb5.conf=/etc/krb5.conf を指定する必要はありません。

1. opt/kafka/config/jaas.conf ファイルを以下の要素で修正します。

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. Kafka クラスターの各ブローカーを設定するには、config/server.properties ファイルのリスナー設定を変更して、リスナーが SASL/GSSAPI ログインを使用するようにします。

   リスナーのセキュリティープロトコルのマップに SASL プロトコルを追加し、不要なプロトコルを削除します。

   以下に例を示します。

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 1
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2
   # ..
   sasl.enabled.mechanisms=GSSAPI 3
   sasl.mechanism.inter.broker.protocol=GSSAPI 4
   sasl.kerberos.service.name=kafka 5
   ...

   1

   クライアントとの汎用通信用のセキュアなリスナー (通信用 TLS をサポート)、およびブローカー間通信用のレプリケーションリスナーの 2 つが設定されています。

   2

   TLS に対応しているリスナーのプロトコル名は SASL_PLAINTEXT になります。コネクターが TLS に対応していない場合、プロトコル名は SASL_PLAINTEXT になります。SSL が必要ない場合は、ssl.* プロパティーを削除できます。

   3

   Kerberos 認証における SASL メカニズムは GSSAPI になります。

   4

   inter-broker 通信の Kerberos 認証。

   5

   認証要求に使用するサービスの名前は、同じ Kerberos 設定を使用している他のサービスと区別するために指定されます。

3. Kafka ブローカーを起動し、JVM パラメーターを使用して Kerberos ログイン設定を指定します。

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   ブローカーおよび ZooKeeper クラスターが設定されていて、Kerberos ベース以外の認証システムで動作している場合は、ZooKeeper およびブローカークラスターを起動し、ログで設定エラーを確認できます。

   ブローカーおよび Zookeeper インスタンスを起動すると、Kerberos 認証用にクラスターが設定されました。