Apache CamelフレームワークのKafkaコンポーネントを使用して簡単なメッセージの送受信（Java DSL編）

# はじめに

Apache kafkaはオープンソースの分散メッセージングシステムで、大量のメッセージを高速にpublish/subscribeすることができます。
ここでは、統合フレームワークであるApache Camelを用いてkafkaとメッセージをやりとりするアプリケーションを作成します。
Apache Camelでkafkaに接続するためにはcamel-kafkaコンポーネントを使用します。

今回の環境では、kafkaはv2.0.0を使用します。環境の構築方法は以下の記事を参照してください。

- [Apacke kafka v2.0.0がリリースされていたのでインストールしてみる](https://qiita.com/mkyz08/items/90d441e86ed692950f34)


また、以前に次の記事で同様のアプリケーションをXML DSLで書いていますが、今回はJava DSLを用いてアプリケーションを作成してみます。また、前回よりも詳細に書いてみたいと思います。

- [Apache CamelフレームワークのKafkaコンポーネントを使用して簡単なメッセージの送受信](https://qiita.com/mkyz08/items/a96035f6bfd3dc82b5e2)


# 作成するアプリケーションの説明

今回作成するアプリケーションは、下図のように1秒ごとに日時のメッセージをkafkaへpublishし、kafkaから同メッセージをsubscribeするものです。


![image.png](https://qiita-image-store.s3.amazonaws.com/0/179607/00b1e492-e879-ad5f-2de6-a5326389d1a5.png)


Apache Camelを利用すれば、このようなkafkaを用いたアプリケーションが以下の数十行のコードだけで実装できます。main関数に全てコードを入れているので、綺麗なコードではありませんが、Camelにより簡単に実装できることが分かってもらえるかと思います。


```java
	public static void main(String[] args) {
		try {
			CamelContext context = new DefaultCamelContext();
			KafkaComponent kafka = new KafkaComponent();
			kafka.setBrokers("kafka1:9092,kafka2:9092,kafka3:9092"); // kafka1, kafka2, kafka3の3サーバをブローカとして登録
			context.addComponent("kafka", kafka);

			context.addRoutes(new RouteBuilder() {
				public void configure() {
					from("timer:trigger?period=1000") // 1000ミリ秒毎に実行
							.routeId("kafka_producer_route")
							.setBody(simple("${date:now:yyyy-MM-dd HH:mm:ss}")) // メッセージのBODYに現在の日時を設定
							.to("kafka:test01"); // トピックtest01にメッセージをpublish

					from("kafka:test01") // トピックtest01からメッセージをsubscribe
							.routeId("kafka_consumer_route")
							.log("body = ${body}"); // メッセージのBODYの内容をログに表示
				}
			});

			context.start();
			Thread.sleep(10000);
			context.stop();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
```

それでは、kafkaのアプリケーションを作成してみましょう。


# kafkaを使用するためのライブラリを追加する

Mavenの場合は、pom.xmlに以下のライブラリを追加します。
camel-kafkaがCamelでKafkaを扱うためのコンポーネントで、${camel.version}には使用しているcamelのバージョンを指定します。
kafka-clientsはKafka用のクライアントライブラリです。

```xml:pom.xml
		<dependency>
			<groupId>org.apache.camel</groupId>
			<artifactId>camel-kafka</artifactId>
			<version>${camel.version}</version>
		</dependency>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>2.0.0</version>
		</dependency>
```

# コンポーネントのURI

camel-kafkaコンポーネントを使用するためのURIは以下のようになります。

```
kafka:topic[?options]
```

コンテキストパス（topic）には対象のトピック名を指定します。


# メッセージをpublish/subscribeするrouteを作成する

kafkaへメッセージをpublishするrouteを作成します。
Java DSLで書いたrouteは以下のようになります。



トピックは自動で作成されますが、トピックの設定を変えたい場合は手動で作成しておきます。
コードは簡単すぎるので特に説明することはありませんね。

```java
				from("timer:trigger?period=1000") // 1000ミリ秒毎に実行
						.routeId("kafka_producer_route")
						.setBody(simple("${date:now:yyyy-MM-dd HH:mm:ss}")) // メッセージのBODYに現在の日時を設定
						.to("kafka:test01"); // トピックtest01にメッセージをpublish
```

次にkafkaへメッセージをsubscribeするrouteを作成します。
Java DSLで書いたrouteは以下のようになります。

```java
				from("kafka:test01") // トピックtest01からメッセージをsubscribe
						.routeId("kafka_consumer_route")
						.log("body = ${body}"); // メッセージのBODYの内容をログに表示
```

これだけではkafkaへ接続できないので、接続するための設定を書きます。
まず最初にkafkaの設定を登録するKafkaConfigurationのインスタンスを生成し、kafkaへ接続するための設定を行います。様々な設定項目があり、後で簡単に説明したいと思います。
以下の例ではいくつかの設定を行っていますが、接続するだけであれば、setBrokersメソッドで接続先のブローカを設定するだけです。それ以外はデフォルトの値が使用されるため設定しなくても構いません。

```java
		KafkaConfiguration kafkaConfig = new KafkaConfiguration();
		kafkaConfig.setBrokers("kafka1:9092,kafka2:9092,kafka3:9092"); // 3台のkafkaブローカへ接続
		kafkaConfig.setGroupId("group1"); // consumerのグループIDを設定
		kafkaConfig.setAutoCommitEnable(true); // 自動コミットをオンに設定
		kafkaConfig.setAutoCommitIntervalMs(5000); // 自動コミットのインターバルを設定
		kafkaConfig.setAutoOffsetReset("earliest"); // コミットされたオフセットのないパーティションのsubscribeした場合の動作
		kafkaConfig.setRequestRequiredAcks("all"); // publishが成功したと判断する条件
		kafkaConfig.setConsumersCount(1); // consumerの数
```

次にKafkaComponentのインスタンスを生成し、先ほど作成したKafkaConfigurationのインスタンスをセットします。

```java
		KafkaComponent kafka = new KafkaComponent();
		kafka.setConfiguration(kafkaConfig);
```

kafkaへ接続するための設定はこれだけです。

これでpublisherとconsumerのrouteの作成とkafkaの接続設定が終わりました。
これらを使用するmain関数などを作成した全体のソースは以下になります。

```java
package example.camelbegginer.kafka;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConfiguration;
import org.apache.camel.impl.DefaultCamelContext;

public class KafkaExample {

	public static void main(String[] args) {
		try {
			CamelContext context = new DefaultCamelContext();
			context.addComponent("kafka", createKafkaComponent());
			context.addRoutes(createProducerRouteBuilder());
			context.addRoutes(createConsumerRouteBuilder());

			context.start();
			Thread.sleep(10000);
			context.stop();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	static KafkaComponent createKafkaComponent() {
		KafkaConfiguration kafkaConfig = new KafkaConfiguration();
		kafkaConfig.setBrokers("kafka1:9092,kafka2:9092,kafka3:9092"); // 3台のkafkaブローカへ接続
		kafkaConfig.setGroupId("group1"); // グループIDを設定
		kafkaConfig.setAutoCommitEnable(true); // 自動コミットをオンに設定
		kafkaConfig.setAutoCommitIntervalMs(5000); // 自動コミットのインターバルを設定
		kafkaConfig.setAutoOffsetReset("earliest"); // コミットされたオフセットのないパーティションの読み出した場合の動作
		kafkaConfig.setRequestRequiredAcks("all"); // publishが成功したと判断する条件
		kafkaConfig.setConsumersCount(1); // consumerの数

		KafkaComponent kafka = new KafkaComponent();
		kafka.setConfiguration(kafkaConfig);

		return kafka;
	}

	static RouteBuilder createProducerRouteBuilder() {
		return new RouteBuilder() {
			public void configure() throws Exception {
				from("timer:trigger?period=1000") // 1000ミリ秒毎に実行
						.routeId("kafka_producer_route")
						.setBody(simple("${date:now:yyyy-MM-dd HH:mm:ss}")) // メッセージのBODYに現在の日時を設定
						.to("kafka:test01"); // トピックtest01にメッセージをpublish
			}
		};
	}

	static RouteBuilder createConsumerRouteBuilder() {
		return new RouteBuilder() {
			public void configure() throws Exception {
				from("kafka:test01") // トピックtest01からメッセージをsubscribe
						.routeId("kafka_consumer_route")
						.log("body = ${body}"); // メッセージのBODYの内容をログに表示
			}
		};
	}
}
```


# 作成したCamelアプリケーションを実行する

このアプリケーションを実行すると、以下のように1秒おきのログが出力されます。

```
[2019-01-31 21:15:38.112], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:38
[2019-01-31 21:15:39.031], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:39
[2019-01-31 21:15:40.034], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:40
[2019-01-31 21:15:41.026], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:41
[2019-01-31 21:15:42.029], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:42
[2019-01-31 21:15:43.024], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:43
[2019-01-31 21:15:44.044], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:44
[2019-01-31 21:15:45.028], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:45
[2019-01-31 21:15:46.032], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, body = 2019-01-31 21:15:46
```

また、以下のコードでヘッダーを出力すると、メッセージのオフセットやパーティションの情報がログに出力されます。

```java
				from("kafka:test01") // トピックtest01からメッセージをsubscribe
						.routeId("kafka_consumer_route")
						.log("body = ${body}") // メッセージのBODYの内容をログに表示
						.log("${headers}"); // ★ここを追加
```

出力されるログは以下のようになります。

```console
2019-02-01 10:10:46.236], [INFO ], kafka_consumer_route, Camel (camel-1) thread #1 - KafkaConsumer[test01], kafka_consumer_route, {breadcrumbId=[B@58a2fc46, kafka.HEADERS=RecordHeaders(headers = [RecordHeader(key = breadcrumbId, value = [73, 68, 45, 109, 107, 121, 45, 80, 67, 45, 49, 53, 52, 56, 57, 56, 51, 52, 52, 51, 56, 55, 51, 45, 48, 45, 49])], isReadOnly = false), kafka.OFFSET=52, kafka.PARTITION=0, kafka.TIMESTAMP=1548983446195, kafka.TOPIC=test01}
```

このように、Apache CamelフレームワークのKafkaコンポーネントを使用して簡単なメッセージの送受信するアプリケーションを作成できました。Camelを使用することで、簡単にkafkaと接続することが理解いただけたでしょうか。


# camel-kafkaコンポーネントの主なプロパティ

最後にcamel-kafkaコンポーネントの主なプロパティについて説明します。
プロパティは以下の表以外にもたくさんあり、詳細は公式ページを参照してください。

- [Kafka Component（旧公式サイト）](http://camel.apache.org/kafka.html)

プロパティはたくさんありますが、いくつかはkafka clientで使用するものと同じですので、違和感なく使えると思います。


| プロパティ名             | producer/consumer | 説明                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | デフォルト値 | 型      |
|--------------------------|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------|---------|
| brokers                  | 共通              | Kafkaブローカーを指定する。フォーマットはhost1:port1で、複数ブローカを指定する場合は、host1:port1,host2:port2とカンマ区切りとする。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |              | String  |
| clientId                 | 共通              | クライアントIDは、クライアントのアプリケーションからの呼び出しをトレースするためのユーザー指定の文字列です。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |              | String  |
| autoCommitEnable         | consumer          | trueに設定すると、consumerが取得されているメッセージのオフセットを定期的に自動でコミットします。自動コミットのインターバルはautoCommitIntervalMsオプションで指定します。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            | TRUE         | Boolean |
| autoCommitIntervalMs     | consumer          | consumerがメッセージのオフセットが自動コミットするインターバル（ミリ秒）を指定する。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | 5000         | Integer |
| autoCommitOnStop         | consumer          | コンシューマが停止する際に、ブローカから最後にsubscribeされたメッセージを明示的に自動コミットするかどうかを指定する。これにはautoCommitEnableオプションがオンになっている必要があります。 設定可能な値は、sync、async、またはnoneです。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | sync         | String  |
| autoOffsetReset          | consumer          | 初期オフセットがない場合、またはオフセットが範囲外の場合は、次の手順を実行します。earliest：オフセットを最も最初のオフセットにリセットします。latest：自動的にオフセットを最新（最後）のオフセットにリセットします。fail：consumerに例外をスローします。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            | latest       | String  |
| consumerRequestTimeoutMs | consumer          | 構成は、クライアントが要求の応答を待つ最大時間を制御します。 タイムアウトが経過する前に応答が受信されない場合、クライアントは必要に応じて要求を再送信するか、または再試行が尽きると要求を失敗させます。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | 40000        | Integer |
| consumersCount           | consumer          | kafkaサーバーに接続するンシューマーの数を指定する。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 | 1            | int     |
| consumerStreams          | consumer          | 同時コンシューマ数。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                | 10           | int     |
| groupId                  | consumer          | このconsumerが属するグループを識別する文字列を指定します。同じグループIDを設定することによって、複数のconsumerがすべて同じコンシューマグループであることを示します。Consumerでは必須のオプションです。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              |              | String  |
| maxPollRecords           | consumer          | poll()の1回の呼び出しで返されるレコードの最大数を指定する。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | 500          | Integer |
| pollTimeoutMs            | consumer          | KafkaConsumerをポーリングするときに使用されるタイムアウト値（ミリ秒）。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             | 5000         | Long    |
| compressionCodec         | producer          | このパラメータを使用すると、producerによって生成されたデータの圧縮コーデックを指定できます。指定できる値はnone、gzip、およびsnappyです。デフォルトでは圧縮されません（none）。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      | none         |         |
| requestRequiredAcks      | producer          | producerからのリクエストが完了したと見なす条件を設定する。メッセージの信頼性のレベルを指定することになります。指定する値は以下の3つがあります。acks=0ですと、リーダーのブローカがダウンするとメッセージをロストしますので、通常はacks=1、またはallを指定します。・acks = 0producerはサーバーからの確認応答を待たずに送信を完了したとみなします。この場合、kafkaがメッセージを受信したことを保証することはできません。各レコードに返されるオフセットは常に-1に設定されます。・acks = 1リーダーがそのメッセージをローカルに書き込むが、他のフォロワーからの完了を待たずに応答します。この場合、メッセージを承認した直後にリーダーにダウンが発生した場合、他のフォロワーがそれをレプリケーションできずにメッセージが失われます。 ・acks = allリーダーが応答を返す前に、他の同期レプリカの書き込みがすべて完了するまで待機します。これにより、少なくとも1つの同期レプリカが存続している限り、メッセージが失われないことが保証されます。 | 1            | String  |
| requestTimeoutMs         | producer          | クライアント（producer）にエラーを返す前に、ブローカーがrequest.required.acksの要件を満たそうまでに待機する時間。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   | 305000       | Integer |
| retries                  | producer          | 0より大きい値を設定すると、クライアントは一時的なエラーで送信に失敗したレコードを再送信（リトライ）します。                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         | 0            | Integer |


# 参考

- 
[Apache CamelフレームワークのKafkaコンポーネントを使用して簡単なメッセージの送受信（昔の記事）](https://qiita.com/mkyz08/items/a96035f6bfd3dc82b5e2)
- [Kafka Component（公式サイト）](https://github.com/apache/camel/blob/master/components/camel-kafka/src/main/docs/kafka-component.adoc)
- [Kafka Component（旧公式サイト）](http://camel.apache.org/kafka.html)
- [Camel Kafka example](https://github.com/apache/camel/tree/master/examples/camel-example-kafka)