QuarkusとConfluent Cloudで始めるイベント駆動アプリケーション

Posted at 2026-08-04

はじめに

近年、アプリケーション間をAPIで直接連携するだけでなく、イベントを介して疎結合に連携するイベント駆動アーキテクチャが注目されています。

イベント駆動アーキテクチャでは、あるアプリケーションで発生した出来事を「イベント」としてメッセージング基盤へ送信し、別のアプリケーションがそのイベントを受信して処理します。

一方で、実際に試してみようとすると、次のような疑問が出てきます。

Kafkaをローカル環境に構築しないといけないのか
JavaアプリケーションからKafkaへどのように接続するのか
ProducerやConsumerをどこまで実装する必要があるのか
認証情報や接続設定をどのように管理するのか

今回は、Javaアプリケーションの開発にQuarkus、イベントストリーミング基盤にConfluent Cloudを使用し、簡単なイベント駆動アプリケーションを作成します。

REST APIで受け取った注文情報をConfluent Cloud上のKafkaトピックへ送信し、そのイベントをQuarkusアプリケーションで受信するところまで試してみます。

Quarkusとは

Quarkusは、コンテナやKubernetes環境での実行を意識して設計されたJavaフレームワークです。

起動時間やメモリ使用量を抑えやすいことに加えて、Dev Modeによるライブリロードや、多数の拡張機能が提供されています。

Kafkaとの連携には、quarkus-messaging-kafka拡張を利用できます。この拡張は、MicroProfile Reactive MessagingをベースとしたQuarkus MessagingによってKafkaとのメッセージ送受信を実現します。アプリケーションではチャネルを通じてメッセージを送受信し、設定ファイルでチャネルとKafkaトピックを関連付けます。

Kafkaとは

Apache Kafkaは、アプリケーション間でイベントを送受信するための分散型イベントストリーミングプラットフォームです。

イベントを送信する側をProducer、イベントを受信して処理する側をConsumerと呼びます。Producerが送信したイベントは、トピックと呼ばれる名前付きの保存先に書き込まれ、Consumerはそのトピックからイベントを読み取ります。

Producer → Kafkaトピック → Consumer

ProducerとConsumerが直接通信するのではなく、Kafkaのトピックを介してイベントをやり取りするため、それぞれを独立して開発、変更、拡張しやすくなります。

今回のアプリケーションでは、Quarkusが注文イベントをordersトピックへ送信し、同じトピックからイベントを受信します。

Confluent Cloudとは

Confluent Cloudは、Apache Kafkaをベースとするフルマネージドのデータストリーミングプラットフォームです。

Kafkaクラスタの構築や運用をサービス側に任せられるため、開発者はアプリケーションやイベント処理の実装に集中できます。Confluent Cloudでは、WebコンソールからKafkaクラスタやトピックを作成し、保存されたメッセージを確認できます。

今回はConfluent Cloud上にKafkaクラスタを作成し、Quarkusアプリケーションから接続します。

今回作成するアプリケーション

今回は、注文情報を扱う簡単なアプリケーションを作成します。

全体の流れは次のとおりです。

クライアント
    |
    | POST /orders
    v
Quarkus REST API
    |
    | 注文イベントを送信
    v
Confluent Cloud
ordersトピック
    |
    | 注文イベントを受信
    v
Quarkus Consumer
    |
    v
コンソールへのログ出力

REST APIへ次のようなJSONを送信します。

{
  "orderId": "order-001",
  "productName": "coffee",
  "quantity": 2
}

Quarkusアプリケーションは、受け取った注文情報をKafkaのordersトピックへ送信します。

同じアプリケーション内に用意したConsumerがordersトピックからイベントを受信し、その内容をコンソールへ出力します。

アプリケーション全体の構成は次のようになります。

Producer側はorders-outチャネルへメッセージを送るだけで、Consumer側はorders-inチャネルからメッセージを受け取るだけです。

チャネルとKafkaトピックの対応付けや、Confluent Cloudへの認証設定は、すべてapplication.properties側に集約されます。

今回作成したサンプルアプリケーションはこちらに公開しています。
https://github.com/ktgrryt/quarkus-confluent-demo

前提環境

今回の検証では、以下の環境を使用します。

JDK 21以上
Git
Visual Studio Codeなどのエディタ
Confluent Cloudアカウント
curlまたはRESTクライアント

本記事ではMaven Wrapperを使用するため、Mavenを個別にインストールしていなくても、生成したプロジェクト内のmvnwまたはmvnw.cmdからビルドできます。

Confluent Cloudの準備

Kafkaクラスタを作成する

Confluent Cloudへログインし、検証用のEnvironmentとKafkaクラスタを作成します。

今回は、入門用途としてBasicクラスタを使用します。

設定例は次のとおりです。

Environment: quarkus-handson
Cluster name: quarkus-kafka-cluster
Cluster type: Basic
Region: 任意のリージョン

クラスタ作成時に選択したクラウドプロバイダーとリージョンは、作成後に変更できないため注意が必要です。

Kafkaトピックを作成する

続いて、イベントの送信先となるKafkaトピックを作成します。

今回は次の名前で作成します。

orders

Kafkaトピックは、Producerがイベントを書き込み、Consumerがイベントを読み取るための名前付きストリームです。

Confluent CloudではKafkaトピックの自動作成がデフォルトで無効になっているため、アプリケーションを起動する前にordersトピックを作成しておきます。

APIキーを作成する

QuarkusからConfluent Cloudへ接続するため、Kafkaクラスタ用のAPIキーを作成します。

Select account for API keyはMy accountを選択します。

画面に表示されるAPI KeyとAPI Secretを保管します。

アプリケーションからの接続には、主に次の情報を使用します。

Bootstrap Server
API Key
API Secret

Confluent Cloud ConsoleのClients画面から、クライアント用の接続情報を確認できます。Confluent Cloudへ接続するKafkaクライアントではTLSによる暗号化と認証が必要であり、今回はSASL_SSLとSASL/PLAINを使用します。

API Secretは、ソースコードやGitリポジトリに含めないように注意します。

Quarkusプロジェクトを作成する

Quarkusプロジェクトは、https://code.quarkus.io/またはQuarkus CLIから作成できます。

今回は次の拡張機能を追加します。

REST Jackson
Messaging - Kafka Connector

コマンドから作成する場合は、次のように実行します。

quarkus create app com.example:quarkus-confluent-demo \
  --extension='rest-jackson,messaging-kafka' \
  --no-code

作成したプロジェクトへ移動します。

cd quarkus-confluent-demo

注文イベントを定義する

最初に、Kafkaへ送信する注文イベントを定義します。

package com.example;

public record OrderEvent(
        String orderId,
        String productName,
        int quantity) {
}

今回はJavaのrecordを使用し、注文番号、商品名、数量を持つシンプルなイベントとしました。

実際の業務アプリケーションでは、イベントの発生日時やイベントID、イベントの種類などを追加することも考えられます。

Kafkaへイベントを送信する

次に、注文情報を受け取るREST APIを作成します。

package com.example;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.smallrye.reactive.messaging.MutinyEmitter;
import jakarta.inject.Inject;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.eclipse.microprofile.reactive.messaging.Channel;

@Path("/orders")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public class OrderResource {

    @Inject
    @Channel("orders-out")
    MutinyEmitter<String> emitter;

    @Inject
    ObjectMapper objectMapper;

    @POST
    public Response createOrder(OrderEvent order)
            throws JsonProcessingException {

        String json = objectMapper.writeValueAsString(order);
        emitter.sendAndAwait(json);

        return Response.accepted(order).build();
    }
}

ポイントは、@Channelで送信先のチャネルを指定している部分です。

@Channel("orders-out")
MutinyEmitter<String> emitter;

orders-outはKafkaトピック名ではなく、Quarkus Messaging内で使用するチャネル名です。

このチャネルとKafkaのordersトピックの対応関係は、後ほどapplication.propertiesに設定します。

Quarkus Messagingでは、アプリケーションのコードはチャネルを通じてメッセージを送受信し、KafkaコネクターがチャネルとKafkaトピックを接続します。これにより、メッセージの送信処理とKafka固有の接続設定を分離できます。

Kafkaからイベントを受信する

続いて、ordersトピックから注文イベントを受信するConsumerを作成します。

package com.example;

import jakarta.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.jboss.logging.Logger;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@ApplicationScoped
public class OrderConsumer {

    private static final Logger LOG = Logger.getLogger(OrderConsumer.class);

    private final List<String> receivedOrders = new CopyOnWriteArrayList<>();

    @Incoming("orders-in")
    public void consume(String orderJson) {
        LOG.infof("注文イベントを受信しました: %s", orderJson);
        receivedOrders.add(orderJson);
    }

    public List<String> getReceivedOrders() {
        return List.copyOf(receivedOrders);
    }
}

@Incomingを付けたメソッドは、指定したチャネルからメッセージを受信します。

@Incoming("orders-in")

また、受信したイベントはCopyOnWriteArrayListへ保持し、getReceivedOrders()から取得できるようにしています。

これは動作確認用の簡易的な実装です。サンプルアプリでは、このメソッドを利用するGET /received-ordersのREST APIを用意しており、Confluent Cloud Consoleを開かずに受信結果を確認できます。受信データはメモリ上にのみ保持されるため、アプリケーションを再起動すると消えます。

今回は受信した注文イベントをログへ出力して保持するだけですが、実際のアプリケーションでは次のような処理が考えられます。

注文データをデータベースへ保存する
在庫を更新する
配送処理を開始する
通知サービスへ別のイベントを送信する

Producerは「イベントを送った後に誰が処理するか」を意識する必要がありません。Consumer側も、REST APIを直接呼び出したアプリケーションについて知る必要がありません。

このように、イベントを介してアプリケーション同士を分離できることが、イベント駆動アーキテクチャの特徴の一つです。

Confluent Cloudへの接続を設定する

src/main/resources/application.propertiesに、Confluent Cloudへの接続設定とチャネルの設定を追加します。

# Confluent Cloudへの接続
kafka.bootstrap.servers=${KAFKA_BOOTSTRAP_SERVERS}
kafka.security.protocol=SASL_SSL
kafka.sasl.mechanism=PLAIN
kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='${KAFKA_API_KEY}' password='${KAFKA_API_SECRET}';

# Producer
mp.messaging.outgoing.orders-out.connector=smallrye-kafka
mp.messaging.outgoing.orders-out.topic=orders
mp.messaging.outgoing.orders-out.value.serializer=org.apache.kafka.common.serialization.StringSerializer

# Consumer
mp.messaging.incoming.orders-in.connector=smallrye-kafka
mp.messaging.incoming.orders-in.topic=orders
mp.messaging.incoming.orders-in.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.orders-in.group.id=quarkus-confluent-demo
mp.messaging.incoming.orders-in.auto.offset.reset=earliest

設定のポイントは、アプリケーション内のチャネルとKafkaトピックを関連付けている部分です。

mp.messaging.outgoing.orders-out.topic=orders
mp.messaging.incoming.orders-in.topic=orders

orders-outチャネルへ送られたメッセージは、Kafkaのordersトピックへ書き込まれます。

一方、ordersトピックから読み取ったメッセージは、orders-inチャネルを通じて@Incomingを付けたメソッドへ渡されます。

接続情報は環境変数から取得する構成にします。

macOSまたはLinux

export KAFKA_BOOTSTRAP_SERVERS="<Bootstrap Server>"
export KAFKA_API_KEY="<API Key>"
export KAFKA_API_SECRET="<API Secret>"

Windows PowerShell

$env:KAFKA_BOOTSTRAP_SERVERS="<Bootstrap Server>"
$env:KAFKA_API_KEY="<API Key>"
$env:KAFKA_API_SECRET="<API Secret>"

このようにしておくことで、APIキーをapplication.propertiesやGitリポジトリへ直接保存せずに済みます。

アプリケーションを起動する

Quarkus Dev Modeでアプリケーションを起動します。

macOSまたはLinux

./mvnw quarkus:dev

Windows

.\mvnw.cmd quarkus:dev

正常に起動すると、アプリケーションは次のURLで待機します。

http://localhost:8080

Quarkus Dev Modeでは、コードを変更して保存するとアプリケーションへ変更が反映されます。KafkaのProducerやConsumerを変更しながら試す場合にも便利です。

注文イベントを送信してみる

別のターミナルから、注文情報をREST APIへ送信します。

curl -i -X POST http://localhost:8080/orders \
  -H "Content-Type: application/json" \
  -d '{
    "orderId": "order-001",
    "productName": "coffee",
    "quantity": 2
  }'

正常に処理されると、HTTP 202が返ります。

HTTP/1.1 202 Accepted

Quarkusアプリケーションのログには、Consumerが受信したイベントが表示されます。

2026-08-03 17:30:30,240 INFO  [com.example.OrderConsumer] (vert.x-worker-thread-1) 注文イベントを受信しました: {"orderId":"order-001","productName":"coffee","quantity":2}

これにより、次の流れを確認できました。

QuarkusのREST APIが注文情報を受け取る
Quarkusが注文イベントをConfluent Cloudへ送信する
イベントがKafkaのordersトピックに保存される
QuarkusのConsumerがイベントを受信する
受信した内容がログへ出力される

Confluent Cloud Consoleで確認する

最後に、Confluent Cloud Consoleからordersトピックを開き、送信したメッセージを確認します。

今回送信した注文情報が、JSON文字列として表示されていれば成功です。

Confluent Cloud Consoleでは、トピックの作成や設定変更に加えて、トピックに保存されているイベントを確認できます。

アプリケーションのログだけでなく、Kafkaトピック上にもメッセージが存在することを確認することで、ProducerとConsumerの間にConfluent Cloudが入っていることを視覚的に理解できます。

今回試して分かったこと

今回の検証では、QuarkusからConfluent Cloud上のKafkaへ接続し、注文イベントを送受信しました。

特に重要だと感じたポイントは、次の3点です。

1. Kafka固有のコードを少なくできる

Quarkus Messagingでは、@Channelや@Incomingを使ってProducerとConsumerを実装できます。

Kafkaへの接続情報やトピック名を設定ファイルに分離できるため、アプリケーションコードでは「どのイベントを送るか」「受信したイベントをどう処理するか」に集中できます。

2. Kafkaクラスタをローカルへ構築しなくても試せる

Confluent Cloudを利用することで、ローカルPCにKafkaクラスタを構築せずにイベント駆動アプリケーションを試すことができました。

ローカル環境ではQuarkusアプリケーションのみを起動し、Kafkaクラスタの管理はConfluent Cloudへ任せる構成になります。

3. ProducerとConsumerを疎結合にできる

ProducerはイベントをKafkaトピックへ送るだけで、そのイベントを誰が処理するかを知る必要がありません。

将来的に別のConsumerを追加した場合も、Producer側を変更せず、同じイベントを異なる目的で利用できます。

例えば、注文イベントに対して次のConsumerを追加できます。

ordersトピック
    ├── 在庫更新サービス
    ├── 配送手配サービス
    ├── 顧客通知サービス
    └── 売上分析サービス

イベントを起点として処理を追加できるため、アプリケーション間を直接接続する場合と比べて、システムを拡張しやすくなります。

後片付け

Confluent Cloudのクラスタは、アプリケーションを停止しただけでは削除されません。

検証が終わったら、不要になった以下のリソースを削除します。

Kafka APIキー
Kafkaクラスタ
検証用Environment

Confluent Cloudでは、作成したKafkaクラスタや利用量に応じて料金が発生する場合があります。検証が終わったら、不要なクラスタを削除し、Confluent Cloud ConsoleのBilling画面で利用状況を確認してください。
無料クレジットや試用条件は変更される可能性があるため、作成時点の画面と公式情報を確認してください。

まとめ

今回は、QuarkusとConfluent Cloudを組み合わせて、簡単なイベント駆動アプリケーションを作成しました。

Quarkusを使用することで、REST APIとKafkaを利用するアプリケーションを比較的少ないコードで実装できました。

また、Kafka基盤としてConfluent Cloudを使用したことで、ローカル環境にKafkaクラスタを構築、運用することなく、アプリケーションの実装とイベント処理の確認に集中できました。

QuarkusとConfluent Cloudの役割は、次のように整理できます。

Quarkus: イベントを生成、送信、処理するJavaアプリケーション
Confluent Cloud: イベントを安全かつ継続的に流通させるストリーミング基盤

今回の実装は文字列形式のJSONを送受信するシンプルなものでしたが、次のステップとして以下も試してみたいと思います。

Confluent Schema Registryを使用したスキーマ管理
AvroやJSON Schemaによるイベント形式の定義
ProducerとConsumerを別々のQuarkusアプリケーションに分割
複数のConsumerによるイベント処理
エラー発生時の再試行やDead Letter Queue
Quarkusアプリケーションのコンテナ化

まずは小さなイベントの送受信から試してみることで、イベント駆動アプリケーションにおけるQuarkusとConfluent Cloudそれぞれの役割を理解できました。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up