CLIを通じてデータをソースするためのKafkaプロデューサーのセットアップ方法

著者はWrite for DOnationsプログラムの一環として、寄付を受け取るためにオープンソースイニシアティブを選択しました。

紹介

Apache Kafkaは、Kafkaクラスタへの基本的なテキストメッセージの生成と消費のためのシェルスクリプトを提供します。これらは探索や実験には便利ですが、実際のアプリケーションはプログラム的にKafkaにアクセスします。この目的のために、Kafkaは広く使用されているプログラミング言語や環境のための多くのクライアントライブラリを提供しています。

このチュートリアルでは、Kafkaトピックにデータを生成するJavaプログラムを作成します。Apache Mavenを使用してJavaプロジェクトを作成し、Javaプロジェクトのビルドおよびパッケージングツールとして、Kafkaクライアントライブラリを依存関係として追加します。次に、メッセージを生成し、それらに関するクラスタ内メタデータを取得することでKafkaクライアントを活用するクラスを実装します。

前提条件

このチュートリアルを完了するためには、次のものが必要です:

少なくとも4GBのRAMと2つのCPUを持つマシン。Ubuntuサーバーの場合は、初期サーバーセットアップガイドに従ってください。
Droplet またはローカルマシンに Java Development Kit (JDK) 8 以上がインストールされていること。Ubuntu に Java をインストールする方法については、「Apt を使用して Ubuntu に Java をインストールする方法」チュートリアルを参照してください。
Droplet またはローカルマシンに Apache Kafka がインストールされ、設定されている必要があります。設定手順については、「Kafka 入門」チュートリアルを参照してください。
Javaプロジェクトの標準ディレクトリレイアウトに関する理解。詳細については、公式Mavenドキュメントの「標準ディレクトリレイアウトの紹介」トピックを参照してください。

ステップ1 - Mavenプロジェクトの作成

このステップでは、Apache Mavenをインストールし、それを使ってKafkaとインターフェースするためのプロジェクトを生成します。

Ubuntu では Maven が公式リポジトリに容易に利用できます。MacOS システムに Maven をインストールするには、Mac OS に Maven をインストールする方法のチュートリアルを参照してください。

まず、次のコマンドを実行して既存のパッケージリストを更新してください:

sudo apt update

次のコマンドを実行してインストールします:

sudo apt install maven

インストールされているかどうかは、そのバージョン番号を読み上げて確認してください:

mvn --version

出力は、あなたのJavaバージョンとプラットフォームによって以下のようになります:

Apache Maven 3.6.3
Maven home: /usr/share/maven
Java version: 11.0.22, vendor: Ubuntu, runtime: /usr/lib/jvm/java-11-openjdk-amd64
Default locale: en, platform encoding: UTF-8
OS name: "linux", version: "5.15.0-100-generic", arch: "amd64", family: "unix"

次に、Kafkaで作業するためのJavaプロジェクトを保存するディレクトリを作成します。

mkdir ~/kafka-projects

新しく作成したディレクトリに移動します:

cd ~/kafka-projects

次に、次のコマンドを実行して空のJavaプロジェクトを生成します:

mvn archetype:generate \
-DgroupId=com.dokafka \
-DartifactId=dokafka \
-DarchetypeArtifactId=maven-archetype-quickstart \
-DarchetypeVersion=1.4 \
-DinteractiveMode=false

ここでは、Mavenにdokafkaという新しいプロジェクトを生成するよう指示します。グループIDはcom.dokafkaです。グループIDはMavenエコシステム全体でこのプロジェクトを一意に識別します。プロジェクトはmaven-archetype-quickstart アーキタイプに従って生成されます。これがMavenがテンプレートを呼ぶ方法です。

Mavenを初めて実行する場合、特に多くの出力が表示されるでしょう。出力の最後は次のようになります:

...
INFO] Generating project in Batch mode
[INFO] ----------------------------------------------------------------------------
[INFO] Using following parameters for creating project from Archetype: maven-archetype-quickstart:1.4
[INFO] ----------------------------------------------------------------------------
[INFO] Parameter: groupId, Value: com.dokafka
[INFO] Parameter: artifactId, Value: dokafka
[INFO] Parameter: version, Value: 1.0-SNAPSHOT
[INFO] Parameter: package, Value: com.dokafka
[INFO] Parameter: packageInPathFormat, Value: com/dokafka
[INFO] Parameter: package, Value: com.dokafka
[INFO] Parameter: groupId, Value: com.dokafka
[INFO] Parameter: artifactId, Value: dokafka
[INFO] Parameter: version, Value: 1.0-SNAPSHOT
[INFO] Project created from Archetype in dir: /root/kafka-projects/dokafka
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  3.537 s
[INFO] Finished at: 2024-03-10T10:43:07Z
[INFO] ------------------------------------------------------------------------

Mavenはその中央リポジトリから必要なJavaパッケージをダウンロードし、maven-archetype-quickstartテンプレートを使用してdokafkaプロジェクトを作成しました。

次のコマンドを実行してプロジェクトディレクトリに移動します:

cd dokafka

プロジェクトの構造はこのようになります:

├── pom.xml
└── src
    ├── main
    │   └── java
    │       └── com
    │           └── dokafka
    │               └── App.java
    └── test
        └── java
            └── com
                └── dokafka
                    └── AppTest.java

前提条件の一部として、ここに示されている標準的なMavenプロジェクト構造について学びました。src/main/javaディレクトリにはプロジェクトのソースコードが含まれ、src/test/javaにはテストソースが含まれ、プロジェクトのルートにあるpom.xmlはMavenの主要な設定ファイルです。

プロジェクトには1つのソースファイル、App.javaしか含まれていません。Mavenが生成した内容を見るために、それを表示してください。

cat src/main/java/com/dokafka/App.java

出力は次の通りです:

package com.dokafka;

/**
 * Hello world!
 *
 */
public class App
{
    public static void main( String[] args )
    {
        System.out.println( "Hello World!" );
    }
}

このコードを実行するには、まず以下のコマンドを実行してプロジェクトをビルドする必要があります:

mvn package

Mavenはコードをコンパイルして、それを実行のためのJARファイルにパッケージします。出力の最後は、完了を示すように次のようになります:

...
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  9.830 s
[INFO] Finished at: 2024-03-10T10:41:24Z
[INFO] ------------------------------------------------------------------------

Mavenは生成されたJARファイルをtargetディレクトリに配置しました。あなたが今構築したAppクラスを実行するには、次のコマンドを実行し、クラスの完全な識別子を渡してください:

java -cp target/dokafka-1.0-SNAPSHOT.jar com.dokafka.App

出力は次の通りです:

Hello World!

あなたはMavenをインストールし、空のJavaプロジェクトを作成しました。次のステップでKafkaに必要な依存関係を追加します。

ステップ2 - Maven依存関係の追加

これから、プロジェクトにJava Kafkaクライアントを追加し、ログ用の他の依存関係も追加します。また、パッケージング中にそれらの依存関係を含めるようにMavenを設定します。

まず、kafka-clients依存関係を追加します。ブラウザでJavaクライアントのMavenリポジトリページに移動し、利用可能な最新バージョンを選択して、Maven用の提供されたXMLスニペットをコピーします。執筆時点で、Javaクライアントライブラリの最新バージョンは3.7.0でした。

依存関係はプロジェクトのルートにあるpom.xmlに追加されます。編集のために開いてください:

nano pom.xml

<dependencies> セクションを見つけて、依存関係の定義を追加します:

...
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
      <scope>test</scope>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.7.0</version>
    </dependency>
...
  </dependencies>

これにより、Kafkaクライアントライブラリがプロジェクトで利用可能になります。ただし、ライブラリ自体には手動で追加する必要がある2つの依存関係があります。これらは、メッセージのログ記録のために使用されるSLF4Jライブラリに由来しており、さまざまなログライブラリをサポートし、開発者がログメッセージを処理する方法に柔軟性を持たせます。追加する必要がある2つの依存関係は次のとおりです:

slf4j-api、それ自体がライブラリであり
slf4j-simpleはログを処理し、それを端末に出力します。

依存関係を定義したら、最終的にビルドされたJARと一緒にそれらを利用できるようにする必要があります。pom.xmlの<build>セクションを見つけ、ハイライトされた行を追加してください:

...
  <build>
    <pluginManagement>
      <plugins>
        ...
      </plugins>
    </pluginManagement>
    <plugins>
        <plugin>
            <artifactId>maven-dependency-plugin</artifactId>
            <executions>
                <execution>
                <phase>package</phase>
                    <goals>
                        <goal>copy-dependencies</goal>
                    </goals>
                    <configuration>
                        <outputDirectory>${project.build.directory}/lib</outputDirectory>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
  </build>
...

ここでは、maven-dependency-plugin プラグインを設定して、パッケージング時にすべての依存関係をコピーします。このプロジェクトの設定では、依存関係の JAR ファイルは target/lib の下に配置されます。既存の <plugins> セクションを <pluginManagement> の下で変更しないでください。

終わったら、ファイルを保存して閉じてください。

プロジェクトをビルドして、すべてが正しく設定されていることを確認してください。

mvn package

出力の終わりはこのような感じになります:

...
[INFO] --- maven-jar-plugin:3.0.2:jar (default-jar) @ dokafka ---
[INFO] Building jar: /root/kafka-projects/dokafka/target/dokafka-1.0-SNAPSHOT.jar
[INFO]
[INFO] --- maven-dependency-plugin:2.8:copy-dependencies (default) @ dokafka ---
[INFO] Copying junit-4.11.jar to /root/kafka-projects/dokafka/target/lib/junit-4.11.jar
[INFO] Copying slf4j-simple-2.0.12.jar to /root/kafka-projects/dokafka/target/lib/slf4j-simple-2.0.12.jar
[INFO] Copying snappy-java-1.1.10.5.jar to /root/kafka-projects/dokafka/target/lib/snappy-java-1.1.10.5.jar
[INFO] Copying zstd-jni-1.5.5-6.jar to /root/kafka-projects/dokafka/target/lib/zstd-jni-1.5.5-6.jar
[INFO] Copying hamcrest-core-1.3.jar to /root/kafka-projects/dokafka/target/lib/hamcrest-core-1.3.jar
[INFO] Copying lz4-java-1.8.0.jar to /root/kafka-projects/dokafka/target/lib/lz4-java-1.8.0.jar
[INFO] Copying slf4j-api-2.0.12.jar to /root/kafka-projects/dokafka/target/lib/slf4j-api-2.0.12.jar
[INFO] Copying kafka-clients-3.7.0.jar to /root/kafka-projects/dokafka/target/lib/kafka-clients-3.7.0.jar
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  5.205 s
[INFO] Finished at: 2024-03-12T06:36:34Z
[INFO] ------------------------------------------------------------------------

target/libの下にあるファイルをリストアップして、依存関係が実際にコピーされているか確認してみることができます:

hamcrest-core-1.3.jar  kafka-clients-3.7.0.jar  slf4j-api-2.0.12.jar     snappy-java-1.1.10.5.jar
junit-4.11.jar         lz4-java-1.8.0.jar       slf4j-simple-2.0.12.jar  zstd-jni-1.5.5-6.jar

Mavenプロジェクトに必要な依存関係を追加しました。これからKafkaへの接続とメッセージのプログラムによる生成に入ります。

ステップ3 - JavaでのKafkaプロデューサーの作成

このステップでは、JavaでKafkaプロデューサーを設定し、トピックにメッセージを書き込みます。

プロジェクトの構造に従い、ソースコードはsrc/main/java/com/dokafkaの下に保存されています。このチュートリアルの残りの部分ではApp.javaは必要ないので、次のコマンドを実行して削除してください:

rm src/main/java/com/dokafka/App.java

プロデューサーコードは ProducerDemo というクラスに保存します。編集のために関連ファイルを作成して開いてください:

nano src/main/java/com/dokafka/ProducerDemo.java

次の行を追加してください:

package com.dokafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerDemo {
  private static final Logger log = LoggerFactory.getLogger(ProducerDemo.class);

  public static void main(String[] args) {
    String bootstrapServers = "localhost:9092";
    String topicName = "java_demo";

    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

    ProducerRecord<String, String> producerRecord =
      new ProducerRecord<>(topicName, "Hello World!");

    producer.send(producerRecord);

    producer.flush();
    producer.close();
  }
}

まず、ProducerDemoクラスを定義し、使用するクラスをインポートしてLoggerを作成します。mainメソッドでは、まずKafkaクラスターのアドレス（bootstrapServers）とメッセージを生成するトピック名（topicName）を宣言します。

次に、Kafkaプロデューサーを操作するための設定を保持するキーと値の辞書に似たPropertiesオブジェクトをインスタンス化します。BOOTSTRAP_SERVERS_CONFIGプロパティをKafkaクラスタのアドレスで設定します。また、KEY_SERIALIZER_CLASS_CONFIGとVALUE_SERIALIZER_CLASS_CONFIGのエントリをStringSerializer.class.getName()に設定します。

これらのプロパティは、生成されたメッセージのキーと値を処理するために使用されるシリアライザーを指定します。シリアライザーは、入力を受け取り、ネットワークを通じて伝送するために準備されたバイトの配列を出力するクラスです。デシリアライザーは逆のことを行い、バイトのストリームから元のオブジェクトを再構築します。ここでは、キーと値の両方が組み込みのStringSerializerを使用して文字列としてシリアライズされます。

次に、KafkaProducerを宣言してインスタンス化します。

KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

そのプロデューサーは、設定用のプロパティとともに、String型のキーと値を受け入れます。

トピックにメッセージを送信するには、KafkaProducerがProducerRecordを受け入れます。これはトピックの名前とメッセージ自体（Hello World!）を使用してインスタンス化します。プロデューサー自体は特定のトピックに接続されていないことに注意してください。

メッセージを送信した後、プロデューサーをフラッシュして閉じます。producer.send() の呼び出しは非同期であり、メッセージが別のスレッドで送信されている間、制御フローはmain メソッドに戻ります。この例のプログラムはその後に終了したいので、プロデューサーに残っているすべてのメッセージをフラッシュして送信させます。そして、close() メソッドを呼び出して、プロデューサーが破棄されることをKafkaに通知します。

次に、ProducerDemoのビルドと実行を処理するスクリプトを作成します。それをrun-producer.shという名前のファイルに保存します。ファイルを作成して編集のために開いてください:

nano run-producer.sh

次の行を追加してください:

#!/bin/bash
mvn clean
mvn package
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.ProducerDemo

作業が終わったら、ファイルを保存して閉じてください。ハイライトされた部分は依存関係の位置を示しています。

次に、それを実行可能としてマークします:

chmod +x run-producer.sh

最後に、これを実行してHello World!メッセージを作成してみてください:

./run-producer.sh

出力は長くなり、その終わりはこのようになります:

...
[main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector - initializing Kafka metrics collector
[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-1] Instantiated an idempotent producer.
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 2ae524ed625438c5
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1710176327832
[kafka-producer-network-thread | producer-1] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=producer-1] Error while fetching metadata with correlation id 1 : {java_demo=UNKNOWN_TOPIC_OR_PARTITION}
[kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.Metadata - [Producer clientId=producer-1] Cluster ID: Z-4Gf_p6T2ygtb6461nKRA
[kafka-producer-network-thread | producer-1] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=producer-1] ProducerId set to 12 with epoch 0
[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
[main] INFO org.apache.kafka.common.metrics.Metrics - Metrics scheduler closed
[main] INFO org.apache.kafka.common.metrics.Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter
[main] INFO org.apache.kafka.common.metrics.Metrics - Closing reporter org.apache.kafka.common.telemetry.internals.ClientTelemetryReporter
[main] INFO org.apache.kafka.common.metrics.Metrics - Metrics reporters closed
[main] INFO org.apache.kafka.common.utils.AppInfoParser - App info kafka.producer for producer-1 unregistered

KafkaProducerが正常に作成されたことをログに記録し、その後に登録解除されました。メッセージは現在java_demoトピックに書き込まれ、kafka-console-consumer.shスクリプトを使用して取得できます。

別のシェルで、Kafkaインストールのディレクトリに移動し、次のコマンドを実行してトピックを読み取ります:

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

出力は次の通りです:

Hello World!

終了するには CTRL+C を押すことができます。

このステップでは、プログラムでjava_demoトピックにメッセージを生成し、Kafka提供のbashスクリプトを使用してそれを読み取る方法を説明しました。次に、メッセージが正常に送信されたときにKafkaが返す情報を利用する方法を学びます。

ステップ4 - コールバックを使用してメタデータを取得する

KafkaProducerのsend()メソッドはコールバックを受け入れ、レコードが受信されたときなどの発生したイベントに対してアクションを実行できます。これは、クラスターがレコードをどのように処理したかに関する情報を取得するのに便利です。

send() コールをコールバックで拡張するには、まず ProducerDemo を編集のために開きます:

nano src/main/java/com/dokafka/ProducerDemo.java

コードを次のように修正してください:

...
    producer.send(producerRecord, new Callback() {
      public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
          log.error("An error occurred!", e);
          return;
        }

        log.info(String.format("Timestamp: %s, partition: %s; offset: %s",
          recordMetadata.timestamp(),
          recordMetadata.partition(),
          recordMetadata.offset()));
      }
    });
...

今、send()メソッドにCallbackインターフェースの実装を渡し、RecordMetadataとオプションでExceptionを受け取るonCompletion()を実装します。次に、エラーが発生した場合は、それをログに記録します。そうでない場合は、レコードのタイムスタンプ、パーティション番号、そして現在クラスターにあるオフセットをログに記録します。このようにメッセージを送信することは非同期であるため、レコードがクラスターに受け入れられたときにあなたのコードが呼び出されます。あなたはそれが発生するのを明示的に待つ必要はありません。

作業が完了したら、ファイルを保存して閉じ、次にプロデューサーを実行します。

./run-producer.sh

出力の最後に新しいメッセージがあります:

...
[kafka-producer-network-thread | producer-1] INFO com.dokafka.ProducerDemo - Timestamp: 1710181715303, partition: 0; offset: 2
...

先ほど生成されたメッセージはクラスタによって受け入れられ、パーティション0に保存されました。

再度実行すると、オフセットが1つ大きくなっていることに気付くでしょう。これはパーティション内のメッセージの位置を示しています:

[kafka-producer-network-thread | producer-1] INFO com.dokafka.ProducerDemo - Timestamp: 1710181831814, partition: 0; offset: 3

結論

この記事では、Mavenを使用してJavaプロジェクトを作成し、Kafkaとのインターフェース用の依存関係を備えています。その後、Kafkaクラスターにメッセージを生産するクラスを開発し、送信されたレコードに関する情報を取得するために拡張しました。

CLIを通じてデータをソースするためのKafkaプロデューサーのセットアップ方法

紹介

前提条件

ステップ1 - Mavenプロジェクトの作成

ステップ2 - Maven依存関係の追加

ステップ3 - JavaでのKafkaプロデューサーの作成

ステップ4 - コールバックを使用してメタデータを取得する

結論

関連記事: