--- title: Set Up Self-Hosted Kafka Private Service Connect in Google Cloud summary: このドキュメントでは、Google Cloud でセルフホスト型 Kafka 用に Private Service Connect を設定し、それをTiDB Cloudで動作させる方法について説明します。 --- # Google Cloud でセルフホスト型 Kafka Private Service Connect を設定する {#set-up-self-hosted-kafka-private-service-connect-in-google-cloud} このドキュメントでは、Google Cloud でセルフホスト型 Kafka 用に Private Service Connect を設定する方法と、それをTiDB Cloudで動作させる方法について説明します。 このメカニズムは次のように機能します。 1. TiDB Cloud VPC は、プライベートエンドポイントを介して Kafka VPC に接続します。 2. Kafka クライアントはすべての Kafka ブローカーと直接通信する必要があります。 3. 各 Kafka ブローカーは、 TiDB Cloud VPC 内の一意のポートにマッピングされます。 4. マッピングを実現するには、Kafka ブートストラップメカニズムと Google Cloud リソースを活用します。 Google Cloud でセルフホスト型 Kafka に Private Service Connect を設定するには、次の2つの方法があります。 - Private Service Connect（PSC）ポートマッピングメカニズムを使用します。この方法では、静的なポートブローカーマッピング設定が必要です。EXTERNALリスナーとアドバタイズリスナーのグループを追加するには、既存のKafkaクラスターを再構成する必要があります。詳細は[PSC ポートマッピングによるセルフホスト型 Kafka Private Service Connect サービスの設定](#set-up-self-hosted-kafka-private-service-connect-service-by-psc-port-mapping)を参照してください。 - [Kafka-proxy](https://github.com/grepplabs/kafka-proxy)を使用してください。この方法では、Kafka クライアントと Kafka ブローカー間のプロキシとして、追加の実行プロセスが導入されます。プロキシはポートとブローカーのマッピングを動的に設定し、リクエストを転送します。既存の Kafka クラスターを再設定する必要はありません。詳細は[Kafka-proxy によるセルフホスト型 Kafka Private Service Connect のセットアップ](#set-up-self-hosted-kafka-private-service-connect-by-kafka-proxy)を参照してください。 このドキュメントでは、Google Cloud の 3つのアベイラビリティゾーン（AZ）にデプロイされた Kafka Private Service Connect サービスへの接続例を示します。同様のポートマッピング原則に基づいて他の構成も可能ですが、このドキュメントでは Kafka Private Service Connect サービスの基本的な設定プロセスについて説明します。本番環境では、運用の保守性と可観測性を強化した、より回復力の高い Kafka Private Service Connect サービスの使用を推奨します。 ## 前提条件 {#prerequisites} 1. 独自の Google Cloud アカウントで Kafka Private Service Connect を設定するには、次の権限があることを確認してください。 - VMノードを管理する - VPCを管理する - サブネットを管理する - ロードバランサーを管理する - Private Service Connect を管理する - VMノードに接続してKafkaノードを構成する 2. 持っていない場合は[TiDB Cloud Dedicatedクラスタを作成する](/tidb-cloud/create-tidb-cluster.md) 。 3. TiDB Cloud Dedicated クラスターから Kafka デプロイメント情報を取得します。 1. [TiDB Cloudコンソール](https://tidbcloud.com)で[**My TiDB**](https://tidbcloud.com/tidbs)ページに移動し、ターゲットの TiDB Cloud Dedicated クラスターの名前をクリックして概要ページに移動します。 2. 概要ページで、TiDB クラスターのリージョンを確認します。Kafka クラスターが同じリージョンにデプロイされることを確認してください。 3. 左側のナビゲーションペインで**Data** > **Changefeed**をクリックし、右上隅の**Create Changefeed**をクリックして、次の情報を入力します。 1. **Destination**で、 **Kafka**を選択します。 2. **Connectivity Method**で、 **Private Service Connect**を選択します。 4. **Reminders before proceeding**で Google Cloud プロジェクトをメモしておいてください。このプロジェクトは、 TiDB Cloudからのエンドポイント作成リクエストの自動承認を承認するために使用します。 5. **Zones of TiDB Cluster**をメモしておいてください。これらのゾーンに TiDB クラスターをデプロイします。ゾーン間のトラフィックを削減するため、これらのゾーンに Kafka をデプロイすることをお勧めします。 6. Kafka Private Service Connect サービスに固有の**Kafka Advertised Listener Pattern**を選択します。 1. 一意のランダム文字列を入力してください。数字または小文字のみ使用できます。この文字列は、後ほど**Kafka Advertised Listener Pattern**を生成する際に使用します。 2. **Check usage and generate**をクリックすると、ランダム文字列が一意であるかどうかを確認し、Kafka ブローカーの EXTERNAL アドバタイズリスナーを組み立てるため、または Kafka-proxy を構成するために使用される**Kafka Advertised Listener Pattern**を生成します。 すべてのデプロイメント情報をメモしてください。後でKafka Private Service Connectサービスを設定する際に必要になります。 次の表は、デプロイメント情報の例を示しています。 | 情報 | 値 | | ------------------------------- | ------------------------------------------------------------------------------------------------------ | | リージョン | オレゴン州 ( `us-west1` ) | | TiDB Cloudの Google Cloud プロジェクト | `tidbcloud-prod-000` | | ゾーン | | | Kafka Advertised Listener Pattern | 一意のランダム文字列: `abc`
生成されたパターン: <broker_id>.abc.us-west1.gcp.3199745.tidbcloud.com:<port> | ## PSC ポートマッピングによるセルフホスト型 Kafka Private Service Connect サービスの設定 {#set-up-self-hosted-kafka-private-service-connect-service-by-psc-port-mapping} PSCポートマッピングメカニズムを使用して、各KafkaブローカーをTiDB Cloud VPCに固有のポートで公開します。次の図は、その仕組みを示しています。 ![Connect to Google Cloud self-hosted Kafka Private Service Connect by port mapping](/media/tidb-cloud/changefeed/connect-to-google-cloud-self-hosted-kafka-private-service-connect-by-portmapping.jpeg) ### ステップ1. Kafkaクラスターをセットアップする {#step-1-set-up-the-kafka-cluster} 新しいクラスターをデプロイする必要がある場合は、 [新しいKafkaクラスターをデプロイする](#deploy-a-new-kafka-cluster)の手順に従ってください。 既存のクラスターを公開する必要がある場合は、 [実行中の Kafka クラスターを再構成する](#reconfigure-a-running-kafka-cluster)の手順に従ってください。 #### 新しいKafkaクラスターをデプロイする {#deploy-a-new-kafka-cluster} **1. Kafka VPC をセットアップする** Kafka クラスターを簡単に構成できるように、Kafka VPC 用に 2つのサブネット (1つは Kafka ブローカー用、もう 1つは要塞ノード用) を作成する必要があります。 [Google Cloud コンソール](https://cloud.google.com/cloud-console)に進み、 [VPCネットワーク](https://console.cloud.google.com/networking/networks/list)ページに移動して、次の属性を持つ Kafka VPC を作成します。 - **Name**: `kafka-vpc` - サブネット - **Name**: `bastion-subnet` ;**Region**: `us-west1` ; **IPv4 range**: `10.0.0.0/18` - **Name**: `brokers-subnet` ;**Region**: `us-west1` ; **IPv4 range**: `10.64.0.0/18` - ファイアウォールルール - `kafka-vpc-allow-custom` - `kafka-vpc-allow-ssh` **2. VMのプロビジョニング** VM をプロビジョニングするには、 [VMインスタンス](https://console.cloud.google.com/compute/instances)ページに移動します。 1. 要塞ノード - **Name**: `bastion-node` - **Region**: `us-west1` - **Zone**: `Any` - **Machine Type**： `e2-medium` - **Image**： `Debian GNU/Linux 12` - **Network**: `kafka-vpc` - **Subnetwork**: `bastion-subnet` - **External IPv4 address**: `Ephemeral` 2. ブローカーノード1 - **Name**: `broker-node1` - **Region**: `us-west1` - **Zone**: `us-west1-a` - **Machine Type**： `e2-medium` - **Image**： `Debian GNU/Linux 12` - **Network**: `kafka-vpc` - **Subnetwork**: `brokers-subnet` - **External IPv4 address**: `None` 3. ブローカーノード2 - **Name**: `broker-node2` - **Region**: `us-west1` - **Zone**: `us-west1-b` - **Machine Type**： `e2-medium` - **Image**： `Debian GNU/Linux 12` - **Network**: `kafka-vpc` - **Subnetwork**: `brokers-subnet` - **External IPv4 address**: `None` 4. ブローカーノード3 - **Name**: `broker-node3` - **Region**: `us-west1` - **Zone**: `us-west1-c` - **Machine Type**： `e2-medium` - **Image**： `Debian GNU/Linux 12` - **Network**: `kafka-vpc` - **Subnetwork**: `brokers-subnet` - **External IPv4 address**: `None` **3. Kafkaランタイムバイナリを準備する** 1. 要塞ノードの詳細ページに移動します。**SSH**をクリックして要塞ノードにログインします。バイナリをダウンロードします。 ```shell # 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 ``` 2. バイナリを各ブローカーノードにコピーします。 ```shell # Run this command to authorize gcloud to access the Cloud Platform with Google user credentials # Follow the instruction in output to finish the login gcloud auth login # Copy binaries to broker nodes gcloud compute scp kafka_2.13-3.7.1.tgz openjdk-22.0.2_linux-x64_bin.tar.gz broker-node1:~ --zone=us-west1-a gcloud compute ssh broker-node1 --zone=us-west1-a --command="tar -zxf kafka_2.13-3.7.1.tgz && tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz" gcloud compute scp kafka_2.13-3.7.1.tgz openjdk-22.0.2_linux-x64_bin.tar.gz broker-node2:~ --zone=us-west1-b gcloud compute ssh broker-node2 --zone=us-west1-b --command="tar -zxf kafka_2.13-3.7.1.tgz && tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz" gcloud compute scp kafka_2.13-3.7.1.tgz openjdk-22.0.2_linux-x64_bin.tar.gz broker-node3:~ --zone=us-west1-c gcloud compute ssh broker-node3 --zone=us-west1-c --command="tar -zxf kafka_2.13-3.7.1.tgz && tar -zxf openjdk-22.0.2_linux-x64_bin.tar.gz" ``` **4. Kafkaブローカーを構成する** 1. 3つのノードでKRaft Kafkaクラスターをセットアップします。各ノードはブローカーとコントローラーの役割を持ちます。各ブローカーに対して、以下の手順を実行します。 1. `listeners`の場合、3つのブローカーはすべて同じであり、ブローカーとコントローラーのロールとして機能します。 1. すべての**コントローラー**ロールノードに同じ CONTROLLER リスナーを設定します。**ブローカー**ロールノードのみを追加する場合は、 `server.properties`の CONTROLLER リスナーは必要ありません。 2. 2 つの**ブローカー**リスナーを構成します。内部アクセスの場合は INTERNAL、 TiDB Cloudからの外部アクセスの場合は EXTERNAL です。 2. `advertised.listeners`については、次の操作を行います。 1. ブローカーノードの内部 IP アドレスを使用して、各ブローカーの INTERNAL アドバタイズリスナーを構成します。これにより、内部 Kafka クライアントはアドバタイズアドレスを介してブローカーに接続できるようになります。 2. TiDB Cloudから取得した**Kafka Advertised Listener Pattern**に基づいて、各ブローカーノードにEXTERNALアドバタイズリスナーを設定することで、TiDB Cloudが複数のブローカーを区別できるようになります。異なるEXTERNALアドバタイズリスナーを設定することで、 TiDB Cloud側のKafkaクライアントはリクエストを適切なブローカーにルーティングできるようになります。 - ``は、Kafka Private Service Connect アクセスポイントでブローカーを区別します。すべてのブローカーの EXTERNAL アドバタイズリスナーのポート範囲を計画してください。これらのポートは、ブローカーが実際にリッスンするポートである必要はありません。これらは、リクエストを異なるブローカーに転送する Private Service Connect のロードバランサーがリッスンするポートです。 - トラブルシューティングを容易にするために、ブローカーごとに異なるブローカー ID を構成することをお勧めします。 3. 計画値: - CONTROLLER ポート: `29092` - INTERNAL ポート: `9092` - EXTERNAL: `39092` - EXTERNAL アドバタイズリスナーポートの範囲: `9093~9095` 2. SSHを使用して各ブローカーノードにログインします。各ブローカーノードごとに、以下の内容を含む設定ファイル`~/config/server.properties`を作成します。 ```properties # 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 is ".abc.us-west1.gcp.3199745.tidbcloud.com:". # 2.2 So the EXTERNAL can be "b1.abc.us-west1.gcp.3199745.tidbcloud.com:9093". Replace with "b" prefix plus "node.id" properties, and replace with a unique port (9093) in the port range of the EXTERNAL advertised listener. 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.abc.us-west1.gcp.3199745.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 ``` ```properties # 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 is ".abc.us-west1.gcp.3199745.tidbcloud.com:". # 2.2 So the EXTERNAL can be "b2.abc.us-west1.gcp.3199745.tidbcloud.com:9094". Replace with "b" prefix plus "node.id" properties, and replace with a unique port (9094) in the port range of the EXTERNAL advertised listener. 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.abc.us-west1.gcp.3199745.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 ``` ```properties # 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 is ".abc.us-west1.gcp.3199745.tidbcloud.com:". # 2.2 So the EXTERNAL can be "b3.abc.us-west1.gcp.3199745.tidbcloud.com:9095". Replace with "b" prefix plus "node.id" properties, and replace with a unique port (9095) in the port range of the EXTERNAL advertised listener. 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.abc.us-west1.gcp.3199745.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 ``` 3. スクリプトを作成し、それを実行して各ブローカーノードで Kafka ブローカーを起動します。 ```shell #!/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 magic ID. $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" & ``` **5. 要塞ノードでKafkaクラスターをテストする** 1. Kafka ブートストラップをテストします。 ```shell export JAVA_HOME=~/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 Service Connect. b1.abc.us-west1.gcp.3199745.tidbcloud.com:9093 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b2.abc.us-west1.gcp.3199745.tidbcloud.com:9094 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b3.abc.us-west1.gcp.3199745.tidbcloud.com:9095 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException ``` 2. 要塞ノードにプロデューサースクリプト`produce.sh`を作成します。 ```shell #!/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`を作成します。 ```shell #!/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つのブローカーすべてに接続されることを確認してください。 ```shell # Test write message. ./produce.sh {one_of_broker_ip}:9092 ``` ```text # 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 ``` ```shell # Test read message ./consume.sh {one_of_broker_ip}:9092 ``` ```text # 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 クラスターを再構成する {#reconfigure-a-running-kafka-cluster} Kafka クラスターが TiDB クラスターと同じリージョンにデプロイされていることを確認してください。ゾーン間のトラフィックを削減するため、ゾーンも同じリージョンに配置することをお勧めします。 **1. ブローカーの EXTERNAL リスナーを構成する** 以下の設定はKafka KRaftクラスターに適用されます。ZKモードの設定も同様です。 1. 構成の変更を計画します。 1. TiDB Cloudからの外部アクセス用に、各ブローカーに EXTERNAL**リスナー**を設定します。EXTERNAL ポートとして一意のポート（例： `39092` ）を選択します。 2. TiDB Cloudから取得した**Kafka Advertised Listener Pattern**に基づいて、各ブローカーノードにEXTERNAL**アドバタイズリスナー**を設定することで、TiDB Cloudが複数のブローカーを区別できるようになります。異なるEXTERNALアドバタイズリスナーを設定することで、 TiDB Cloud側のKafkaクライアントはリクエストを適切なブローカーにルーティングできるようになります。 - ``は、Kafka Private Service Connect アクセスポイントでブローカーを区別します。すべてのブローカーの EXTERNAL アドバタイズリスナーのポート範囲を計画します（例： `range from 9093` ）。これらのポートは、ブローカーが実際にリッスンするポートである必要はありません。これらは、リクエストを別のブローカーに転送する Private Service Connect のロードバランサーがリッスンするポートです。 - トラブルシューティングを容易にするために、ブローカーごとに異なるブローカー ID を構成することをお勧めします。 2. SSHを使用して各ブローカーノードにログインします。各ブローカーの設定ファイルを以下の内容に変更します。 ```properties # Add EXTERNAL listener listeners=INTERNAL:...,EXTERNAL://0.0.0.0:39092 # Add EXTERNAL advertised listeners based on the "Kafka Advertised Listener Pattern" in the "Prerequisites" section # 1. The pattern is ".abc.us-west1.gcp.3199745.tidbcloud.com:". # 2. So the EXTERNAL can be "bx.abc.us-west1.gcp.3199745.tidbcloud.com:xxxx". Replace with "b" prefix plus "node.id" properties, and replace with a unique port in the port range of the EXTERNAL advertised listener. # For example advertised.listeners=...,EXTERNAL://b1.abc.us-west1.gcp.3199745.tidbcloud.com:9093 # Configure EXTERNAL map listener.security.protocol.map=...,EXTERNAL:PLAINTEXT ``` 3. すべてのブローカーを再構成したら、Kafka ブローカーを 1つずつ再起動します。 **2. 内部ネットワークで EXTERNAL リスナーの設定をテストする** Kafka と OpenJDK を Kafka クライアントノードにダウンロードできます。 ```shell # 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 ``` 次のスクリプトを実行して、ブートストラップが期待どおりに動作するかどうかをテストします。 ```shell export JAVA_HOME=~/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 Service Connect. b1.abc.us-west1.gcp.3199745.tidbcloud.com:9093 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b2.abc.us-west1.gcp.3199745.tidbcloud.com:9094 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b3.abc.us-west1.gcp.3199745.tidbcloud.com:9095 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException ``` ### ステップ2. KafkaクラスターをPrivate Service Connectとして公開する {#step-2-expose-the-kafka-cluster-as-private-service-connect} 1. [ネットワークエンドポイントグループ](https://console.cloud.google.com/compute/networkendpointgroups/list)ページに移動します。以下の手順でネットワークエンドポイントグループを作成します。 - **Name**: `kafka-neg` - **Network endpoint group type**: `Port Mapping NEG(Regional)` - **Region**: `us-west1` - **Network**: `kafka-vpc` - **Subnet**: `brokers-subnet` 2. ネットワークエンドポイントグループの詳細ページに移動し、ネットワークエンドポイントを追加して、ブローカーノードへのポートマッピングを構成します。 1. ネットワークエンドポイント1 - **Instance**: `broker-node1` - **VM Port**: `39092` - **Client Port**: `9093` 2. ネットワークエンドポイント2 - **Instance**: `broker-node2` - **VM Port**: `39092` - **Client Port**: `9094` 3. ネットワークエンドポイント3 - **Instance**: `broker-node3` - **VM Port**: `39092` - **Client Port**: `9095` 3. [負荷分散](https://console.cloud.google.com/net-services/loadbalancing/list/loadBalancers)ページに移動します。以下の手順でロードバランサーを作成します。 - **Type of load balancer**: `Network Load Balancer` - **Proxy or Passthrough**: `Passthrough` - **Public facing or internal**： `Internal` - **Load Balancer name**: `kafka-lb` - **Region**: `us-west1` - **Network**: `kafka-vpc` - バックエンド構成 - **Backend type**： `Port mapping network endpoint group` - **Protocol**： `TCP` - **Port mapping network endpoint group**: `kafka-neg` - フロントエンド構成 - **Subnetwork**: `brokers-subnet` - **Ports**: `All` 4. [**Private Service Connect** > **PUBLISH SERVICE**](https://console.cloud.google.com/net-services/psc/list/producers)に進みます。 - **Load Balancer Type**: `Internal passthrough Network Load Balancer` - **Internal load balancer**： `kafka-lb` - **Service name**： `kafka-psc` - **Subnets**: `RESERVE NEW SUBNET` - **Name**: `psc-subnet` - **VPC Network**: `kafka-vpc` - **Region**: `us-west1` - **IPv4 range**: `10.128.0.0/18` - **Accepted projects**: [前提条件](#prerequisites)で取得したTiDB Cloudの Google Cloud プロジェクト (例: `tidbcloud-prod-000` )。 5. `kafka-psc`の詳細ページに移動します。**Service attachment**（例： `projects/tidbcloud-dp-stg-000/regions/us-west1/serviceAttachments/kafka-psc` ）を書き留めます。TiDB Cloudでこの PSC に接続する際に使用します。 6. VPC ネットワーク`kafka-vpc`の詳細ページに移動し、すべてのブローカーへの PSC トラフィックを許可するファイアウォールルールを追加します。 - **Name**: `allow-psc-traffic` - **Direction of traffic**： `Ingress` - **Action on match**： `Allow` - **Targets**： `All instances in the network` - **Source filter**: `IPv4 ranges` - **Source IPv4 ranges**: `10.128.0.0/18` 。psc-subnetの範囲。 - **Protocols and ports**: すべて許可 ### ステップ3. TiDB Cloudから接続する {#step-3-connect-from-tidb-cloud} 1. [TiDB Cloudコンソール](https://tidbcloud.com)に戻り、クラスターが**Private Service Connect**経由で Kafka クラスターに接続するための changefeed を作成します。詳細については、 [Apache Kafka にシンクする](/tidb-cloud/changefeed-sink-to-apache-kafka.md)を参照してください。 2. **Configure the changefeed target** > **Connectivity Method** > **Private Service Connect**に進むときは、次のフィールドに対応する値を入力し、必要に応じてその他のフィールドを入力します。 - **Kafka Advertised Listener Pattern**: `abc` 。これは、 [前提条件](#prerequisites)で**Kafka Advertised Listener Pattern**を生成するために使用する一意のランダム文字列と同じです。 - **Service Attachment**: PSC の Kafka サービスアタッチメント (例: `projects/tidbcloud-dp-stg-000/regions/us-west1/serviceAttachments/kafka-psc` )。 - **Bootstrap Ports**: `9092,9093,9094` 3. [Apache Kafka にシンクする](/tidb-cloud/changefeed-sink-to-apache-kafka.md)の手順に進みます。 ## Kafka-proxy によるセルフホスト型 Kafka Private Service Connect のセットアップ {#set-up-self-hosted-kafka-private-service-connect-by-kafka-proxy} Kafka-proxyの動的ポートマッピングメカニズムを使用して、各KafkaブローカーをTiDB Cloud VPCに固有のポートで公開します。次の図は、その仕組みを示しています。 ![Connect to Google Cloud self-hosted Kafka Private Service Connect by Kafka proxy](/media/tidb-cloud/changefeed/connect-to-google-cloud-self-hosted-kafka-private-service-connect-by-kafka-proxy.jpeg) ### ステップ1. Kafka-proxyを設定する {#step-1-set-up-kafka-proxy} TiDB クラスターと同じリージョンで既に Kafka クラスターが稼働していると仮定します。VPC ネットワークから Kafka クラスターに接続できます。Kafka クラスターは、セルフホスト型でも、Confluent などのサードパーティプロバイダーが提供するものでも構いません。 1. [インスタンスグループ](https://console.cloud.google.com/compute/instanceGroups/list)ページに移動し、Kafka-proxy のインスタンスグループを作成します。 - **Name**: `kafka-proxy-ig` - インスタンステンプレート: - **Name**: `kafka-proxy-tpl` - **Location**： `Regional` - **Region**: `us-west1` - **Machine type**: `e2-medium`。ワークロードに応じて独自のマシンタイプを選択できます。 - **Network**: Kafka クラスターに接続できる VPC ネットワーク。 - **Subnetwork**: Kafka クラスターに接続できるサブネット。 - **External IPv4 address**： `Ephemeral` 。Kafka-proxyの設定を容易にするため、インターネットアクセスを有効にしてください。本番環境では**None**を選択し、任意の方法でノードにログインできます。 - **Location**： `Single zone` - **Region**: `us-west1` - **Zone**: ブローカーのゾーンの 1つを選択します。 - **Autoscaling mode**: `Off` - **Minimum number of instances**: `1` - **Maximum number of instances**: `1` 。Kafka-proxyはクラスターモードをサポートしていないため、デプロイできるインスタンスは1つだけです。各Kafka-proxyはローカルポートをブローカーのポートにランダムにマッピングするため、プロキシごとにマッピングが異なります。ロードバランサーの背後に複数のKafka-proxyをデプロイすると、問題が発生する可能性があります。Kafkaクライアントが1つのプロキシに接続し、別のプロキシを経由してブローカーにアクセスすると、リクエストが誤ったブローカーにルーティングされる可能性があります。 2. kafka-proxy-ig のノードの詳細ページに移動します。**SSH**をクリックしてノードにログインします。バイナリをダウンロードします。 ```shell # You can choose another version wget https://github.com/grepplabs/kafka-proxy/releases/download/v0.3.11/kafka-proxy-v0.3.11-linux-amd64.tar.gz tar -zxf kafka-proxy-v0.3.11-linux-amd64.tar.gz ``` 3. Kafka-proxy を実行し、Kafka ブローカーに接続します。 ```shell # There are three kinds of parameters that need to feed to the Kafka-proxy # 1. --bootstrap-server-mapping defines the bootstrap mapping. Suggest that you configure three mappings, one for each zone for resilience. # a) Kafka broker address; # b) Local address for the broker in Kafka-proxy; # c) Advertised listener for the broker if Kafka clients bootstrap from Kafka-proxy # 2. --dynamic-sequential-min-port defines the start port of the random mapping for other brokers # 3. --dynamic-advertised-listener defines advertised listener address for other brokers based on the pattern obtained from the "Prerequisites" section # a) The pattern: .abc.us-west1.gcp.3199745.tidbcloud.com: # b) Make sure to replace with a fixed lowercase string, for example, "brokers". You can use your own string. This step will help TiDB Cloud route requests properly. # c) Remove ":" # d) The advertised listener address would be: brokers.abc.us-west1.gcp.3199745.tidbcloud.com ./kafka-proxy server \ --bootstrap-server-mapping "{address_of_broker1},0.0.0.0:9092,b1.abc.us-west1.gcp.3199745.tidbcloud.com:9092" \ --bootstrap-server-mapping "{address_of_broker2},0.0.0.0:9093,b2.abc.us-west1.gcp.3199745.tidbcloud.com:9093" \ --bootstrap-server-mapping "{address_of_broker3},0.0.0.0:9094,b3.abc.us-west1.gcp.3199745.tidbcloud.com:9094" \ --dynamic-sequential-min-port=9095 \ --dynamic-advertised-listener=brokers.abc.us-west1.gcp.3199745.tidbcloud.com > ./kafka_proxy.log 2>&1 & ``` 4. Kafka-proxy ノードでブートストラップをテストします。 ```shell # 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=~/jdk-22.0.2 ./kafka_2.13-3.7.1/bin/kafka-broker-api-versions.sh --bootstrap-server 0.0.0.0:9092 # Expected output of the last few lines (the actual order might be different) # There might be exceptions or errors because advertised listeners cannot be resolved in your 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 Service Connect. b1.abc.us-west1.gcp.3199745.tidbcloud.com:9092 (id: 1 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b2.abc.us-west1.gcp.3199745.tidbcloud.com:9093 (id: 2 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException b3.abc.us-west1.gcp.3199745.tidbcloud.com:9094 (id: 3 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException brokers.abc.us-west1.gcp.3199745.tidbcloud.com:9095 (id: 4 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException brokers.abc.us-west1.gcp.3199745.tidbcloud.com:9096 (id: 5 rack: null) -> ERROR: org.apache.kafka.common.errors.DisconnectException ... ``` ### ステップ2. Kafka-proxyをPrivate Service Connect サービスとして公開する {#step-2-expose-kafka-proxy-as-private-service-connect-service} 1. [負荷分散](https://console.cloud.google.com/net-services/loadbalancing/list/loadBalancers)ページに移動して、ロードバランサーを作成します。 - **Type of load balancer**: `Network Load Balancer` - **Proxy or Passthrough**: `Passthrough` - **Public facing or internal**： `Internal` - **Load Balancer name**: `kafka-proxy-lb` - **Region**: `us-west1` - **Network**: あなたのネットワーク - バックエンド構成 - **Backend type**： `Instance group` - **Protocol**： `TCP` - **Instance group**: `kafka-proxy-ig` - フロントエンド構成 - **Subnetwork**: サブネット - **Ports**: `All` - ヘルスチェック: - **Name**: `kafka-proxy-hc` - **Scope**： `Regional` - **Protocol**： `TCP` - **Port**: `9092`。Kafka-proxy のブートストラップポートの 1つを選択できます。 2. [**Private Service Connect** > **PUBLISH SERVICE**](https://console.cloud.google.com/net-services/psc/list/producers)に進みます。 - **Load Balancer Type**: `Internal passthrough Network Load Balancer` - **Internal load balancer**： `kafka-proxy-lb` - **Service name**： `kafka-proxy-psc` - **Subnets**: `RESERVE NEW SUBNET` - **Name**: `proxy-psc-subnet` - **VPC Network**: あなたのネットワーク - **Region**: `us-west1` - **IPv4 range**: ネットワーク計画に基づいて CIDR を設定します - **Accepted projects**: [前提条件](#prerequisites)で取得したTiDB Cloudの Google Cloud プロジェクト (例: `tidbcloud-prod-000` )。 3. **kafka-proxy-psc**の詳細ページに移動します。 `Service attachment` （例： `projects/tidbcloud-dp-stg-000/regions/us-west1/serviceAttachments/kafka-proxy-psc` ）をメモします。これは、 TiDB Cloudがこの PSC に接続する際に使用されます。 4. VPC ネットワークの詳細ページに移動し、すべてのブローカーの PSC トラフィックを許可するファイアウォールルールを追加します。 - **Name**: `allow-proxy-psc-traffic` - **Direction of traffic**： `Ingress` - **Action on match**： `Allow` - **Targets**: ネットワーク内のすべてのインスタンス - **Source filter**: `IPv4 ranges` - **Source IPv4 ranges**: proxy-psc-subnet の CIDR - **Protocols and ports**: すべて許可 ### ステップ3. TiDB Cloudから接続する {#step-3-connect-from-tidb-cloud} 1. [TiDB Cloudコンソール](https://tidbcloud.com)に戻り、クラスターが**Private Service Connect**経由で Kafka クラスターに接続するための changefeed を作成します。詳細については、 [Apache Kafka にシンクする](/tidb-cloud/changefeed-sink-to-apache-kafka.md)を参照してください。 2. **Configure the changefeed target** > **Connectivity Method** > **Private Service Connect**に進んだ後、次のフィールドに対応する値を入力し、必要に応じてその他のフィールドを入力します。 - **Kafka Advertised Listener Pattern**: `abc` . [前提条件](#prerequisites)で**Kafka Advertised Listener Pattern**を生成するために使用する一意のランダム文字列と同じです。 - **Service Attachment**: PSC の kafka-proxy サービスアタッチメント (例: `projects/tidbcloud-dp-stg-000/regions/us-west1/serviceAttachments/kafka-proxy-psc` )。 - **Bootstrap Ports**: `9092,9093,9094` 3. 引き続き[Apache Kafka にシンクする](/tidb-cloud/changefeed-sink-to-apache-kafka.md)のガイドラインに従ってください。 ## FAQ {#faq} ### 2つの異なるTiDB Cloudプロジェクトから同じ Kafka Private Service Connect サービスに接続するにはどうすればよいですか? {#how-to-connect-to-the-same-kafka-private-service-connect-service-from-two-different-tidb-cloud-projects} すでにこのドキュメントの手順に従って最初のプロジェクトからの接続を正常に設定していて、2 番目のプロジェクトから 2 番目の接続を設定する場合は、次のようにして 2つの異なるTiDB Cloudプロジェクトから同じ Kafka Private Service Connect サービスに接続できます。 - PSC ポートマッピングによって Kafka PSC を設定する場合は、次の手順を実行します。 1. このドキュメントの冒頭の指示に従ってください。[ステップ1. Kafkaクラスターをセットアップする](#step-1-set-up-the-kafka-cluster)に進んだら、 [実行中の Kafka クラスターを再構成する](#reconfigure-a-running-kafka-cluster)セクションに従って、EXTERNAL リスナーとアドバタイズリスナーの別のグループを作成してください。このグループの名前は`EXTERNAL2`とします。`EXTERNAL2`のポート範囲は EXTERNAL と重複できないことに注意してください。 2. ブローカーを再構成した後、ネットワークエンドポイントの別のグループをネットワークエンドポイントグループに追加し、ポート範囲を`EXTERNAL2`リスナーにマップします。 3. 新しい変更フィードを作成するには、次の入力でTiDB Cloud接続を構成します。 - 新しいブートストラップポート - 新しい Kafka アドバタイズリスナーパターン - 同じサービスアタッチメント - [Kafka-proxy によるセルフホスト型 Kafka Private Service Connect のセットアップ](#set-up-self-hosted-kafka-private-service-connect-by-kafka-proxy)の場合は、新しい Kafka アドバタイズリスナーパターンを使用して、最初から新しい Kafka-proxy PSC を作成します。