Helidon 2.0 with Kafka

原文はこちら。
This article was written by Daniel Kec (Java Developer, Oracle).
https://medium.com/helidon/helidon-2-0-with-kafka-b97ba4f422bd

Helidon 2.0でReactive Messagingがサポートされ、Helidon SE/MPで利用可能な最初のメッセージングコネクタの一つとしてKafkaコネクタが加わりました。Helidon MPはMicroProfileの仕様に準拠しており、CDIの素晴らしい機能を全て活用しています。Helidon SEでは、シンプルなAPI、明示的な設定、そして魔法ではない（no-magic）アプローチを楽しむことができます。

Configuration

Connecting Kafkaへの接続にあたっては通常自明でない構成が必要で、このような構成のオーケストレーションは難しいことがあります。簡単のため、Helidon SE/MPでは、Reactive Messagingのための外部構成でMicroProfileのReactive Messaging仕様で定義されているものと同じ記法をサポートします。

MicroProfile Reactive Messaging Specification – Configuration
https://download.eclipse.org/microprofile/microprofile-reactive-messaging-1.0/microprofile-reactive-messaging-spec.html#_configuration

以下は簡単な例です。

      
        mp.messaging:
      
         incoming.from-kafka:
      
           connector: helidon-kafka
      
           topic: messaging-test-topic-1
      
           auto.offset.reset: latest
      
           enable.auto.commit: true
      
           group.id: example-group-1
      
         outgoing.to-kafka:
      
           connector: helidon-kafka
      
           topic: messaging-test-topic-1
      
         connector:
      
           helidon-kafka:
      
             bootstrap.servers: localhost:9092
      
             key.serializer: org.apache.kafka.common.serialization.StringSerializer
      
             value.serializer: org.apache.kafka.common.serialization.StringSerializer
      
             key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
      
             value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

view raw helidon-kafka-article-01.yaml hosted with ❤ by GitHub

この例では、同じトピックに接続しているfrom-kafkaという着信チャネルと、to-kafkaという送出チャネルの簡単な構成を示しています。Helidon SE と MP の両方でまったく同じ設定を使用してみましょう。

Kafka in Helidon MP

Helidon MPはMicroProfile Reactive Messaging仕様を実装しているため、必要な依存関係を追加し、アプリケーションスコープのbeanのメソッドに注釈を付けるだけでOKです。

MicroProfile Reactive Messaging Specification ver.1.0
https://download.eclipse.org/microprofile/microprofile-reactive-messaging-1.0/microprofile-reactive-messaging-spec.html

      
                <dependency>
      
                    <groupId>io.helidon.microprofile.messaging</groupId>
      
                    <artifactId>helidon-microprofile-messaging</artifactId>
      
                </dependency>
      
                <dependency>
      
                    <groupId>io.helidon.messaging.kafka</groupId>
      
                    <artifactId>helidon-messaging-kafka</artifactId>
      
                </dependency>

view raw helidon-kafka-article-10-pom.xml hosted with ❤ by GitHub

メッセージのリスニングはシンプルです。

      
        @ApplicationScoped
      
        public class ExampleBean {
      
           @Incoming("from-kafka")
      
           public void broadcast(String payload) {
      
               System.out.println("Kafka says: " + payload);
      
           }

view raw helidon-kafka-article-02-mp.java hosted with ❤ by GitHub

メッセージ送信時には、バックプレッシャーの可能性を考慮する必要があります。このような作業のために、バッファを内蔵したSubmissionPublisherを使いましょう。

      
        public class ExampleBean {
      
           private final SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
      
           public void sendMessage(String message) {
      
               publisher.submit(message);
      
           }
      
           @Outgoing("to-kafka")
      
           public Publisher<String> preparePublisher() {
      
               return ReactiveStreams
      
                       .fromPublisher(FlowAdapters.toPublisher(publisher))
      
                       .buildRs();
      
           }

view raw helidon-kafka-article-04-mp.java hosted with ❤ by GitHub

Kafka in Helidon SE

Helidon SEではCDIを使わずにReactive Messagingを利用するための専用APIがあります。このAPIはMicroProfileの仕様の一部ではありませんが、コネクタや構成、メッセージラッピングや確認応答（acknowledgement）と互換性があります。これらすべては、Helidon SE/MP間の移行を可能な限りスムーズにするためのものです。

      
        <dependency>
      
           <groupId>io.helidon.messaging</groupId>
      
           <artifactId>helidon-messaging</artifactId>
      
        </dependency>
      
        <dependency>
      
           <groupId>io.helidon.messaging.kafka</groupId>
      
           <artifactId>helidon-messaging-kafka</artifactId>
      
        </dependency>

view raw helidon-kafka-article-02-pom.xml hosted with ❤ by GitHub

メッセージのリスニングは、Helidon MPの場合と同様簡単です。

      
        Channel<String> fromKafka = Channel.create("from-kafka");
      
        KafkaConnector kafkaConnector = KafkaConnector.create();
      
        Messaging.builder()
      
               .connector(kafkaConnector)
      
               .listener(fromKafka, payload -> {
      
                   System.out.println("Kafka says: " + payload);
      
               })
      
               .build()
      
               .start();

view raw helidon-kafka-article-05-se.java hosted with ❤ by GitHub

Helidon SEのMessaging APIがEmitterを提供しているので、メッセージを送信するためにSubmissionPublisherをわざわざ使う必要はありません。

      
        Channel<String> toKafka = Channel.create("to-kafka");
      
        KafkaConnector kafkaConnector = KafkaConnector.create();
      
        Emitter<String> emitter = Emitter.create(toKafka);
      
        Messaging.builder()
      
               .emitter(emitter)
      
               .connector(kafkaConnector)
      
               .build()
      
               .start();
      
        emitter.send("Hello Kafka!");

view raw helidon-kafka-article-06-se.java hosted with ❤ by GitHub

Acknowledgement

メッセージが適切に受信されたことが確実な時点までオフセットコミットを遅延させることが現実的な場合があります。多くは非同期処理中に発生する可能性があり、リアクティブストリームをブロックすることはできませんが、バックプレッシャー以外のフィードバックが必要です。reactive messagingの確認応答機能を使うと非常に便利です。Kafkaコネクタは auto.commit がオフのときに、確認済みメッセージのオフセットのみをコミットしています。

自動コミットをOffにする例です。

      
        mp.messaging:
      
         incoming.from-kafka:
      
           connector: helidon-kafka
      
           topic: messaging-test-topic-1
      
           enable.auto.commit: false
      
           ack.timeout.millis: 10000
      
           group.id: example-group-1

view raw helidon-kafka-article-07-se.yaml hosted with ❤ by GitHub

この例では、遅延オフセットコミットの原因となっている遅延確認応答を確認できます。

      
        Channel<String> fromKafka = Channel.create("from-kafka");
      
        KafkaConnector kafkaConnector = KafkaConnector.create();
      
        Messaging messaging = Messaging.builder()
      
               .connector(kafkaConnector)
      
               .subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
      
                       //Apply back-pressure, flatMapCompletionStage requests one by one
      
                       .flatMapCompletionStage(message -> {
      
                           return CompletableFuture.runAsync(() -> {
      
                               //Do something lengthy
      
                               Multi.timer(300, TimeUnit.MILLISECONDS, executor)
      
                                       .first()
      
                                       .await();
      
                               //Acknowledge message has been consumed
      
                               message.ack();
      
                           });
      
                       })
      
                       .ignore())
      
               .build()
      
               .start();

view raw helidon-kafka-article-08-se.java hosted with ❤ by GitHub

この例のように、次から次へとメッセージをリクエストすることでバックプレッシャーをかけているため、利用率はあまり良くありません。非同期メッセージの消費中に致命的な障害が発生した場合、消費されなかったメッセージは確認応答を返さないので、Kafka コネクタはオフセットをコミットしません。それゆえメッセージの消失はありません。リバランス後にメッセージを再び消費します。

これだけのことを20行の読みやすい関数型コードに詰め込んだのに、多くの設定が隠されていることがわかります。ではno-magicアプローチはどうでしょうか？

では、もっと明確にしてみましょう。

Explicit Configuration

Helidon SEのReactive Messagingは任意のチャネルのPublisherコネクタもしくはSubscriberコネクタに対しHelidon Configを直接渡すことができます。以下の例では、Kafka コネクタが独自の Configビルダーを持っていることがわかります。そのため、Kafka に接続するたびに、頻繁に使用するクライアントのプロパティを調べる必要はありません。

      
        Channel<String> fromKafka = Channel.<String>builder()
      
               .publisherConfig(KafkaConnector.configBuilder()
      
                       .bootstrapServers("localhost:9092")
      
                       .groupId("example-group-1")
      
                       .topic("messaging-test-topic-1")
      
                       .autoOffsetReset(KafkaConfigBuilder.AutoOffsetReset.LATEST)
      
                       .enableAutoCommit(false)
      
                       .keyDeserializer(StringDeserializer.class)
      
                       .valueDeserializer(StringDeserializer.class)
      
                       .build()
      
               )
      
               .build();
      
        // Prepare Kafka connector, can be used by any channel
      
        KafkaConnector kafkaConnector = KafkaConnector.create();
      
        Messaging.builder()
      
               .connector(kafkaConnector)
      
               .subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
      
                       //Apply back-pressure, flatMapCompletionStage request one by one
      
                       .flatMapCompletionStage(message -> {
      
                           return CompletableFuture.runAsync(() -> {
      
                               //Do something lengthy
      
                               Multi.timer(300, TimeUnit.MILLISECONDS, executor)
      
                                       .first()
      
                                       .await();
      
                               //Acknowledge message has been consumed
      
                               message.ack();
      
                           });
      
                       })
      
                       .ignore())
      
               .build()
      
               .start();

view raw helidon-kafka-article-09-se.java hosted with ❤ by GitHub

一つのスニペットに全て詰まっています。アブラカダブラ！まさにno-magicですね✨

HelidonにおけるReactive Messagingについて、詳細は以下をどうぞ。

Helidon Messaging with Kafka Examples
https://github.com/oracle/helidon/tree/master/examples/messaging
Reactive Messaging with Helidon 2.0
https://medium.com/helidon/reactive-messaging-with-helidon-2-0-f5de1ca5dc63
https://logico-jp.io/2020/03/29/reactive-messaging-with-helidon-2-0/

Helidon 2.0 with Kafka

Helidon Screencast – MicroProfile Reactive Messaging

コメントを残すコメントをキャンセル

コメントを入力してください。

以下に詳細を記入するか、アイコンをクリックしてログインしてください。

メール (必須) (アドレスが公開されることはありません)

名前 (必須)

サイト

WordPress.com アカウントを使ってコメントしています。 ( ログアウト / 変更 )

Google アカウントを使ってコメントしています。 ( ログアウト / 変更 )

Twitter アカウントを使ってコメントしています。 ( ログアウト / 変更 )

Facebook アカウントを使ってコメントしています。 ( ログアウト / 変更 )

キャンセル

%s と連携中

月	火	水	木	金	土	日
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30

	mp.messaging:

	incoming.from-kafka:
	connector: helidon-kafka
	topic: messaging-test-topic-1
	auto.offset.reset: latest
	enable.auto.commit: true
	group.id: example-group-1

	outgoing.to-kafka:
	connector: helidon-kafka
	topic: messaging-test-topic-1

	connector:
	helidon-kafka:
	bootstrap.servers: localhost:9092
	key.serializer: org.apache.kafka.common.serialization.StringSerializer
	value.serializer: org.apache.kafka.common.serialization.StringSerializer
	key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
	value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

	<dependency>
	<groupId>io.helidon.microprofile.messaging</groupId>
	<artifactId>helidon-microprofile-messaging</artifactId>
	</dependency>
	<dependency>
	<groupId>io.helidon.messaging.kafka</groupId>
	<artifactId>helidon-messaging-kafka</artifactId>
	</dependency>

	@ApplicationScoped
	public class ExampleBean {

	@Incoming("from-kafka")
	public void broadcast(String payload) {
	System.out.println("Kafka says: " + payload);
	}

	public class ExampleBean {

	private final SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

	public void sendMessage(String message) {
	publisher.submit(message);
	}

	@Outgoing("to-kafka")
	public Publisher<String> preparePublisher() {
	return ReactiveStreams
	.fromPublisher(FlowAdapters.toPublisher(publisher))
	.buildRs();
	}

	<dependency>
	<groupId>io.helidon.messaging</groupId>
	<artifactId>helidon-messaging</artifactId>
	</dependency>
	<dependency>
	<groupId>io.helidon.messaging.kafka</groupId>
	<artifactId>helidon-messaging-kafka</artifactId>
	</dependency>

	Channel<String> fromKafka = Channel.create("from-kafka");
	KafkaConnector kafkaConnector = KafkaConnector.create();

	Messaging.builder()
	.connector(kafkaConnector)
	.listener(fromKafka, payload -> {
	System.out.println("Kafka says: " + payload);
	})
	.build()
	.start();

	Channel<String> toKafka = Channel.create("to-kafka");
	KafkaConnector kafkaConnector = KafkaConnector.create();

	Emitter<String> emitter = Emitter.create(toKafka);

	Messaging.builder()
	.emitter(emitter)
	.connector(kafkaConnector)
	.build()
	.start();

	emitter.send("Hello Kafka!");

	Channel<String> fromKafka = Channel.create("from-kafka");
	KafkaConnector kafkaConnector = KafkaConnector.create();

	Messaging messaging = Messaging.builder()
	.connector(kafkaConnector)
	.subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
	//Apply back-pressure, flatMapCompletionStage requests one by one
	.flatMapCompletionStage(message -> {
	return CompletableFuture.runAsync(() -> {
	//Do something lengthy
	Multi.timer(300, TimeUnit.MILLISECONDS, executor)
	.first()
	.await();
	//Acknowledge message has been consumed
	message.ack();
	});
	})
	.ignore())
	.build()
	.start();

	Channel<String> fromKafka = Channel.<String>builder()
	.publisherConfig(KafkaConnector.configBuilder()
	.bootstrapServers("localhost:9092")
	.groupId("example-group-1")
	.topic("messaging-test-topic-1")
	.autoOffsetReset(KafkaConfigBuilder.AutoOffsetReset.LATEST)
	.enableAutoCommit(false)
	.keyDeserializer(StringDeserializer.class)
	.valueDeserializer(StringDeserializer.class)
	.build()
	)
	.build();

	// Prepare Kafka connector, can be used by any channel
	KafkaConnector kafkaConnector = KafkaConnector.create();

	Messaging.builder()
	.connector(kafkaConnector)
	.subscriber(fromKafka, ReactiveStreams.<Message<String>>builder()
	//Apply back-pressure, flatMapCompletionStage request one by one
	.flatMapCompletionStage(message -> {
	return CompletableFuture.runAsync(() -> {
	//Do something lengthy
	Multi.timer(300, TimeUnit.MILLISECONDS, executor)
	.first()
	.await();
	//Acknowledge message has been consumed
	message.ack();
	});
	})
	.ignore())
	.build()
	.start();