AWS でセルフホスト型 Kafka プライベートリンクサービスをセットアップする

このドキュメントでは、AWS でセルフホスト型 Kafka 用の Private Link サービスを設定し、それをTiDB Cloudで動作させる方法について説明します。

このメカニズムは次のように機能します。

1. TiDB Cloud VPC は、プライベートエンドポイントを介して Kafka VPC に接続します。
2. Kafka クライアントはすべての Kafka ブローカーと直接通信する必要があります。
3. 各 Kafka ブローカーは、 TiDB Cloud VPC 内のエンドポイントの一意のポートにマッピングされます。
4. マッピングを実現するには、Kafka ブートストラップ メカニズムと AWS リソースを活用します。

次の図にその仕組みを示します。

このドキュメントでは、AWS の 3つのアベイラビリティゾーン (AZ) にデプロイされた Kafka Private Link サービスへの接続例を示します。同様のポートマッピング原則に基づいて他の構成も可能ですが、このドキュメントでは Kafka Private Link サービスの基本的な設定手順について説明します。本番環境では、運用の保守性と可観測性を強化した、より耐障害性の高い Kafka Private Link サービスの使用をお勧めします。

前提条件

1. 独自の AWS アカウントで Kafka Private Link サービスを設定するには、次の権限があることを確認してください。

   • EC2ノードを管理する
   • VPCを管理する
   • サブネットを管理する
   • セキュリティグループを管理する
   • ロードバランサーの管理
   • エンドポイントサービスの管理
   • EC2 ノードに接続して Kafka ノードを構成する

2. 持っていない場合はTiDB Cloud Dedicatedクラスタを作成する 。

3. TiDB Cloud Dedicated クラスターから Kafka デプロイメント情報を取得します。

   1. TiDB Cloudコンソールで、TiDB クラスターのクラスター概要ページに移動し、左側のナビゲーションペインでData > Changefeedをクリックします。
   2. 概要ページで、TiDB クラスターのリージョンを確認します。Kafka クラスターが同じリージョンにデプロイされることを確認してください。
   3. Create Changefeedをクリックします。
      1. Destinationで、 Kafkaを選択します。
      2. Connectivity MethodでPrivate Linkを選択します。
   4. 続行する前に、 TiDB Cloud AWS アカウントの情報をリマインダーに書き留めておいてください。この情報は、TiDB Cloud がKafka Private Link サービスのエンドポイントを作成することを承認する際に使用されます。
   5. Number of AZsを選択します。この例では、 3 AZsを選択します。KafkaクラスターをデプロイするAZのIDをメモしておいてください。AZ名とAZ IDの関係を知りたい場合は、 AWS リソースのアベイラビリティゾーン IDを参照してください。
   6. Kafka プライベートリンクサービスに固有のKafka Advertised Listener Patternを入力します。
      1. 一意のランダム文字列を入力してください。数字または小文字のみ使用できます。この文字列は、後ほどKafka Advertised Listener Patternを生成する際に使用します。
      2. Check usage and generateをクリックすると、ランダム文字列が一意であるかどうかが確認され、Kafka ブローカーの外部アドバタイズリスナーを組み立てるために使用されるKafka Advertised Listener Patternが生成されます。

すべてのデプロイメント情報をメモしてください。後でKafka Private Linkサービスを設定する際に必要になります。

次の表は、デプロイメント情報の例を示しています。

ステップ1. Kafkaクラスターをセットアップする

新しいクラスターをデプロイする必要がある場合は、 新しいKafkaクラスターをデプロイ手順に従ってください。

既存のクラスターを公開する必要がある場合は、 実行中の Kafka クラスターを再構成する手順に従ってください。

新しいKafkaクラスターをデプロイ

1. Kafka VPC を設定する

Kafka VPC には次のものが必要です。

• ブローカー用のプライベートサブネットが 3つ (AZ ごとに 1つ)。
• 任意の AZ に 1つのパブリックサブネットがあり、インターネットに接続できる要塞ノードと 3つのプライベートサブネットがあるため、Kafka クラスターを簡単にセットアップできます。本番環境では、Kafka VPC に接続できる独自の要塞ノードが必要になる場合があります。

サブネットを作成する前に、AZ IDとAZ名のマッピングに基づいてAZ内にサブネットを作成します。以下のマッピングを例に挙げます。

• usw2-az1 => us-west-2a
• usw2-az2 => us-west-2c
• usw2-az3 => us-west-2b

次の AZ にプライベートサブネットを作成します。

• us-west-2a
• us-west-2c
• us-west-2b

Kafka VPC を作成するには、次の手順を実行します。

1.1. Kafka VPCを作成する

1. AWSコンソール > VPCダッシュボードに進み、Kafka をデプロイするリージョンに切り替えます。

2. Create VPCをクリックします。VPC settingsページで以下の情報を入力します。

   1. VPC onlyを選択します。
   2. Name tagにタグを入力します (例: Kafka VPC 。
   3. IPv4 CIDR manual inputを選択し、 IPv4 CIDR を入力します (例: 10.0.0.0/16 )。
   4. その他のオプションはデフォルト値を使用します。 Create VPCをクリックします。
   5. VPC の詳細ページで、VPC ID (例: vpc-01f50b790fa01dffa ) をメモします。

1.2. Kafka VPC にプライベートサブネットを作成する

1. サブネット一覧ページに進みます。

2. Create subnetをクリックします。

3. 前にメモしておいたVPC ID (この例ではvpc-01f50b790fa01dffa ) を選択します。

4. 以下の情報を含む3つのサブネットを追加します。TiDB Cloudでは、ブローカーのadvertised.listener設定でAZ IDをエンコードする必要があるため、後でブローカーを設定する際に便利なように、サブネット名にAZ IDを含めることをお勧めします。

   • サブネット1 in us-west-2a

     • Subnet name: broker-usw2-az1
     • Availability Zone: us-west-2a
     • IPv4 subnet CIDR block： 10.0.0.0/18

   • サブネット2 in us-west-2c

     • Subnet name: broker-usw2-az2
     • Availability Zone: us-west-2c
     • IPv4 subnet CIDR block： 10.0.64.0/18

   • サブネット3 in us-west-2b

     • Subnet name: broker-usw2-az3
     • Availability Zone: us-west-2b
     • IPv4 subnet CIDR block： 10.0.128.0/18

5. Create subnetをクリックします。Subnets Listingページが表示されます。

1.3. Kafka VPC にパブリックサブネットを作成する

1. Create subnetをクリックします。

2. 前にメモしておいたVPC ID (この例ではvpc-01f50b790fa01dffa ) を選択します。

3. 次の情報を使用して、任意の AZ にパブリックサブネットを追加します。

   • Subnet name: bastion
   • IPv4 subnet CIDR block： 10.0.192.0/18

4. 要塞サブネットをパブリックサブネットに構成します。

   1. VPCダッシュボード > インターネットゲートウェイに進みます。kafka-vpc-igwという名前のインターネットゲートウェイを作成します。

   2. Internet gateways DetailページのActionsで、 Attach to VPCをクリックして、インターネット ゲートウェイを Kafka VPC に接続します。

   3. VPCダッシュボード > ルートテーブルに進みます。Kafka VPC のインターネット ゲートウェイへのルート テーブルを作成し、次の情報を含む新しいルートを追加します。

      • Name: kafka-vpc-igw-route-table
      • VPC : Kafka VPC
      • Route:
        • Destination： 0.0.0.0/0
        • Target： Internet Gateway, kafka-vpc-igw

   4. ルートテーブルを要塞サブネットに接続します。ルートテーブルのDetailページで、 Subnet associations > Edit subnet associationsをクリックし、要塞サブネットを追加して変更を保存します。

2. Kafkaブローカーを設定する

2.1. 要塞ノードを作成する

EC2 リストページに進みます。要塞サブネットに要塞ノードを作成します。

• Name: bastion-node

• Amazon Machine Image: Amazon Linux

• Instance Type: t2.small

• Key pair: kafka-vpc-key-pair 。 kafka-vpc-key-pairという名前の新しいキーペアを作成します。 kafka-vpc-key-pair.pemをローカルマシンにダウンロードして、後で設定します。

• ネットワーク設定

  • VPC : Kafka VPC
  • Subnet: bastion
  • Auto-assign public IP: Enable
  • Security Group：どこからでもSSHログインを許可する新しいセキュリティグループを作成します。本番環境の安全性を確保するために、ルールを絞り込むことができます。

2.2. ブローカーノードを作成する

EC2 リストページに進みます。ブローカー サブネットに、各 AZ に 1つずつ、合計3つのブローカーノードを作成します。

• サブネットbroker-usw2-az1のブローカー 1

  • Name: broker-node1

  • Amazon Machine Image: Amazon Linux

  • Instance Type: t2.large

  • Key pair： kafka-vpc-key-pairを再利用します

  • ネットワーク設定

    • VPC : Kafka VPC
    • Subnet: broker-usw2-az1
    • Auto-assign public IP: Disable
    • Security Group: Kafka VPCからのTCPをすべて許可する新しいセキュリティグループを作成します。本番環境での安全性を考慮して、ルールを絞り込むことができます。
      • Protocol： TCP
      • Port range: 0 - 65535
      • Source: 10.0.0.0/16

• サブネットbroker-usw2-az2のブローカー 2

  • Name: broker-node2

  • Amazon Machine Image: Amazon Linux

  • Instance Type: t2.large

  • Key pair： kafka-vpc-key-pairを再利用します

  • ネットワーク設定

    • VPC : Kafka VPC
    • Subnet: broker-usw2-az2
    • Auto-assign public IP: Disable
    • Security Group: Kafka VPCからのTCPをすべて許可する新しいセキュリティグループを作成します。本番環境での安全性を考慮して、ルールを絞り込むことができます。
      • Protocol： TCP
      • Port range: 0 - 65535
      • Source: 10.0.0.0/16

• サブネットbroker-usw2-az3のブローカー 3

  • Name: broker-node3

  • Amazon Machine Image: Amazon Linux

  • Instance Type: t2.large

  • Key pair： kafka-vpc-key-pairを再利用します

  • ネットワーク設定

    • VPC : Kafka VPC
    • Subnet: broker-usw2-az3
    • Auto-assign public IP: Disable
    • Security Group: Kafka VPCからのTCPをすべて許可する新しいセキュリティグループを作成します。本番環境での安全性を考慮して、ルールを絞り込むことができます。
      • Protocol： TCP
      • Port range: 0 - 65535
      • Source: 10.0.0.0/16

2.3. Kafkaランタイムバイナリの準備

1. 要塞ノードの詳細ページに移動します。パブリックIPv4アドレスを取得します。SSHを使用して、先ほどダウンロードしたkafka-vpc-key-pair.pemを使用してノードにログインします。

   chmod 400 kafka-vpc-key-pair.pem
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{bastion_public_ip} # replace {bastion_public_ip} with the IP address of your bastion node, for example, 54.186.149.187
   scp -i "kafka-vpc-key-pair.pem" kafka-vpc-key-pair.pem ec2-user@{bastion_public_ip}:~/
   

2. バイナリをダウンロードします。

   # Download Kafka and OpenJDK, and then extract the files. You can choose the binary version based on your preference.
   wget https://archive.apache.org/dist/kafka/3.7.1/kafka_2.13-3.7.1.tgz
   tar -zxf kafka_2.13-3.7.1.tgz
   wget https://download.java.net/java/GA/jdk22.0.2/c9ecb94cd31b495da20a27d4581645e8/9/GPL/openjdk-22.0.2_linux-x64_bin.tar.gz
   tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz
   

3. バイナリを各ブローカーノードにコピーします。

   # Replace {broker-node1-ip} with your broker-node1 IP address
   scp -i "kafka-vpc-key-pair.pem" kafka_2.13-3.7.1.tgz ec2-user@{broker-node1-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node1-ip} "tar -zxf kafka_2.13-3.7.1.tgz"
   scp -i "kafka-vpc-key-pair.pem" openjdk-22.0.2_linux-x64_bin.tar.gz ec2-user@{broker-node1-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node1-ip} "tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz"
   
   # Replace {broker-node2-ip} with your broker-node2 IP address
   scp -i "kafka-vpc-key-pair.pem" kafka_2.13-3.7.1.tgz ec2-user@{broker-node2-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node2-ip} "tar -zxf kafka_2.13-3.7.1.tgz"
   scp -i "kafka-vpc-key-pair.pem" openjdk-22.0.2_linux-x64_bin.tar.gz ec2-user@{broker-node2-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node2-ip} "tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz"
   
   # Replace {broker-node3-ip} with your broker-node3 IP address
   scp -i "kafka-vpc-key-pair.pem" kafka_2.13-3.7.1.tgz ec2-user@{broker-node3-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node3-ip} "tar -zxf kafka_2.13-3.7.1.tgz"
   scp -i "kafka-vpc-key-pair.pem" openjdk-22.0.2_linux-x64_bin.tar.gz ec2-user@{broker-node3-ip}:~/
   ssh -i "kafka-vpc-key-pair.pem" ec2-user@{broker-node3-ip} "tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz"
   

2.4. 各ブローカーノードにKafkaノードを設定する

2.4.1 3つのノードを持つKRaft Kafkaクラスターをセットアップする

各ノードはブローカーとコントローラーの役割を担います。各ブローカーに対して以下の操作を実行してください。

1. listeners項目の場合、3つのブローカーはすべて同じであり、ブローカーとコントローラーのロールとして機能します。

   1. すべてのコントローラーロールノードに同じ CONTROLLER リスナーを設定します。ブローカーロールノードのみを追加する場合は、 server.propertiesの CONTROLLER リスナーは必要ありません。
   2. ブローカーリスナーを 2つ構成します。INTERNALは内部アクセス用、 EXTERNALはTiDB Cloudからの外部アクセス用です。

2. advertised.listeners項目については、次の操作を行います。

   1. 各ブローカーに対して、ブローカーノードの内部IPアドレスを使用して、INTERNALアドバタイズリスナーを設定します。アドバタイズされた内部Kafkaクライアントは、このアドレスを使用してブローカーにアクセスします。

   2. TiDB Cloudから取得したKafka Advertised Listener Patternに基づいて、各ブローカーノードに外部アドバタイズリスナーを設定することで、 TiDB Cloudが複数のブローカーを区別できるようになります。異なる外部アドバタイズリスナーを設定することで、 TiDB CloudのKafkaクライアントはリクエストを適切なブローカーにルーティングできるようになります。

      • <port>は、Kafka プライベートリンクサービスのアクセスポイントでブローカーを区別します。すべてのブローカーの EXTERNAL アドバタイズリスナーのポート範囲を計画してください。これらのポートは、ブローカーが実際にリッスンするポートである必要はありません。これらは、リクエストを異なるブローカーに転送するプライベートリンクサービスのロードバランサーがリッスンするポートです。
      • Kafka Advertised Listener PatternのAZ IDは、ブローカーがデプロイされている場所を示します。TiDB Cloud は、 AZ ID に基づいてリクエストを異なるエンドポイント DNS 名にルーティングします。

   トラブルシューティングを容易にするために、ブローカーごとに異なるブローカー ID を構成することをお勧めします。

3. 計画値は次のとおりです。

   • CONTROLLER port: 29092
   • INTERNAL port： 9092
   • EXTERNAL: 39092
   • EXTERNAL advertised listener ports range: 9093~9095

2.4.2. 設定ファイルを作成する

SSHを使用してすべてのブローカーノードにログインします。以下の内容の設定ファイル~/config/server.propertiesを作成します。

# brokers in usw2-az1

# broker-node1 ~/config/server.properties
# 1. Replace {broker-node1-ip}, {broker-node2-ip}, {broker-node3-ip} with the actual IP addresses.
# 2. Configure EXTERNAL in "advertised.listeners" based on the "Kafka Advertised Listener Pattern" in the "Prerequisites" section.
# 2.1 The pattern for AZ(ID: usw2-az1) is "<broker_id>.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:<port>".
# 2.2 So the EXTERNAL can be "b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093". Replace <broker_id> with "b" prefix plus "node.id" properties, and replace <port> with a unique port (9093) in the port range of the EXTERNAL advertised listener.
# 2.3 If there are more broker role nodes in the same AZ, you can configure them in the same way.
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@{broker-node1-ip}:29092,2@{broker-node2-ip}:29092,3@{broker-node3-ip}:29092
listeners=INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:29092,EXTERNAL://0.0.0.0:39092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://{broker-node1-ip}:9092,EXTERNAL://b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093
controller.listener.names=CONTROLLER
listener.security.protocol.map=INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
log.dirs=./data

# brokers in usw2-az2

# broker-node2 ~/config/server.properties
# 1. Replace {broker-node1-ip}, {broker-node2-ip}, {broker-node3-ip} with the actual IP addresses.
# 2. Configure EXTERNAL in "advertised.listeners" based on the "Kafka Advertised Listener Pattern" in the "Prerequisites" section.
# 2.1 The pattern for AZ(ID: usw2-az2) is "<broker_id>.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:<port>".
# 2.2 So the EXTERNAL can be "b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094". Replace <broker_id> with "b" prefix plus "node.id" properties, and replace <port> with a unique port (9094) in the port range of the EXTERNAL advertised listener.
# 2.3 If there are more broker role nodes in the same AZ, you can configure them in the same way.
process.roles=broker,controller
node.id=2
controller.quorum.voters=1@{broker-node1-ip}:29092,2@{broker-node2-ip}:29092,3@{broker-node3-ip}:29092
listeners=INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:29092,EXTERNAL://0.0.0.0:39092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://{broker-node2-ip}:9092,EXTERNAL://b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094
controller.listener.names=CONTROLLER
listener.security.protocol.map=INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
log.dirs=./data

# brokers in usw2-az3

# broker-node3 ~/config/server.properties
# 1. Replace {broker-node1-ip}, {broker-node2-ip}, {broker-node3-ip} with the actual IP addresses.
# 2. Configure EXTERNAL in "advertised.listeners" based on the "Kafka Advertised Listener Pattern" in the "Prerequisites" section.
# 2.1 The pattern for AZ(ID: usw2-az3) is "<broker_id>.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:<port>".
# 2.2 So the EXTERNAL can be "b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095". Replace <broker_id> with "b" prefix plus "node.id" properties, and replace <port> with a unique port (9095) in the port range of the EXTERNAL advertised listener.
# 2.3 If there are more broker role nodes in the same AZ, you can configure them in the same way.
process.roles=broker,controller
node.id=3
controller.quorum.voters=1@{broker-node1-ip}:29092,2@{broker-node2-ip}:29092,3@{broker-node3-ip}:29092
listeners=INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:29092,EXTERNAL://0.0.0.0:39092
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://{broker-node3-ip}:9092,EXTERNAL://b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095
controller.listener.names=CONTROLLER
listener.security.protocol.map=INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
log.dirs=./data

2.4.3 Kafkaブローカーを起動する

スクリプトを作成し、それを実行して各ブローカーノードで Kafka ブローカーを起動します。

#!/bin/bash

# Get the directory of the current script
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Set JAVA_HOME to the Java installation within the script directory
export JAVA_HOME="$SCRIPT_DIR/jdk-22.0.2"
# Define the vars
KAFKA_DIR="$SCRIPT_DIR/kafka_2.13-3.7.1/bin"
KAFKA_STORAGE_CMD=$KAFKA_DIR/kafka-storage.sh
KAFKA_START_CMD=$KAFKA_DIR/kafka-server-start.sh
KAFKA_DATA_DIR=$SCRIPT_DIR/data
KAFKA_LOG_DIR=$SCRIPT_DIR/log
KAFKA_CONFIG_DIR=$SCRIPT_DIR/config

# Cleanup step, which makes it easy for multiple experiments
# Find all Kafka process IDs
KAFKA_PIDS=$(ps aux | grep 'kafka.Kafka' | grep -v grep | awk '{print $2}')
if [ -z "$KAFKA_PIDS" ]; then
  echo "No Kafka processes are running."
else
  # Kill each Kafka process
  echo "Killing Kafka processes with PIDs: $KAFKA_PIDS"
  for PID in $KAFKA_PIDS; do
    kill -9 $PID
    echo "Killed Kafka process with PID: $PID"
  done
  echo "All Kafka processes have been killed."
fi

rm -rf $KAFKA_DATA_DIR
mkdir -p $KAFKA_DATA_DIR
rm -rf $KAFKA_LOG_DIR
mkdir -p $KAFKA_LOG_DIR

# Magic id: BRl69zcmTFmiPaoaANybiw, you can use your own
$KAFKA_STORAGE_CMD format -t "BRl69zcmTFmiPaoaANybiw" -c "$KAFKA_CONFIG_DIR/server.properties" > $KAFKA_LOG_DIR/server_format.log
LOG_DIR=$KAFKA_LOG_DIR nohup $KAFKA_START_CMD "$KAFKA_CONFIG_DIR/server.properties" &

2.5. 要塞ノードでクラスター設定をテストする

1. Kafka ブートストラップをテストします。

   export JAVA_HOME=/home/ec2-user/jdk-22.0.2
   
   # Bootstrap from INTERNAL listener
   ./kafka_2.13-3.7.1/bin/kafka-broker-api-versions.sh --bootstrap-server {one_of_broker_ip}:9092 | grep 9092
   # Expected output (the actual order might be different)
   {broker-node1-ip}:9092 (id: 1 rack: null) -> (
   {broker-node2-ip}:9092 (id: 2 rack: null) -> (
   {broker-node3-ip}:9092 (id: 3 rack: null) -> (
   
   # Bootstrap from EXTERNAL listener
   ./kafka_2.13-3.7.1/bin/kafka-broker-api-versions.sh --bootstrap-server {one_of_broker_ip}:39092
   # Expected output for the last 3 lines (the actual order might be different)
   # The difference in the output from "bootstrap from INTERNAL listener" is that exceptions or errors might occur because advertised listeners cannot be resolved in Kafka VPC.
   # We will make them resolvable in TiDB Cloud side and make it route to the right broker when you create a changefeed connect to this Kafka cluster by Private Link.
   b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   

2. 要塞ノードにプロデューサー スクリプトproduce.shを作成します。

   #!/bin/bash
   BROKER_LIST=$1 # "{broker_address1},{broker_address2}..."
   
   # Get the directory of the current script
   SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
   # Set JAVA_HOME to the Java installation within the script directory
   export JAVA_HOME="$SCRIPT_DIR/jdk-22.0.2"
   # Define the Kafka directory
   KAFKA_DIR="$SCRIPT_DIR/kafka_2.13-3.7.1/bin"
   TOPIC="test-topic"
   
   # Create a topic if it does not exist
   create_topic() {
       echo "Creating topic if it does not exist..."
       $KAFKA_DIR/kafka-topics.sh --create --topic $TOPIC --bootstrap-server $BROKER_LIST --if-not-exists --partitions 3 --replication-factor 3
   }
   
   # Produce messages to the topic
   produce_messages() {
       echo "Producing messages to the topic..."
       for ((chrono=1; chrono <= 10; chrono++)); do
           message="Test message "$chrono
           echo "Create "$message
           echo $message | $KAFKA_DIR/kafka-console-producer.sh --broker-list $BROKER_LIST --topic $TOPIC
       done
   }
   create_topic
   produce_messages
   

3. 要塞ノードにコンシューマー スクリプトconsume.shを作成します。

   #!/bin/bash
   
   BROKER_LIST=$1 # "{broker_address1},{broker_address2}..."
   
   # Get the directory of the current script
   SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
   # Set JAVA_HOME to the Java installation within the script directory
   export JAVA_HOME="$SCRIPT_DIR/jdk-22.0.2"
   # Define the Kafka directory
   KAFKA_DIR="$SCRIPT_DIR/kafka_2.13-3.7.1/bin"
   TOPIC="test-topic"
   CONSUMER_GROUP="test-group"
   # Consume messages from the topic
   consume_messages() {
       echo "Consuming messages from the topic..."
       $KAFKA_DIR/kafka-console-consumer.sh --bootstrap-server $BROKER_LIST --topic $TOPIC --from-beginning --timeout-ms 5000 --consumer-property group.id=$CONSUMER_GROUP
   }
   consume_messages
   

4. produce.shとconsume.shを実行して、Kafkaクラスターが実行中であることを確認してください。これらのスクリプトは、後ほどネットワーク接続テストにも再利用されます。スクリプトは--partitions 3 --replication-factor 3のトピックを作成します。これら3つのブローカーすべてにデータが含まれていることを確認してください。ネットワーク接続がテストされることを保証するために、スクリプトが3つのブローカーすべてに接続することを確認してください。

   # Test write message.
   ./produce.sh {one_of_broker_ip}:9092
   

   # Expected output
   Creating topic if it does not exist...
   
   Producing messages to the topic...
   Create Test message 1
   >>Create Test message 2
   >>Create Test message 3
   >>Create Test message 4
   >>Create Test message 5
   >>Create Test message 6
   >>Create Test message 7
   >>Create Test message 8
   >>Create Test message 9
   >>Create Test message 10
   

   # Test read message
   ./consume.sh {one_of_broker_ip}:9092
   

   # Expected example output (the actual message order might be different)
   Consuming messages from the topic...
   Test message 3
   Test message 4
   Test message 5
   Test message 9
   Test message 10
   Test message 6
   Test message 8
   Test message 1
   Test message 2
   Test message 7
   [2024-11-01 08:54:27,547] ERROR Error processing message, terminating consumer process:  (kafka.tools.ConsoleConsumer$)
   org.apache.kafka.common.errors.TimeoutException
   Processed a total of 10 messages
   

実行中の Kafka クラスターを再構成する

Kafka クラスターが TiDB クラスターと同じリージョンおよび AZ にデプロイされていることを確認してください。ブローカーが異なる AZ にある場合は、正しい AZ に移動してください。

1. ブローカーの外部リスナーを構成する

以下の設定はKafka KRaftクラスターに適用されます。ZKモードの設定も同様です。

1. 構成の変更を計画します。

   1. TiDB Cloudからの外部アクセス用に、各ブローカーに EXTERNALリスナーを設定します。EXTERNAL ポートとして、一意のポート（例： 39092 ）を選択します。

   2. TiDB Cloudから取得したKafka Advertised Listener Patternに基づいて、各ブローカーノードにEXTERNALアドバタイズリスナーを設定することで、 TiDB Cloudが複数のブローカーを区別できるようになります。異なるEXTERNALアドバタイズリスナーを設定することで、TiDB CloudのKafkaクライアントはリクエストを適切なブローカーにルーティングできるようになります。

      • <port>は、Kafka プライベートリンクサービスのアクセスポイントでブローカーを区別します。すべてのブローカーの EXTERNAL アドバタイズリスナーのポート範囲を計画してください（例： range from 9093 ）。これらのポートは、ブローカーが実際にリッスンするポートである必要はありません。これらは、リクエストを異なるブローカーに転送するプライベートリンクサービスのロードバランサーがリッスンするポートです。
      • Kafka Advertised Listener PatternのAZ IDは、ブローカーがデプロイされている場所を示します。TiDB Cloud は、 AZ ID に基づいてリクエストを異なるエンドポイント DNS 名にルーティングします。

   トラブルシューティングを容易にするために、ブローカーごとに異なるブローカー ID を構成することをお勧めします。

2. SSHを使用して各ブローカーノードにログインします。各ブローカーの設定ファイルを以下の内容に変更します。

   # brokers in usw2-az1
   
   # Add EXTERNAL listener
   listeners=INTERNAL:...,EXTERNAL://0.0.0.0:39092
   
   # Add EXTERNAL advertised listeners based on the "Kafka Advertised Listener Pattern" in "Prerequisites" section
   # 1. The pattern for AZ(ID: usw2-az1) is "<broker_id>.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:<port>"
   # 2. So the EXTERNAL can be "b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093", replace <broker_id> with "b" prefix plus "node.id" properties, replace <port> with a unique port(9093) in EXTERNAL advertised listener ports range
   advertised.listeners=...,EXTERNAL://b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093
   
   # Configure EXTERNAL map
   listener.security.protocol.map=...,EXTERNAL:PLAINTEXT
   

   # brokers in usw2-az2
   
   # Add EXTERNAL listener
   listeners=INTERNAL:...,EXTERNAL://0.0.0.0:39092
   
   # Add EXTERNAL advertised listeners based on the "Kafka Advertised Listener Pattern" in "Prerequisites" section
   # 1. The pattern for AZ(ID: usw2-az2) is "<broker_id>.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:<port>"
   # 2. So the EXTERNAL can be "b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094". Replace <broker_id> with "b" prefix plus "node.id" properties, and replace <port> with a unique port(9094) in EXTERNAL advertised listener ports range.
   advertised.listeners=...,EXTERNAL://b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094
   
   # Configure EXTERNAL map
   listener.security.protocol.map=...,EXTERNAL:PLAINTEXT
   

   # brokers in usw2-az3
   
   # Add EXTERNAL listener
   listeners=INTERNAL:...,EXTERNAL://0.0.0.0:39092
   
   # Add EXTERNAL advertised listeners based on the "Kafka Advertised Listener Pattern" in "Prerequisites" section
   # 1. The pattern for AZ(ID: usw2-az3) is "<broker_id>.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:<port>"
   # 2. So the EXTERNAL can be "b2.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095". Replace <broker_id> with "b" prefix plus "node.id" properties, and replace <port> with a unique port(9095) in EXTERNAL advertised listener ports range.
   advertised.listeners=...,EXTERNAL://b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095
   
   # Configure EXTERNAL map
   listener.security.protocol.map=...,EXTERNAL:PLAINTEXT
   

3. すべてのブローカーを再構成したら、Kafka ブローカーを 1つずつ再起動します。

2. 内部ネットワークで外部リスナーの設定をテストする

Kafka と OpenJDK を Kafka クライアント ノードにダウンロードできます。

# Download Kafka and OpenJDK, and then extract the files. You can choose the binary version based on your preference.
wget https://archive.apache.org/dist/kafka/3.7.1/kafka_2.13-3.7.1.tgz
tar -zxf kafka_2.13-3.7.1.tgz
wget https://download.java.net/java/GA/jdk22.0.2/c9ecb94cd31b495da20a27d4581645e8/9/GPL/openjdk-22.0.2_linux-x64_bin.tar.gz
tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz

次のスクリプトを実行して、ブートストラップが期待どおりに動作するかどうかをテストします。

export JAVA_HOME=/home/ec2-user/jdk-22.0.2

# Bootstrap from the EXTERNAL listener
./kafka_2.13-3.7.1/bin/kafka-broker-api-versions.sh --bootstrap-server {one_of_broker_ip}:39092

# Expected output for the last 3 lines (the actual order might be different)
# There will be some exceptions or errors because advertised listeners cannot be resolved in your Kafka network.
# We will make them resolvable in TiDB Cloud side and make it route to the right broker when you create a changefeed connect to this Kafka cluster by Private Link.
b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException

ステップ 2. Kafka クラスターをプライベートリンクサービスとして公開する

1. ロードバランサーを設定する

異なるポートを持つ4つのターゲットグループを持つネットワークロードバランサーを作成します。1つのターゲットグループはブートストラップ用で、他のターゲットグループは異なるブローカーにマッピングされます。

1. ブートストラップターゲットグループ => 9092 => broker-node1:39092、broker-node2:39092、broker-node3:39092
2. ブローカーターゲットグループ1 => 9093 => broker-node1:39092
3. ブローカーターゲットグループ2 => 9094 => broker-node2:39092
4. ブローカーターゲットグループ3 => 9095 => broker-node3:39092

ブローカーロールノードが複数ある場合は、マッピングを追加する必要があります。ブートストラップターゲットグループに少なくとも1つのノードがあることを確認してください。耐障害性を確保するため、各AZに1つずつ、合計3つのノードを追加することをお勧めします。

ロードバランサーを設定するには、次の手順を実行します。

1. 対象グループに進み、4つのターゲットグループを作成します。

   • ブートストラップターゲットグループ

     • Target type: Instances
     • Target group name： bootstrap-target-group
     • Protocol： TCP
     • Port: 9092
     • IP address type: IPv4
     • VPC : Kafka VPC
     • Health check protocol： TCP
     • Register targets: broker-node1:39092, broker-node2:39092, broker-node3:39092

   • ブローカーターゲットグループ1

     • Target type: Instances
     • Target group name： broker-target-group-1
     • Protocol： TCP
     • Port: 9093
     • IP address type: IPv4
     • VPC : Kafka VPC
     • Health check protocol： TCP
     • Register targets: broker-node1:39092

   • ブローカーターゲットグループ2

     • Target type: Instances
     • Target group name： broker-target-group-2
     • Protocol： TCP
     • Port: 9094
     • IP address type: IPv4
     • VPC : Kafka VPC
     • Health check protocol： TCP
     • Register targets: broker-node2:39092

   • ブローカーターゲットグループ3

     • Target type: Instances
     • Target group name： broker-target-group-3
     • Protocol： TCP
     • Port: 9095
     • IP address type: IPv4
     • VPC : Kafka VPC
     • Health check protocol： TCP
     • Register targets: broker-node3:39092

2. ロードバランサーに進み、ネットワークロードバランサーを作成します。

   • Load balancer name: kafka-lb
   • Schema: Internal
   • Load balancer IP address type: IPv4
   • VPC : Kafka VPC
   • Availability Zones:
     • usw2-az1とbroker-usw2-az1 subnet
     • usw2-az2とbroker-usw2-az2 subnet
     • usw2-az3とbroker-usw2-az3 subnet
   • Security groups: 次のルールで新しいセキュリティグループを作成します。
     • 受信ルールは、Kafka VPCからのすべてのTCPを許可します：タイプ - {ports of target groups} （例： 9092-9095 ）、ソース - {CIDR of TiDB Cloud} 。リージョン内のTiDB CloudのCIDRを取得するには、 TiDB Cloudコンソールの左上隅にあるコンボボックスを使用してターゲットプロジェクトに切り替え、左側のナビゲーションペインでProject Settings > Network Accessをクリックし、 Project CIDR > AWSをクリックします。
     • アウトバウンドルールは、Kafka VPC へのすべての TCP を許可します: タイプ - All TCP 、宛先 - Anywhere-IPv4
   • リスナーとルーティング:
     • プロトコル: TCP ; ポート: 9092 ; 転送先: bootstrap-target-group
     • プロトコル: TCP ; ポート: 9093 ; 転送先: broker-target-group-1
     • プロトコル: TCP ; ポート: 9094 ; 転送先: broker-target-group-2
     • プロトコル: TCP ; ポート: 9095 ; 転送先: broker-target-group-3

3. 要塞ノードでロードバランサーをテストします。この例では、Kafka ブートストラップのみをテストします。ロードバランサーは Kafka EXTERNAL リスナーをリッスンしているため、EXTERNAL アドバタイズリスナーのアドレスは要塞ノードでは解決できません。ロードバランサーの詳細ページからkafka-lb DNS 名（例： kafka-lb-77405fa57191adcb.elb.us-west-2.amazonaws.com ）をメモしておいてください。要塞ノードでスクリプトを実行してください。

   # Replace {lb_dns_name} to your actual value
   export JAVA_HOME=/home/ec2-user/jdk-22.0.2
   ./kafka_2.13-3.7.1/bin/kafka-broker-api-versions.sh --bootstrap-server {lb_dns_name}:9092
   
   # Expected output for the last 3 lines (the actual order might be different)
   b1.usw2-az1.abc.us-west-2.aws.3199015.tidbcloud.com:9093 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   b2.usw2-az2.abc.us-west-2.aws.3199015.tidbcloud.com:9094 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   b3.usw2-az3.abc.us-west-2.aws.3199015.tidbcloud.com:9095 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException
   
   # You can also try bootstrap in other ports 9093/9094/9095. It will succeed probabilistically because NLB in AWS resolves LB DNS to the IP address of any availability zone and disables cross-zone load balancing by default.
   # If you enable cross-zone load balancing in LB, it will succeed. However, it is unnecessary and might cause additional cross-AZ traffic.
   

2. プライベートリンクサービスを設定する

1. エンドポイントサービスに進みます。 Create endpoint serviceをクリックして、Kafka ロードバランサーのプライベートリンクサービスを作成します。

   • Name: kafka-pl-service
   • Load balancer type: Network
   • Load balancers： kafka-lb
   • Included Availability Zones: usw2-az1,usw2-az2, usw2-az3
   • Require acceptance for endpoint: Acceptance required
   • Enable private DNS name： No

2. Service nameをメモしておいてください。TiDB Cloudに提供する必要があります（例com.amazonaws.vpce.us-west-2.vpce-svc-0f49e37e1f022cd45 ）。

3. kafka-pl-service の詳細ページで、 Allow principalsタブをクリックし、 TiDB Cloudの AWS アカウントにエンドポイントの作成を許可します。TiDB Cloudの AWS アカウントは前提条件 （例： arn:aws:iam::<account_id>:root ）で取得できます。

ステップ3. TiDB Cloudから接続する

1. TiDB Cloudコンソールに戻って、TiDB Cloud Dedicatedクラスターの変更フィードを作成し、Private Linkを使用してKafkaクラスターに接続します。詳細については、 Apache Kafka にシンクするを参照してください。

2. Configure the changefeed target > Connectivity Method > Private Linkに進むときは、次のフィールドに対応する値を入力し、必要に応じてその他のフィールドを入力します。

   • Kafka Type: 3 AZs。クラスターが同じ 3つの AZ にデプロイされていることを確認します。
   • Kafka Advertised Listener Pattern: abc 。これは、 前提条件でKafka Advertised Listener Patternを生成するために使用する一意のランダム文字列と同じです。
   • Endpoint Service Name: Kafka サービス名。
   • Bootstrap Ports: 9092。背後に専用のブートストラップ ターゲットグループを構成するため、1つのポートで十分です。

3. Apache Kafka にシンクするの手順に進みます。

これでタスクは正常に完了しました。

FAQ

2つの異なるTiDB Cloudプロジェクトから同じ Kafka Private Link サービスに接続するにはどうすればよいですか?

このドキュメントに従って最初のプロジェクトからの接続をすでに正常に設定している場合は、次のようにして 2 番目のプロジェクトから同じ Kafka Private Link サービスに接続できます。

1. このドキュメントの冒頭の指示に従ってください。

2. ステップ1. Kafkaクラスターをセットアップするに進んだら、 実行中の Kafka クラスターを再構成するに従って、EXTERNAL リスナーとアドバタイズリスナーの別のグループを作成します。このグループの名前はEXTERNAL2とします。EXTERNAL2 のポート範囲はEXTERNALと重複できないことに注意してください。

3. ブローカーを再構成した後、ブートストラップおよびブローカー ターゲットグループを含む別のターゲットグループをロードバランサーに追加します。

4. 次の情報を使用してTiDB Cloud接続を構成します。

   • 新しいブートストラップポート
   • 新しい Kafka アドバタイズリスナーグループ
   • 同じエンドポイントサービス