Open LibertyとMicroProfile Reactive MessagingでKafkaのイベントを読み取る

Javaのアプリケーションサーバである [Open Liberty](https://openliberty.io/) では、MicroProfile Reactive Messagingと呼ばれる仕様のAPIを使うことでイベント処理することが可能です。

この記事ではMicroProfile Reactive MessagingでApache Kafkaのイベントを読み取って処理するConsumerを実装してみます。イベントの送信については別の記事で紹介する予定です。

![スクリーンショット 2022-11-06 21.54.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137947/9c45c24e-f10a-88e9-365f-d921de83f121.png)


## 本記事で使用するコンポーネント

本記事では以下のコンポーネント・バージョンを使用します。
今後のバージョンアップで動作しない可能性がありますので、本番投入する際には事前に手元で動作確認することをお勧めします。

|コンポーネント名|バージョン|
|:--|:--|
|JDK|OpenJDK 17.0.4.1 (IBM Semeru Runtime Open Edition)|
|Open Liberty|22.0.0.11|


## Open Liberty Javaアプリケーションの雛形生成

まずはConsumerを実装するOpen Libertyアプリケーションを作成します。
アプリケーションの雛形は、Open Libertyのサイトにある雛形生成機能を使用すると簡単です。

Starterアプリケーションの[雛形生成ページ](https://openliberty.io/start/)から、下図の通り値を入力して「Generate Project」ボタンを押下します。
「Group」と「Artifact」は任意の値でOKです。

![スクリーンショット 2022-11-06 21.56.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/137947/52e641e4-7f9f-3432-5596-23644799aa0b.png)

:::note warn
この記事を書いている時点では、Java EE/Jakarta EE Versionに「9.1」を指定するとConsumerとして動作しないので注意してください。
:::


## 依存ライブラリのインストール

MicroProfile Reactive MessagingのAPIからKafkaのイベントを扱うために、依存ライブラリをインストールしましょう。
先ほど生成した雛形プロジェクトではライブラリの管理方法にMavenを指定していました。Mavenの設定ファイル `pom.xml` の `dependencies` 要素に以下の依存ライブラリを指定します。インストールはEclipseやIntelliJ等のIDE側で自動的に実行されます。

```xml
    <!-- 必要なものだけ記載 -->
    <dependencies>
        <dependency>
            <groupId>org.eclipse.microprofile.reactive.messaging</groupId>
            <artifactId>microprofile-reactive-messaging-api</artifactId>
            <version>2.0.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-jdk14</artifactId>
            <version>1.7.36</version> <!-- 2.x.y はNG -->
        </dependency>
    </dependencies>
```

末尾に追加している `slf4j-jdk14` は、Open LibertyでMicroProfile Reactive MessagingとKafkaクライアントライブラリのログを出力するために使用します。

:::note warn
slf4j-jdk14は本記事執筆時点 (2022/11) で表のバージョンよりも高い 2.x.y というバージョンが使用可能です。しかしバージョン2系を指定するとアプリケーションが正常に起動しない場合がありますので上記のバージョンを指定する方が安全です。
:::


## Featureの指定

続いてOpen LibertyにMicroProfile Reactive Messagingを使用するために、Featureを指定して読み込ませます。

Open Libertyで使用されるモジュールは「Feature」という形で細分化されています。必要なFeatureのみ使用することで、アプリケーションのサイズを小さく保つことが可能です。

今回使用するMicroProfile Reactive Messagingも専用のFeatureが用意されています。
Open Libertyの設定ファイル `server.xml` に `mpReactiveMessaging-1.0` Featureを指定します。

ついでに `<logging>` 要素でログ出力も有効化しておきます。

```xml
<!-- src/main/liberty/config/server.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<server description="My Liberty Server">

    <!-- Enable features -->
    <featureManager>
        <feature>mpReactiveMessaging-1.0</feature>
    </featureManager>

    <!-- ...中略... -->

    <!-- Logging -->
    <logging traceSpecification="REACTIVEMESSAGE=all:org.apache.kafka.*=all"/>

</server>

```

## Kafkaの接続情報を定義

ライブラリとFeatureをインストールしたら、続いてApache Kafkaとの接続情報を定義します。接続情報は MicroProfile Config の設定ファイル `microprofile-config.properties` に定義します。

`<プロジェクトルート>/src/main/resources/META-INF/microprofile-config.properties` に以下の通り設定値を定義してください。

```properties
mp.messaging.incoming.purchases.connector=liberty-kafka
mp.messaging.incoming.purchases.bootstrap.servers=localhost:9092
mp.messaging.incoming.purchases.group.id=liberty-sample-consumer
mp.messaging.incoming.purchases.key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.purchases.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.purchases.auto.offset.reset=earliest
```

設定値は `mp.messaging.<incoming|outgoing>.<チャンネル名>.<プロパティ名>=<設定値>` という書式で定義します。

ドット区切り3つ目の要素は、イベントを読み取る場合は `incoming` を指定してください。逆にイベントを送る場合は `outgoing` を指定します。
`<プロパティ名>` はApache Kafkaの設定値名に準拠するため、さらに詳細に設定したい場合はApache Kafkaのドキュメントを参照してください。

ここではKafka Broker `localhost:9092` に接続し、イベントのKeyとValueはそれぞれ `String` 型であることを定義しています。`connector` プロパティのみ特殊なプロパティであり、Kafkaと接続する際には固定値 `liberty-kafka` を指定します。

参考: [Apache Kafka Consumer Config](https://kafka.apache.org/documentation/#consumerconfigs)


## イベント操作アプリケーションの実装

最後にKafkaからイベントを読み取るアプリケーションを実装します。
ここではイベントのValueに指定された購買商品名をログ出力する想定で実装を行います。

実装サンプルは以下の通りです。
たった20行程度の簡単なサンプルですが、これだけでイベントを処理可能です。

```java
// PurchaseResource.java
package com.example.resource;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import javax.enterprise.context.ApplicationScoped;
import java.util.logging.Logger;

@ApplicationScoped
public class PurchaseResource {
    private static final Logger logger = Logger.getLogger(PurchaseResource.class.getName());

    @Incoming("purchases")
    public void consumePurchases(String item) {
        final String itemMessage = String.format("Consumed item purchase: item name = %s", item);
        logger.info(itemMessage);
    }
}
```

最も重要なパートは `@Incoming("<チャンネル名>")` です。
`@Incoming` はMicroProfile Reactive Messagingで提供された、イベントを読み取るアノテーションです。アノテーションに指定したチャンネル名は `microprofile-config.properties` で定義したチャンネル名に対応する設定値と紐づけられます。

:::note info
このサンプルコードでは `@Incoming` を付加したメソッドの引数にイベントのValueのみ指定しています。イベントのKeyやTopic名を取得したい場合は、以下の3つの実装を追加します。

(1) 引数に `Message` 型を指定する
(2) `ConsumerRecord` 型で引数の値をキャストする、
(3) メソッドが返却する型に `CompletionStage` を指定する

詳しいサンプルは以下のBlog記事を参照ください。
[Blog - Access specific properties in Kafka messages and set the SameSite attribute on cookies in Open Liberty 20.0.0.3](https://openliberty.io/blog/2020/03/13/access-specific-kafka-properties-samesite-20003.html)
:::


## アプリケーションの稼働確認

前節まででアプリケーションの準備が整ったので、実際にKafkaと接続してイベントを読み取ります。
アプリケーション稼働前にKafka Brokerを動作させ、Topic `purchases` を作成してください。

また、Starterアプリケーションには既にOpen Libertyを起動するMaven Pluginが組み込まれています。以下のコマンドでOpen Libertyを起動します。ビルドされたWARファイルは自動的にOpen Libertyサーバにデプロイされます。

```shell
$ mvn liberty:dev
```

大量のログが出力されますが、`Successfully joined group with generation Generation` というログが出力されていればKafka Brokerに接続してイベントを読み出す準備ができています。

`kafka-console-producer` 等のProducerからイベントを送信してみます。

```shell
$ kafka-console-producer.sh --bootstrap-server localhost:9092 \
    --topic purchases \
    --property "parse.key=true" --property "key.separator=:"

>alice:apple
>bob:banana
>cris:orange
>^C
```

Open Libertyアプリケーション側には、ほぼリアルタイムにイベントのValueがログ出力されています。

```shell
...
[INFO] [2022/11/06 22:36:06:474 JST] 0000003c com.example.resource.PurchaseResource                        I Consumed item purchase: item name = apple
[INFO] [2022/11/06 22:36:08:018 JST] 0000003a com.example.resource.PurchaseResource                        I Consumed item purchase: item name = banana
[INFO] [2022/11/06 22:36:10:404 JST] 0000003d com.example.resource.PurchaseResource                        I Consumed item purchase: item name = orange
```

## 最終系の構成

アプリケーションの最終的なディレクトリ構成は以下の通りです。
`/path/to/project` は任意のプロジェクトルートのパスが入ります。

```shell
$ tree /path/to/project

/path/to/project
├── pom.xml
└── src
    └── main
        ├── java
        │   └── com
        │       └── example
        │           └── resource
        │               └── PurchaseResource.java
        ├── liberty
        │   └── config
        │       └── server.xml
        └── resources
            └── META-INF
                └── microprofile-config.properties
```


## 参考

* [OpenLiberty - Creating reactive Java microservices](https://openliberty.io/guides/microprofile-reactive-messaging.html)
* [Sending and receiving messages between microservices with MicroProfile Reactive Messaging](https://openliberty.io/blog/2019/09/13/microprofile-reactive-messaging.html)