MicronautでApache Kafkaにアクセスする

# Micronaut Kafka

MicronautにApache Kafkaのサポートがあるというので、試してみることにしました。

[Kafka Support](https://docs.micronaut.io/1.0.4/guide/index.html#kafka)

Micronautを使って、Apache KafkaのProducerおよびConsumerを作ることができるようです。

[Micronaut Kafka](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/)

# 環境

今回使ったMicronautのバージョン。

```shell
$ mn -V
| Micronaut Version: 1.0.4
| JVM Version: 1.8.0_191
```

Micronaut 1.0.4では、Apache Kafka 2.0.1に対応しているようです。

> Upgrade to Kafka 2.0.1

[Micronaut Kafka](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/)

これに合わせて、今回はApache Kafka 2.0.1をダウンロード。

```shell
$ wget http://ftp.tsukuba.wide.ad.jp/software/apache/kafka/2.0.1/kafka_2.12-2.0.1.tgz
$ tar xf kafka_2.12-2.0.1.tgz
$ cd kafka_2.12-2.0.1
```

Apache KafkaのQuick Startに習い、Apache ZooKeeperとApache Kafka Brokerを起動します。

[Quick Start / Start the server](https://kafka.apache.org/20/documentation.html#quickstart_startserver)

```shell
## Apache Zookeeper
$ bin/zookeeper-server-start.sh config/zookeeper.properties

## Apache Kafka（Broker）
$ bin/kafka-server-start.sh config/server.properties
```

Topicは、`my-topic`という名前で作成しました。

```shell
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic my-topic
Created topic "my-topic".
```


# アプリケーションを作る

それでは、アプリケーションを作ってみます。ProducerとConsumer（ここではListenerとしていますが）用のプロジェクトを作成。

```shell
## Producer
$ mn create-app --features kafka --build maven hello-kafka-producer

## Listener（Consumer）
$ mn create-app --features kafka --build maven hello-kafka-listener
```

以降、この2つのプロジェクトに対してソースコードを追加、編集していきます。


## Producerを作る

では、Apache Kafka Brokerにメッセージを送信するProducerを書いてみます。

```shell
$ cd hello-kafka-producer
```

`main`クラスは、自動生成されたまま使います。

src/main/java/hello/kafka/producer/Application.java

```java
package hello.kafka.producer;

import io.micronaut.runtime.Micronaut;

public class Application {

    public static void main(String[] args) {
        Micronaut.run(Application.class);
    }
}
```

Producerは、`@KafkaClient`アノテーションを付けたインターフェースを作成するようです。

[Defining @KafkaClient Methods](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#kafkaClientMethods)

こんな感じで作成。インターフェースに対する実装クラスの作成は不要です。

src/main/java/hello/kafka/producer/MessageClient.java

```java
package hello.kafka.producer;

import io.micronaut.configuration.kafka.annotation.KafkaClient;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.reactivex.Single;

@KafkaClient(acks = KafkaClient.Acknowledge.ALL)
public interface MessageClient {
    @Topic("my-topic")
    Single<String> send(@KafkaKey String key, Single<String> message);
}
```

設定は、`@KafkaClient`アノテーションおよび設定ファイルで行います。

[](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#kafkaClientConfiguration)

`@KafkaClient`アノテーションでは、ACKの設定だけ行いました。

```java
@KafkaClient(acks = KafkaClient.Acknowledge.ALL)
public interface MessageClient {
```

設定ファイル、こちら。

src/main/resources/application.yml

```yaml
---
micronaut:
    application:
        name: hello-kafka-producer

---
kafka:
    bootstrap:
        servers: localhost:9092
    producers:
        default:
            retries: 5
```

`kafka.bootstrap.servers`では、Apache Kafka Brokerに対する接続設定を行います。

Producerの振る舞いについては、`kafka.producers.[client-id]`で行います。

`id`を指定することで、`@KafkaClient`単位に設定を行うことができます。

[Per @KafkaClient Producer Properties](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#_per_kafkaclient_producer_properties)

今回は特に`id`を指定していないので、`default`という名前になっています。

メッセージの送信は、`@Topic`アノテーションにトピック名を指定したメソッドを使って行います。

```java
    @Topic("my-topic")
    Single<String> send(@KafkaKey String key, Single<String> message);
```

キーを指定する場合は、`@KafkaKey`を使用します。キーを使わない場合は、指定不要です（キーの引数自体を省略できます）。

また、RxJavaなど、Reactive Streamsまわりの型も使えるので、今回はこちらを利用しました。

[Reactive and Non-Blocking Method Definitions](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#_reactive_and_non_blocking_method_definitions)

最後に、`@KafkaClient`を使う`@Controller`。

src/main/java/hello/kafka/producer/MessageController.java

```java
package hello.kafka.producer;

import javax.inject.Inject;

import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Body;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Post;
import io.reactivex.Single;

@Controller("/message")
public class MessageController {
    @Inject
    MessageClient messageClient;

    @Post(value = "/{key}", consumes = MediaType.TEXT_PLAIN, produces = MediaType.TEXT_PLAIN)
    public Single<String> message(String key, @Body Single<String> value) {
        return messageClient.send(key, value).map(v -> String.format("message [%s] sended", v));
    }
}
```

ビルド。

```shell
$ ./mvnw package
```

これで、Producer側は完了。


## Listener（Consumer）

続いて、Consumer側。

```shell
$ cd hello-kafka-listener
```

[Kafka Consumers Using @KafkaListener](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#kafkaListener)

Consumerは、`@KafkaListener`アノテーションを使って作成します。こちらは、`class`定義になります。

src/main/java/hello/kafka/listener/MessageListener.java

```java
package hello.kafka.listener;

import java.util.List;

import io.micronaut.configuration.kafka.Acknowledgement;
import io.micronaut.configuration.kafka.annotation.KafkaKey;
import io.micronaut.configuration.kafka.annotation.KafkaListener;
import io.micronaut.configuration.kafka.annotation.OffsetReset;
import io.micronaut.configuration.kafka.annotation.OffsetStrategy;
import io.micronaut.configuration.kafka.annotation.Topic;
import io.reactivex.Single;

@KafkaListener(groupId = "message-group", offsetReset = OffsetReset.EARLIEST, offsetStrategy = OffsetStrategy.DISABLED)
public class MessageListener {
    @Topic("my-topic")
    public void receive(@KafkaKey String key, Single<String> message, Acknowledgement acknowledgement) {
        message.subscribe(m -> {
            System.out.printf("Received key / message = %s / %s%n", key, m);
            acknowledgement.ack();
        });
    }
}
```

`@KafkaListener`アノテーションでConsumerの設定を行い、メッセージを受け取るメソッドには`@Topic`アノテーションを指定します。

```java
@KafkaListener(groupId = "message-group", offsetReset = OffsetReset.EARLIEST, offsetStrategy = OffsetStrategy.DISABLED)
public class MessageListener {
    @Topic("my-topic")
    public void receive(@KafkaKey String key, Single<String> message, Acknowledgement acknowledgement) {
```

使う型は、Reactiveなものを。

[Receiving and returning Reactive Types](https://micronaut-projects.github.io/micronaut-kafka/1.0.4/guide/#_receiving_and_returning_reactive_types)

また、ACKを使うようにもしています。

```java
        message.subscribe(m -> {
            System.out.printf("Received key / message = %s / %s%n", key, m);
            acknowledgement.ack();
        });
```

今回は設定はBrokerへの接続先のみとしました。あと、同一ホストでProducerも立ち上げることにするので、`micronaut.server.port`は`8080`以外で設定。

src/main/resources/application.yml

```yaml
---
micronaut:
    application:
        name: hello-kafka-listener
    server:
        port: 9080

---
kafka:
    bootstrap:
        servers: localhost:9092
```

こちらもビルド。

```shell
$ ./mvnw package
```

これで、Consumer側も準備完了です。


# 動かしてみる

では、動作確認してみましょう。

```shell
## Producerを起動
$ java -jar target/hello-kafka-producer-0.1.jar


## Consumerを起動
$ java -jar target/hello-kafka-listener-0.1.jar
```

Producerに適当にメッセージを放り込んでみます。

```shell
$ curl -H 'Content-Type: text/plain' localhost:8080/message/key1 -d 'value1'
message [value1] sended

$ curl -H 'Content-Type: text/plain' localhost:8080/message/key2 -d 'value2'
message [value2] sended

$ curl -H 'Content-Type: text/plain' localhost:8080/message/key3 -d 'value3'
message [value3] sended
```

Consumer側には、こんな感じで受信したメッセージが表示されます。

```shell
Received key / message = key1 / value1
Received key / message = key2 / value2
Received key / message = key3 / value3
```

動かせたようですね。