このエントリは以下のエントリをベースにしています。
This entry is based on the following one written by Otavio Santana (Software engineer in Tomitribe).
https://dzone.com/articles/using-jakarta-eemicroprofile-to-connect-to-apache
Apache Kafkaは、分散ストリーミングプラットフォームで、以下の機能を備えています。
- レコードストリームをパブリッシュおよびサブスクライブ
- レコードストリームを耐障害性のある永続的な方法で格納
- ストリーム発生時にストリーム処理
Apache Kafkaは、Javaの世界で成功事例があり、このエントリではJakarta EE/MicroProfileの世界でApache Kafkaを活用する方法をご紹介します。
Apache Kafka Core Concepts
Kafkaは1個以上のサーバー上でクラスタとして動作します。このサーバー（群）は複数のデータセンターにまたがることができます。Kafkaクラスタはトピック（Topic）と呼ばれるカテゴリにレコードストリームを格納し、各レコードはKey、Value、タイムスタンプで構成されています。
ドキュメントによると、Kafkaには主要なAPIが4個あります。
Apache Kafka
https://kafka.apache.org/intro
- Producer API
- アプリケーションが1個以上のKafkaのトピックに対してレコードストリームをパブリッシュするためのAPI
- Consumer API
- アプリケーションが1個以上のトピックをサブスクライブし、トピックに作成されたレコードストリームを処理するためのAPI
- Streams API
- 入力ストリームを1個以上のトピックから消費したり、1個以上のトピックに対して出力ストリームを生成したり、効率的に入力ストリームを出力ストリームに変換したり、という、ストリーム・プロセッサとしてアプリケーションが振る舞うためのAPI
- Connector API
- 既存のアプリケーションやデータシステムとKafkaのトピックを接続する、再利用可能なプロデューサーやコンシューマーを作成、実行するためのAPI
Install Apache Kafka
公式ドキュメントにはApache Kafkaをはじめるにあたってのよいエントリがあり、Zookeeperを使ったインストール方法が紹介されています。かいつまんでご紹介すると、KafkaはZookeeperを使ってクラスタメンバーの構成やトピックの構成などを行います。
Quickstart
https://kafka.apache.org/quickstart
2.1系をダウンロードします。
Downloads
https://kafka.apache.org/downloads
（訳注）
2019/05/01現在、最新版は2.2.0ですが、原文では2.1.0 (Scala 2.11)をインストールするように指示しています。
tar-ballを展開します。
tar -xzf kafka_2.11-2.1.0.tgz
cd kafka_2.11-2.1.0
`まず、Zookeeperインスタンスを起動します。
bin/zookeeper-server-start.sh config/zookeeper.properties
そして、Apache Kafkaを起動します。
bin/kafka-server-start.sh config/server.properties
Using Docker
Dockerを使用するという手もあります。ZookeeperとApache Kafkaの2つのDockerイメージが必要なので、このチュートリアルではdocker-composeを使用します。以下の手順に従ってください。
Install Docker
https://docs.docker.com/install/
Install Docker Compose
https://docs.docker.com/compose/install/
docker-compose.ymlを作成し、以下のように構成します。
version: '3.2'
services:
zookeeper:
image: "confluent/zookeeper"
networks:
- microprofile
ports:
- 2181:2181
kafka:
image: "confluent/kafka"
networks:
- microprofile
ports:
- 9092:9092
environment:
- KAFKA_ADVERTISED_HOST_NAME=kafka
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_ADVERTISED_PORT=9092
depends_on:
- zookeeper
networks:
microprofile:
以下のコマンドを実行します。
docker-compose -f docker-compose.yml up -d
localhostで接続できるよう、kafkaをホストのlocalhostとして定義するため、以下の値を /etc/hosts に追加します。
127.0.0.1 localhost kafka
Application With Eclipse MicroProfile
Eclipse MicroProfileは、複数のランタイム間での可搬性があるマイクロサービスアプリケーションプラットフォームを定義するために構成されたオープンソースプロジェクトです。Eclipse MicroProfileには、マイクロサービスとクラウドの時代にエンタープライズ・アプリケーションをより簡単に作成するためのプランがあります。Apache Kafkaと接続するために、Reactive Messageが使えるEclipse MicroProfile Reactiveがあります。
How to Write Reactive Applications with MicroProfile
https://www.eclipse.org/community/eclipse_newsletter/2018/september/reactive_mp.php
CDI 2.0とJava SEを使い、サンプルアプリケーションでApache Kafkaからの送受信メッセージをスムーズに作成しますが、その前に、このプロジェクトではMavenを使用するため、まずは依存関係を定義する必要があります。このプロジェクトは、MicroProfile Configurationの実装、CDI 2.0の実装、およびMicroProfile Reactiveという依存関係を有しています。
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<weld-core.version>3.1.0.Final</weld-core.version>
<slf4j-api.version>1.7.26</slf4j-api.version>
</properties>
<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>${slf4j-api.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>io.smallrye</groupId>
<artifactId>smallrye-config</artifactId>
<version>1.3.5</version>
</dependency>
<dependency>
<groupId>io.smallrye.reactive</groupId>
<artifactId>smallrye-reactive-messaging-provider</artifactId>
<version>0.0.4</version>
</dependency>
<dependency>
<groupId>io.smallrye.reactive</groupId>
<artifactId>smallrye-reactive-messaging-kafka</artifactId>
<version>0.0.4</version>
<exclusions>
<exclusion>
<!-- this avoid having the logging working, to exclude it -->
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.smallrye.reactive</groupId>
<artifactId>smallrye-reactive-streams-operators</artifactId>
<version>0.4.1</version>
</dependency>
<dependency>
<groupId>org.jboss.weld.se</groupId>
<artifactId>weld-se-core</artifactId>
<version>${weld-core.version}</version>
</dependency>
<dependency>
<groupId>javax.enterprise</groupId>
<artifactId>cdi-api</artifactId>
<version>2.0.SP1</version>
</dependency>
</dependencies>
続いて、 src/main/resources/META-INFに配置する構成ファイルを作成します。
1個目はCDIを有効にするためのbean.xmlファイルです。
<beans xmlns="http://xmlns.jcp.org/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee
http://xmlns.jcp.org/xml/ns/javaee/beans_1_1.xsd"
bean-discovery-mode="all">
</beans>
もう一つはmicroprofile-config.propertiesで、レコードのシリアライズ・デシリアライズの実装など、Apache Kafkaと接続するEclipse MicroProfileへのセットアップ構成をkey-value形式で記述します。
# Kafka Sink
smallrye.messaging.sink.data.type=io.smallrye.reactive.messaging.kafka.Kafka
smallrye.messaging.sink.data.bootstrap.servers=localhost:9092
smallrye.messaging.sink.data.key.serializer=org.apache.kafka.common.serialization.StringSerializer
smallrye.messaging.sink.data.value.serializer=org.apache.kafka.common.serialization.StringSerializer
smallrye.messaging.sink.data.acks=1
## Kafka Source
smallrye.messaging.source.kafka.type=io.smallrye.reactive.messaging.kafka.Kafka
smallrye.messaging.source.kafka.bootstrap.servers=localhost:9092
smallrye.messaging.source.kafka.topic=kafka
smallrye.messaging.source.kafka.group.id=demo
smallrye.messaging.source.kafka.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
smallrye.messaging.source.kafka.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
すべての設定が完了したら、最後にApache Kafkaからメッセージを送受信するクラスを作成します。APIでは Incoming と Outgoing という注釈が使えます。 Sender クラスにはBlockingQueueがあり、キューにテキストがあるときにメッセージを送信します。
Interface BlockingQueue
https://docs.oracle.com/javase/jp/8/docs/api/java/util/concurrent/BlockingQueue.html
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html
@ApplicationScoped
public class Receiver {
private static final Logger LOGGER = Logger.getLogger(Receiver.class.getName());
@Incoming("kafka")
public CompletionStage<Void> consume(KafkaMessage<String, String> message) {
String payload = message.getPayload();
String key = message.getKey();
MessageHeaders headers = message.getHeaders();
Integer partition = message.getPartition();
Long timestamp = message.getTimestamp();
LOGGER.info("received: " + payload + " from topic " + message.getTopic());
return message.ack();
}
}
@ApplicationScoped
public class Sender {
private static final Logger LOGGER = Logger.getLogger(Sender.class.getName());
private BlockingQueue<String> messages = new LinkedBlockingQueue<>();
public void add(String message) {
messages.add(message);
}
@Outgoing("data")
public CompletionStage<KafkaMessage<String, String>> send() {
return CompletableFuture.supplyAsync(() -> {
try {
String message = messages.take();
LOGGER.info("Sending message to kafka with the message: " + message);
return KafkaMessage.of("kafka", "key", message);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
});
}
}
最後に、CDIコンテナを起動するクラスを作成します。完了したら、Kafkaにメッセージを送信してみましょう。結果はログで確認できます。
public class App {
public static void main(String[] args) {
SeContainer container = SeContainerInitializer.newInstance().initialize();
Sender sender = container.select(Sender.class).get();
sender.add("Hello world");
sender.add("Otávio");
sender.add("Poliana");
sender.add("Clement");
}
}
Apache Kafkaの可能性と、このプロジェクトがBig-Dataプレーヤにとってアクセシブルになった理由がわかりました。この簡単な例から、Eclipse MicroProfileとの統合に不安がないことがわかります。このサンプルコードを手助けしてくれたClement Escoffierに感謝します。サンプルコードはGitHubにUp済みです。
kafka-cdi-microprofile-hello-world – A hello world using Apache Kafka and Eclipse Microprofile
https://github.com/soujava/kafka-cdi-microprofile-hello-world
