Java: Apache Kafka® のサンプルコード¶
このチュートリアルでは、Apache Kafka® クラスターにメッセージを生成し、このクラスターからメッセージを消費する Java クライアントアプリケーションを実行します。
チュートリアルの実行後に、提供された ソースコード をリファレンスとして使用して、独自の Kafka クライアントアプリケーションを開発します。
前提条件¶
クライアント¶
- Java 1.8 以降（デモアプリケーションの実行用）
- Maven（デモアプリケーションのコンパイル用）
Kafka クラスター¶
- このチュートリアルは、どの環境の Kafka クラスターにも対応しています。
- Confluent Cloud 内
- ローカルホスト 上
- 任意のリモート Kafka クラスター
- Confluent Cloud で実行している場合は、API キーおよびシークレットで Confluent Cloud クラスターにアクセスできることが必要です。
- Confluent Cloud にサインアップし、プロモーションコード
C50INTEGを使用した最初の 20 名のユーザーは、$50 相当を無料で使用できます（詳細）。 - Confluent Cloud で Kafka クラスター、資格情報、および ACL を自動的に作成する方法については、「Confluent Cloud 向け ccloud-stack ユーティリティ」を参照してください。
- Confluent Cloud にサインアップし、プロモーションコード
セットアップ¶
confluentinc/examples GitHub リポジトリのクローンを作成し、
6.0.6-postブランチをチェックアウトします。git clone https://github.com/confluentinc/examples cd examples git checkout 6.0.6-post
Java のサンプルのディレクトリに変更します。
cd clients/cloud/java/Kafka クラスターに接続するための構成パラメーターを含むローカルファイル（
$HOME/.confluent/java.configなど）を作成します。以下のテンプレートのいずれかをベースとして、クラスターへの接続情報でこのファイルをカスタマイズします。{{ BROKER_ENDPOINT }}、{{CLUSTER_API_KEY }}、および{{ CLUSTER_API_SECRET }}に値を代入します（この値を手作業で見つける方法については、「Configure Confluent Cloud Clients」を参照してください。 または、Confluent Cloud 向け ccloud-stack ユーティリティ を使用して値を自動作成します）。Confluent Cloud のテンプレート構成ファイル
# Required connection configs for Kafka producer, consumer, and admin bootstrap.servers={{ BROKER_ENDPOINT }} security.protocol=SASL_SSL sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}'; sasl.mechanism=PLAIN # Required for correctness in Apache Kafka clients prior to 2.6 client.dns.lookup=use_all_dns_ips # Best practice for Kafka producer to prevent data loss acks=allローカルホストのテンプレート構成ファイル
# Kafka bootstrap.servers=localhost:9092
基本プロデューサーおよびコンシューマー、Kafka Streams¶
このサンプルでは、プロデューサーアプリケーションが Kafka クラスターのトピックに Kafka データを書き込みます。Kafka クラスターにトピックがまだ存在しない場合、プロデューサーアプリケーションは Kafka Admin Client API を使用してトピックを作成します。Kafka に書き込まれる各レコードには、ユーザー名（alice など）を表すキーと、json フォーマットのカウント値（{"count": 0} など）があります。コンシューマーアプリケーションは、同じ Kafka トピックを読み取り、各レコードの処理中にカウントのローリング合計を保持します。
レコードの生成¶
Java コードをコンパイルします。
mvn clean package
プロデューサーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- トピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.ProducerExample" \ -Dexec.args="$HOME/.confluent/java.config test1"
プロデューサーがすべてのメッセージを送信したことを確認します。次のように表示されます。
... Producing record: alice {"count":0} Producing record: alice {"count":1} Producing record: alice {"count":2} Producing record: alice {"count":3} Producing record: alice {"count":4} Producing record: alice {"count":5} Producing record: alice {"count":6} Producing record: alice {"count":7} Producing record: alice {"count":8} Producing record: alice {"count":9} Produced record to topic test1 partition [0] @ offset 0 Produced record to topic test1 partition [0] @ offset 1 Produced record to topic test1 partition [0] @ offset 2 Produced record to topic test1 partition [0] @ offset 3 Produced record to topic test1 partition [0] @ offset 4 Produced record to topic test1 partition [0] @ offset 5 Produced record to topic test1 partition [0] @ offset 6 Produced record to topic test1 partition [0] @ offset 7 Produced record to topic test1 partition [0] @ offset 8 Produced record to topic test1 partition [0] @ offset 9 10 messages were produced to topic test1 ...プロデューサーコード を表示します。
レコードの消費¶
コンシューマーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- 前の手順で使用したトピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.ConsumerExample" \ -Dexec.args="$HOME/.confluent/java.config test1"
コンシューマーがすべてのメッセージを受信したことを確認します。次のように表示されます。
... Polling Consumed record with key alice and value {"count":0}, and updated total count to 0 Consumed record with key alice and value {"count":1}, and updated total count to 1 Consumed record with key alice and value {"count":2}, and updated total count to 3 Consumed record with key alice and value {"count":3}, and updated total count to 6 Consumed record with key alice and value {"count":4}, and updated total count to 10 Consumed record with key alice and value {"count":5}, and updated total count to 15 Consumed record with key alice and value {"count":6}, and updated total count to 21 Consumed record with key alice and value {"count":7}, and updated total count to 28 Consumed record with key alice and value {"count":8}, and updated total count to 36 Consumed record with key alice and value {"count":9}, and updated total count to 45 Polling ...入力し終わったら、
Ctrl + Cキーを押します（"+" はキーを同時に押すことを意味します）。コンシューマーコード を表示します。
Kafka Streams¶
Kafka Streams アプリケーションを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- 前の手順で使用したトピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.StreamsExample" \ -Dexec.args="$HOME/.confluent/java.config test1"
Kafka Streams アプリケーションがすべてのメッセージを処理したことを確認します。次のように表示されます。
... [Consumed record]: alice, 0 [Consumed record]: alice, 1 [Consumed record]: alice, 2 [Consumed record]: alice, 3 [Consumed record]: alice, 4 [Consumed record]: alice, 5 [Consumed record]: alice, 6 [Consumed record]: alice, 7 [Consumed record]: alice, 8 [Consumed record]: alice, 9 ... [Running count]: alice, 0 [Running count]: alice, 1 [Running count]: alice, 3 [Running count]: alice, 6 [Running count]: alice, 10 [Running count]: alice, 15 [Running count]: alice, 21 [Running count]: alice, 28 [Running count]: alice, 36 [Running count]: alice, 45 ...
入力し終わったら、
Ctrl + Cキーを押します（"+" はキーを同時に押すことを意味します）。Kafka Streams コード を表示します。
Avro と Confluent Cloud Schema Registry¶
このサンプルは、前のサンプルと似ていますが、値は Avro フォーマットです。また、Confluent Cloud Schema Registry と統合されています。Confluent Cloud Schema Registry を使用する前に、その 可用性と制限 を確認してください。
Confluent Cloud GUI の「Quick Start for Schema Management on Confluent Cloud」にある説明に従って、Confluent Cloud Schema Registry を有効にし、API キーおよびシークレットを作成して接続します。
ご使用の VPC が Confluent Cloud Schema Registry 公衆インターネットエンドポイントに接続できることを確認します。
Schema Registry に接続するための構成パラメーターを含むローカル構成ファイル（
$HOME/.confluent/java.configなど）をアップデートします。Confluent Cloud のテンプレート構成ファイル
# Required connection configs for Kafka producer, consumer, and admin bootstrap.servers={{ BROKER_ENDPOINT }} security.protocol=SASL_SSL sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}'; sasl.mechanism=PLAIN # Required for correctness in Apache Kafka clients prior to 2.6 client.dns.lookup=use_all_dns_ips # Best practice for Kafka producer to prevent data loss acks=all # Required connection configs for Confluent Cloud Schema Registry schema.registry.url=https://{{ SR_ENDPOINT }} basic.auth.credentials.source=USER_INFO basic.auth.user.info={{ SR_API_KEY }}:{{ SR_API_SECRET }}ローカルホストのテンプレート構成ファイル
# Kafka bootstrap.servers=localhost:9092 # Confluent Schema Registry schema.registry.url=http://localhost:8081
Schema Registry サブジェクトのリストを表示して Confluent Cloud Schema Registry の資格情報を確認します。次の例の
{{ SR_API_KEY }}、{{ SR_API_SECRET }}、および{{ SR_ENDPOINT }}に値を代入します。curl -u {{ SR_API_KEY }}:{{ SR_API_SECRET }} https://{{ SR_ENDPOINT }}/subjects
Avro レコードの生成¶
Avro プロデューサーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- トピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.ProducerAvroExample" \ -Dexec.args="$HOME/.confluent/java.config test2"
プロデューサー Avro コード を表示します。
Avro レコードの消費¶
Avro コンシューマーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- トピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.ConsumerAvroExample" \ -Dexec.args="$HOME/.confluent/java.config test2"
コンシューマー Avro コード を表示します。
Avro Kafka Streams¶
Avro Kafka Streams アプリケーションを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- 前の手順で使用したものと同じトピック名
mvn exec:java -Dexec.mainClass="io.confluent.examples.clients.cloud.StreamsAvroExample" \ -Dexec.args="$HOME/.confluent/java.config test2"
Kafka Streams Avro コード を表示します。
Confluent Cloud Schema Registry によるスキーマ進化¶
Confluent Cloud Schema Registry に登録されたスキーマサブジェクトを表示します。次の出力の
<SR API KEY>、<SR API SECRET>、および<SR ENDPOINT>に値を代入します。curl -u <SR API KEY>:<SR API SECRET> https://<SR ENDPOINT>/subjects
サブジェクト
test2-valueが存在することを確認します。["test2-value"]
サブジェクト test2-value のスキーマ情報を表示します。次の出力の
<SR API KEY>、<SR API SECRET>、および<SR ENDPOINT>に値を代入します。curl -u <SR API KEY>:<SR API SECRET> https://<SR ENDPOINT>/subjects/test2-value/versions/1
サブジェクト
test2-valueのスキーマ情報を確認します。{"subject":"test2-value","version":1,"id":100001,"schema":"{\"name\":\"io.confluent.examples.clients.cloud.DataRecordAvro\",\"type\":\"record\",\"fields\":[{\"name\":\"count\",\"type\":\"long\"}]}"}スキーマを進化させるために、Confluent Cloud Schema Registry で、スキーマの新しいバージョンと古いバージョンの間の スキーマの互換性をテスト することができます。pom.xml では、スキーマレジストリのサブジェクト名が
test2-valueにハードコーディングされます。トピック名test2を使用しなかった場合は、これを変更してください。続けて、DataRecordAvro2a.avsc のローカルスキーマ互換性をテストすると、これは失敗します。DataRecordAvro2b.avsc のローカルスキーマ互換性をテストすると、これは成功します。# DataRecordAvro2a.avsc compatibility test: FAIL mvn schema-registry:test-compatibility "-DschemaRegistryUrl=https://{{ SR_ENDPOINT }}" "-DschemaRegistryBasicAuthUserInfo={{ SR_API_KEY }}:{{ SR_API_SECRET }}" "-DschemaLocal=src/main/resources/avro/io/confluent/examples/clients/cloud/DataRecordAvro2a.avsc" # DataRecordAvro2b.avsc compatibility test: PASS mvn schema-registry:test-compatibility "-DschemaRegistryUrl=https://{{ SR_ENDPOINT }}" "-DschemaRegistryBasicAuthUserInfo={{ SR_API_KEY }}:{{ SR_API_SECRET }}" "-DschemaLocal=src/main/resources/avro/io/confluent/examples/clients/cloud/DataRecordAvro2b.avsc"