はじめに

GxP の清田です。

イベント駆動型アーキテクチャを実現するための技術の1つである、
Apach Kafka (以降 Kafka と記載) を動かして学んでみます。

素振りをしようと思った動機は、@yusuke_arclamp さんの『マイクロサービスの次に来るかもしれない言葉について』や、
社内イベント『TechTalk「この20年の Agile, DevOps, Microservices の流れについて」』で Kafka を耳にして気になっていたからです。

Microservice におけるシステム間連携の方法として、イベント駆動型アーキテクチャには以下の特徴があります。

同期的なAPIの連携でなく非同期の連携
データを共有するのではなくイベントを配信する
イベントの発行元は誰が受け取るかを気にしなくて良い(=疎結合)

これまでの開発経験で触れたことのないものだったので、一度動くアプリケーションを作って恐怖心をなくしておこうと思います。

対象読者

イベント駆動型アーキテクチャに興味のある人
Kafka を 動かして概念の一部を理解したい人
- Docker Compose でコンテナを起動して CLI で動かして学ぶ
- Spring for Apache Kafka で簡単な producer, consumer を実装する
イベントを複数のシステムが受け取って、それぞれの解釈で処理する 実装のイメージを掴みたい人

上記のいずれかに該当する場合、楽しんでいただけるかも知れません。

記載しないこと

脱 Zookeeper を支える技術である Apache Kafka Raft (KRaft) について
Kafka のクラウド環境でのデプロイ(Amazon Managed Streaming for Apache Kafka など)について
イベントの配信の保証(At Least One や Exactly Once) について
RabbitMQ などの message brokers との比較について
Amazon Simple Queue Service や Cloud Pub/Sub などのパブリッククラウドで提供されているサービスとの使い分けについて

上記の話題については、学習中で知識を有していないため記載しないです。(いずれも気になる)

Kafkaとは

あたりの説明で日本語でざっと読めるものとしては、以下の記事が大変参考になりました。
この記事では概念の1つずつの紹介はしていないので、参照いただけますとイメージを掴みやすいと思います。

個人的には、動かして学ぶことでこの辺りの記載を実感を持って理解することができました。

Apache Kafka（以降、Kafka）はスケーラビリティに優れた分散メッセージキューです。
Kafkaは処理性能を重視したメッセージキューであり、複数台のマシンでクラスタを構成して分散処理を行うことで、高いスループットを発揮します。後からクラスタにマシンを追加することで、処理性能とデータ保持容量をスケールアウトすることもできます。また、Kafkaはクラスタ内でデータを複製するため、一部のマシンに障害が発生してもデータを失うことなく処理を継続できます。

※イメージを掴んだ上で、Confluent のドキュメントや Kafaka のドキュメントを確認しました

Docker Compose でコンテナを起動して CLI で動かして学ぶ

まずは低レイヤーから学んでいきます。
登場人物は以下の図のものです。

GitHub にサンプルを用意(KiyotaTakeshi/kafka-playground) しましたので手元で動かして確認ができます。

さくっと broker 3台でクラスターを作成します。
※今回は Zookeeper が必要な image(bitnami/kafka) を使用しています

$ git clone https://github.com/KiyotaTakeshi/kafka-playground.git

$ cd  kafka-playground

$ docker compose up -d

$ docker compose ps
NAME                           COMMAND                  SERVICE             STATUS              PORTS
kafka-playground-kafka1-1      "/opt/bitnami/script…"   kafka1              running             0.0.0.0:49760->9092/tcp
kafka-playground-kafka2-1      "/opt/bitnami/script…"   kafka2              running             0.0.0.0:49761->9092/tcp
kafka-playground-kafka3-1      "/opt/bitnami/script…"   kafka3              running             0.0.0.0:49762->9092/tcp
kafka-playground-zookeeper-1   "/opt/bitnami/script…"   zookeeper           running             0.0.0.0:2181->2181/tcp

任意の broker コンテナに入り、topic を作成します。

# どの broker に exec するかは任意(今回は kafka1)
$ docker compose exec kafka1 bash

# まずは replica は1でつくる
# partitions は4でつくる
$ /opt/bitnami/kafka/bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --replication-factor 1 --partitions 4

作成したものを確認します。
出力から以下のことがわかります。

broker が3台に対して、 partition を4つに設定したため、1002(数字は実行によって異なります) の broker が2つの partiton の Leader になっている
replica は1にしているので、各 partition は1台の broker のみが保持している
- Isr(In-Sync Replica) が Leader と同じ broker のみ

$ /opt/bitnami/kafka/bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-topic

Topic: test-topic	TopicId: sqlIpd0oSEC3d7PJC3c_Dg	PartitionCount: 4	ReplicationFactor: 1	Configs: segment.bytes=1073741824
	Topic: test-topic	Partition: 0	Leader: 1002	Replicas: 1002	Isr: 1002
	Topic: test-topic	Partition: 1	Leader: 1001	Replicas: 1001	Isr: 1001
	Topic: test-topic	Partition: 2	Leader: 1003	Replicas: 1003	Isr: 1003
	Topic: test-topic	Partition: 3	Leader: 1002	Replicas: 1002	Isr: 1002

では、producer と consumer でメッセージのやり取りができることを確認します。
※関係する概念だけ記載した図です

シェルを2つ起動する必要があります。

# シェル1
# consumer の起動
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

# シェル2
# どの Broker に exec するかは任意
$ docker compose exec kafka1 bash

# producer を起動しメッセージを送る
$ /opt/bitnami/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>hello
>world

consumer 側でメッセージを受信できました。

# シェル1
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

hello
world

もう1つシェルを立ち上げて、 consumer を増やしてみます。

# シェル3
$ docker compose exec kafka1 bash

$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

producer からメッセージを送ると、

# シェル2
$ /opt/bitnami/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>goodbye

2つの consumer がメッセージを受信できていることがわかります。

# シェル1
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

goodbye

# シェル3
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

goodbye

このことから、 同じ topic から2つの consumer がメッセージを受信できる ことがわかります。
イベント駆動型アーキテクチャにおける、「イベントを複数のシステムが受け取る」が実現 できていますね(まだ CLI 上ですが)。

2つの consumer が別の subscriber として振る舞ったのはなぜでしょうか？
理由は、別の consumer group だからです。

producer を停止して、確認してみます。

# シェル2
$ /opt/bitnami/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
console-consumer-32834
console-consumer-40587

では、2つの consumer を同じ consumer group で立ち上げます。

# シェル1
# console-consumer という consumer group で起動
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic --group console-consumer

# シェル3
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic test-topic --group console-consumer

この状態で、 producer からメッセージを送ると

# シェル2
$ /opt/bitnami/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>hello
>world
>goodbye

2分の1でそれぞれの consumer がメッセージを受信します。

# シェル1
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
> --topic test-topic --group console-consumer

goodbye

# シェル3
$ /opt/bitnami/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
> --topic test-topic --group console-consumer

hello
world

つまり、 とある topic に対して同じ処理をする consumer のスケールアウト が実現できました(今回はただコンソールに出力されるだけですが)。

ここまでで

別の consumer group で起動して、同じ topic を複数の consumer で受け取る
- イベントのシステムごとの解釈
同じ consumer group で起動して、topic を複数の同じ処理をする consumer 間で受け取る
- 同じ解釈をするシステムのスケールアウト

が確認できました。

では、次は broker 側のスケールアウトについて見ていきます。

以下の図のように、複数の replica を指定することで Topic の partition を複数の broker に持つことができます。

まず、コンテナを停止しホスト側にマウントしているデータも吹き飛ばします。
(先程作成した topic や consumer group 等がなくなります)

$ docker compose down && rm -rf .docker

コンテナを再起動し、コンテナに入ります。

$ docker compose up -d && docker compose ps

$ docker compose exec kafka1 bash

今度は、 replica 数を3(broker と同じ台数)で指定し、作成します。

# 作成
$ /opt/bitnami/kafka/bin/kafka-topics.sh --create --topic test-topic-replicated \
> --bootstrap-server localhost:9092 --replication-factor 3 --partitions 4

作成したものを確認します。
出力から以下のことがわかります。

broker が3台に対して、 partition を4つに設定したため、1002(数字は実行によって異なります) の broker が2つの partiton の Leader になっている
replica を3にしているので、各 partition は3台の broker が保持している
- Isr(In-Sync Replica) が3台

$ /opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test-topic-replicated

Topic: test-topic-replicated	TopicId: w7ITwLFVTZi3m2AKNAUYIg	PartitionCount: 4	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test-topic-replicated	Partition: 0	Leader: 1002	Replicas: 1002,1001,1003	Isr: 1002,1001,1003
	Topic: test-topic-replicated	Partition: 1	Leader: 1001	Replicas: 1001,1003,1002	Isr: 1001,1003,1002
	Topic: test-topic-replicated	Partition: 2	Leader: 1003	Replicas: 1003,1002,1001	Isr: 1003,1002,1001
	Topic: test-topic-replicated	Partition: 3	Leader: 1002	Replicas: 1002,1003,1001	Isr: 1002,1003,1001

では、 broker を停止してみましょう。

# 任意のコンテナを停止
$ docker compose stop kafka2

topic を確認してみると(コンテナ内部に入ってコマンド実行でもいいです)、出力から以下のことがわかります。

停止した broker が Leader でなくなっている
停止した broker が Isr(In-Sync Replica) に含まれなくなっている
動作している broker で partition の Leader を分け合っている

$ docker compose exec -T kafka1 /bin/bash <<EOF
/opt/bitnami/kafka/bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-topic-replicated
EOF

Topic: test-topic-replicated	TopicId: w7ITwLFVTZi3m2AKNAUYIg	PartitionCount: 4	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test-topic-replicated	Partition: 0	Leader: 1001	Replicas: 1002,1001,1003	Isr: 1001,1003
	Topic: test-topic-replicated	Partition: 1	Leader: 1001	Replicas: 1001,1003,1002	Isr: 1001,1003
	Topic: test-topic-replicated	Partition: 2	Leader: 1003	Replicas: 1003,1002,1001	Isr: 1003,1001
	Topic: test-topic-replicated	Partition: 3	Leader: 1003	Replicas: 1002,1003,1001	Isr: 1003,1001

では、止めていた broker を起動します。

$ docker compose start kafka2

再度 topic を確認してみると以下のことがわかりました。

Isr(In-Sync Replica) に再度起動した broker が含まれている
再度起動した broker は partition の Leader には選出されていない

$ docker compose exec -T kafka1 /bin/bash <<EOF
/opt/bitnami/kafka/bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-topic-replicated
EOF

Topic: test-topic-replicated	TopicId: w7ITwLFVTZi3m2AKNAUYIg	PartitionCount: 4	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test-topic-replicated	Partition: 0	Leader: 1001	Replicas: 1002,1001,1003	Isr: 1001,1003,1002
	Topic: test-topic-replicated	Partition: 1	Leader: 1001	Replicas: 1001,1003,1002	Isr: 1001,1003,1002
	Topic: test-topic-replicated	Partition: 2	Leader: 1003	Replicas: 1003,1002,1001	Isr: 1003,1001,1002
	Topic: test-topic-replicated	Partition: 3	Leader: 1003	Replicas: 1002,1003,1001	Isr: 1003,1001,1002

※ちなみに、この状態で別の broker (今回だと kafka3)を停止すると、
先程起動した broker が Leader に含まれる挙動になりました

これらの検証で、 broker の台数を増やし replica に設定することで、
producer, consumer とやりとりをする基盤としての broker の可用性を高めることができるとわかりました。

最後に broker が実体としてどのように topic を保存しているのかを確認します。

ホスト側にマウントしているデータを確認します。topic の partition ごとにディレクトリが切られているようです。

$ ls .docker/kafka1/data/test-topic-replicated-?
.docker/kafka1/data/test-topic-replicated-0:
00000000000000000000.index	00000000000000000000.log	00000000000000000000.timeindex	leader-epoch-checkpoint		partition.metadata

.docker/kafka1/data/test-topic-replicated-1:
00000000000000000000.index	00000000000000000000.log	00000000000000000000.timeindex	leader-epoch-checkpoint		partition.metadata

.docker/kafka1/data/test-topic-replicated-2:
00000000000000000000.index	00000000000000000000.log	00000000000000000000.timeindex	leader-epoch-checkpoint		partition.metadata

.docker/kafka1/data/test-topic-replicated-3:
00000000000000000000.index	00000000000000000000.log	00000000000000000000.timeindex	leader-epoch-checkpoint		partition.metadata

$ ls -ld .docker/kafka1/data/test-topic-replicated-*
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:03 .docker/kafka1/data/test-topic-replicated-0
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:03 .docker/kafka1/data/test-topic-replicated-1
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka1/data/test-topic-replicated-2
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka1/data/test-topic-replicated-3

replica を3で指定したため、それぞれの broker に同様のデータが存在しています。

$ ls -ld .docker/kafka2/data/test-topic-replicated-?
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:12 .docker/kafka2/data/test-topic-replicated-0
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka2/data/test-topic-replicated-1
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka2/data/test-topic-replicated-2
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:12 .docker/kafka2/data/test-topic-replicated-3

$ ls -ld .docker/kafka3/data/test-topic-replicated-?
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka3/data/test-topic-replicated-0
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 15:33 .docker/kafka3/data/test-topic-replicated-1
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:03 .docker/kafka3/data/test-topic-replicated-2
drwxr-xr-x  7 t.kiyota  AD\Domain Users  224 Dec 15 16:03 .docker/kafka3/data/test-topic-replicated-3

以上で CLI での検証は終わります。
コンテナ起動して CLI で動作検証することで Kafka の主要な概念についてイメージが膨らんだのではないでしょうか。

Spring for Apache Kafka で簡単な producer, consumer を実装する

次は Kafka とやりとりをするアプリケーションを開発し、イベント駆動型アーキテクチャの理解を深めていきます。

最終的に動作するものは、
GitHub のサンプル(KiyotaTakeshi/kafka-spring-sample)の Gif を見ていただくとイメージがしやすいと思います。
Qiita の容量制限を超えているのか Gif が貼れなかった...

図にすると以下のものです。

order producer(Spring Boot) がWebリクエストを受け付けて、 topic(order-events) にイベントを発行
order notification(Spring Boot) がイベントを受け取ってメールを送信
- コンテナ起動している mailhog にて受信したメールを確認できる
order rdb(Sring Boot) がイベントを受け取って MySQL に保存

という構成です。

producer 側のアプリケーションはイベントを発行するところまでしか気にしない
consumer はイベントを受け取って各々の処理を行う

というイベント駆動型アーキテクチャをシンプルですが実現しました。
(他にもオブジェクトのまま NoSQL に保存する consumer があってもいいかも)

では order producer の実装を見ていきます。

topic の作成は Configuration クラスにより行っています。

@Configuration
public class AutoCreateConfig {

    // if you create manual
    // $ ./bin/kafka-topics.sh --create --topic order-events --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3
    @Bean
    public NewTopic orderEvents(){
        return TopicBuilder.name("order-events")
                .partitions(3)
                .replicas(3)
                .build();
    }
}

controller がリクエストを受け付けて、

@RestController
@RequestMapping("/order-events")
@Slf4j
@AllArgsConstructor
public class OrderEventsController {

    private OrderEventProducer orderEventProducer;

    @PostMapping
    public ResponseEntity<OrderEvent> postOrderEvent(@RequestBody @Valid OrderEvent orderEvent)
            throws JsonProcessingException {

        log.info("invoke kafka producer");
        orderEvent.setOrderEventType(OrderEventType.NEW);
        orderEventProducer.sendOrderEvent(orderEvent);

        return ResponseEntity.status(HttpStatus.CREATED).body(orderEvent);
    }

kafkaTemplate を使用して topic に対して非同期で message を送信します。

    public ListenableFuture<SendResult<Integer, String>> sendOrderEvent(OrderEvent orderEvent) throws JsonProcessingException {
        Integer key = orderEvent.getOrderEventId();
        String value = objectMapper.writeValueAsString(orderEvent);
        ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(TOPIC, key, value);

        listenableFuture.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable ex) {
                handleFailure(key, value, ex);
            }

            @Override
            public void onSuccess(SendResult<Integer, String> result) {
                handleSuccess(key, value, result);
            }
        });
        return listenableFuture;
    }

イベントを配るだけなのでシンプルな実装になっています。

application.yaml は接続先の Kafka Cluster の情報やシリアライズ、デシリアライズの設定を記載します。
admin の設定は上記の topic の生成で必要なものです。

spring:
  kafka:
    template:
      default-topic: "order-events"
    producer:
      bootstrap-servers:
        - localhost:9092
        - localhost:9093
        - localhost:9094
      key-serializer: org.apache.kafka.common.serialization.IntegerSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    admin:
      properties:
        bootstrap.servers: localhost:9092,localhost:9093,localhost:9094

次に order notification の実装を見ていきます。

こちらの application.yaml は consumer として接続先の Kafka Cluster の情報を記載しています。
また、メールの送信先の情報も記載しています。
(今回はコンテナ起動した mailhog を使用していますが、 mailtrap のようなクラウドサービスを使ってもいいかもです)

group-id: order-notification-events-group で指定してるのは consumer group です。

spring:
  kafka:
    consumer:
      bootstrap-servers:
        - localhost:9092
        - localhost:9093
        - localhost:9094
      key-deserializer: org.apache.kafka.common.serialization.IntegerDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: order-notification-events-group
  mail:
    # if you use mailtrap instead of mailhog
    # host: smtp.mailtrap.io
    host: localhost
    port: 1025
    # username: dummy
    # password: dummy
    protocol: smtp

@KafkaListener で topic を指定することで、 event を受け取って処理することができます。

@Component
@Slf4j
@AllArgsConstructor
public class OrderEventsConsumer {

    private final OrderEventsService orderEventsService;

    @KafkaListener(topics = {"order-events"})
    public void onMessage(ConsumerRecord<Integer, String> consumerRecord) throws JsonProcessingException {
        log.info("ConsumerRecord: {} ", consumerRecord);
        orderEventsService.processOrderEvent(consumerRecord);
    }
}

テンプレートエンジンである Thymeleaf で、
event として受け取ったオブジェクトを本文に含めた HTML 形式のメールを送信します。

    private void sendMail(NotificationEmail notificationEmail) {
        MimeMessagePreparator messagePreparer = mimeMessage -> {
            MimeMessageHelper helper = new MimeMessageHelper(mimeMessage, "UTF-8");
            helper.setFrom("kafka-spring-sample@example.com");
            helper.setTo(notificationEmail.getRecipient());
            helper.setSubject(notificationEmail.getSubject());
            helper.setText(build(notificationEmail.getBody()), true);
        };
        try {
            mailSender.send(messagePreparer);
            log.info("Email sent!!");
        } catch (MailException e) {
            log.error("Exception occurred when sending mail", e);
            throw new RuntimeException("Exception occurred when sending mail to " + notificationEmail.getRecipient(), e);
        }
    }

最後に order rdb の実装を見ていきます。

application.yaml は order notification と同じ Kafka に関する設定に加えて DB 周りの設定を記載しています。

order notification とは別の consumer group の group-id: order-rdb-events-group が指定されていることがポイントです。

CLI で確認したように 同じ topic から consumer が別の処理をするため(同じイベントを複数のシステムが受け取るため)に、別のIDが指定 されています。

spring:
  kafka:
    consumer:
      bootstrap-servers:
        - localhost:9092
        - localhost:9093
        - localhost:9094
      key-deserializer: org.apache.kafka.common.serialization.IntegerDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: order-rdb-events-group
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/kafka-spring-sample?createDatabaseIfNotExist=true
    username: root
    password: 1qazxsw2
  jpa:
    show-sql: true
    properties:
      hibernate:
        dialect: org.hibernate.dialect.MySQL8Dialect
        format_sql: true
    hibernate:
      ddl-auto: update
# パラメータの値をログに表示 ex.) binding parameter [1] as [INTEGER] - [1]
# 2021-11-21 00:55:11.604 TRACE 11400 --- [ main] o.h.type.descriptor.sql.BasicBinder      : binding parameter [1] as [BIGINT] - [1]
logging.level.org.hibernate.type.descriptor.sql.BasicBinder: trace

topic を指定してイベントを受け取る処理は同じです。

@Component
@Slf4j
@AllArgsConstructor
public class OrderEventsConsumer {

    private final OrderEventsService orderEventsService;

    @KafkaListener(topics = {"order-events"})
    public void onMessage(ConsumerRecord<Integer, String> consumerRecord) throws JsonProcessingException {
        log.info("ConsumerRecord: {} ", consumerRecord);
        orderEventsService.processOrderEvent(consumerRecord);
    }
}

あとは受け取ったイベントのモデルで MySQL に保存しています。

    private void save(OrderEvent orderEvent) {
        orderEvent.getOrder().setOrderEvent(orderEvent);
        orderEventsRepository.save(orderEvent);
        log.info("Successfully Persisted the Order Event {}", orderEvent);
    }

では、 HTTP リクエストしてみましょう。
※サンプルコード(KiyotaTakeshi/kafka-spring-sample)に postman のコレクションを配置しています

もろもろ必要なコンテナを起動します。

$ docker compose up -d

$ docker compose ps                                                             

NAME                COMMAND                  SERVICE             STATUS              PORTS
order-kafka1        "/opt/bitnami/script…"   kafka1              running             0.0.0.0:9092->9092/tcp
order-kafka2        "/opt/bitnami/script…"   kafka2              running             0.0.0.0:9093->9093/tcp
order-kafka3        "/opt/bitnami/script…"   kafka3              running             0.0.0.0:9094->9094/tcp
order-mailhog       "MailHog"                mailhog             running             0.0.0.0:1025->1025/tcp, 0.0.0.0:8025->8025/tcp
order-mysql         "docker-entrypoint.s…"   mysql               running             0.0.0.0:3306->3306/tcp
order-zookeeper     "/opt/bitnami/script…"   zookeeper           running             0.0.0.0:2181->2181/tcp

root project でビルドして executable jar を生成します。
IDE で起動しても問題ないです！

$ export JAVA_HOME=`/usr/libexec/java_home -v 11`

$ java -version                                                                
openjdk version "11.0.11" 2021-04-20 LTS
OpenJDK Runtime Environment Corretto-11.0.11.9.1 (build 11.0.11+9-LTS)
OpenJDK 64-Bit Server VM Corretto-11.0.11.9.1 (build 11.0.11+9-LTS, mixed mode)

$ ./gradlew clean build

$ ls -l order-notification-consumer/build/libs/order-notification-consumer-0.0.1-SNAPSHOT.jar

$ ls -l order-producer/build/libs/order-producer-0.0.1-SNAPSHOT.jar

$ ls -l order-rdb-consumer/build/libs/order-rdb-consumer-0.0.1-SNAPSHOT.jar

topic を作成するため、 order producer から起動します。

$ java -jar order-producer/build/libs/order-producer-0.0.1-SNAPSHOT.jar

そして order notification, order rdb を起動すると準備完了です。

$ java -jar order-notification-consumer/build/libs/order-notification-consumer-0.0.1-SNAPSHOT.jar

$ java -jar order-rdb-consumer/build/libs/order-rdb-consumer-0.0.1-SNAPSHOT.jar

HTTP リクエストをすると、

$ curl --location --request POST 'http://localhost:8080/order-events' \
--header 'Content-Type: application/json' \
--data-raw '{
    "orderEventId": null,
    "order": {
        "id": 43333,
        "name": "チョコモナカジャンボ",
        "price": 145
    }
}'

DB にレコードが保存され、

select * from orders;

/*
+-----+----------+-----+--------------+
|id   |name      |price|order_event_id|
+-----+----------+-----+--------------+
|43333|チョコモナカジャンボ|145  |1             |
+-----+----------+-----+--------------+
*/

チョコモナカジャンボの購入を受け付けたメールが届きました。

イベント駆動型アーキテクチャで複数のアプリが各々の解釈でイベントを処理するイメージが掴めましたね。

では consumer を複数起動してみましょう。複数のシェルで executable jar を実行してみます。
そうすると興味深いログが出力されます。

order-notification-events-group: partitions assigned: [order-events-0] から、
consumer がどの partition と疎通するかがわかります。

# シェル1
$ java -jar order-notification-consumer/build/libs/order-notification-consumer-0.0.1-SNAPSHOT.jar

2021-12-15 18:49:59.734  INFO 70648 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : order-notification-events-group: partitions assigned: [order-events-2]

# シェル2
$ java -jar order-notification-consumer/build/libs/order-notification-consumer-0.0.1-SNAPSHOT.jar

2021-12-15 18:49:59.732  INFO 70918 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : order-notification-events-group: partitions assigned: [order-events-1]

# シェル3
$ java -jar order-notification-consumer/build/libs/order-notification-consumer-0.0.1-SNAPSHOT.jar

2021-12-15 18:49:59.740  INFO 71159 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : order-notification-events-group: partitions assigned: [order-events-0]

この状態でリクエストしてみても、consumer group が同じ consumer を複数起動しているだけなので、
event は1回しか処理されず、バニラモナカジャンボの購入を受け付けたメールも一度しか送信しません。

2021-12-15 18:53:44.635  INFO 70918 --- [ntainer#0-0-C-1] c.k.o.service.OrderEventsService         : orderEvent: OrderEvent(orderEventId=null, orderEventType=NEW, order=Order(id=41111, name=バニラモナカジャンボ, price=145))
2021-12-15 18:53:45.087  INFO 70918 --- [ntainer#0-0-C-1] c.k.o.service.OrderEventsService         : Email sent!!

consumer が1台ダウンしてしまった仮定して、プロセスを1つ止めてみます。
すると、 (Re-)joining group が走って、起動している consumer で partition が割り振られます。

# シェル1
2021-12-15 18:49:17.045  INFO 70648 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-order-notification-events-group-1, groupId=order-notification-events-group] (Re-)joining group

2021-12-15 18:57:54.080  INFO 70648 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : order-notification-events-group: partitions assigned: [order-events-2]

# シェル2
2021-12-15 18:57:54.078  INFO 71159 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : order-notification-events-group: partitions assigned: [order-events-1, order-events-0]

consumer として実装するアプリケーションを水平にスケールすることで、可用性を高める事ができます。
(それぞれの partition に均等にイベントが配信されるのであれば、 partition の数まで consumer を増やせば処理の並列度も上がりシステム全体のスループットも向上すると思います)

最後に

気になっていたイベント駆動アーキテクチャについて、動くものを作ったことで理解が進みました。(知らないものに対する恐怖心が和らぎました)

一方で、今回のメール送信にあたる「通知サービス」のように、
サービスを分割する粒度や設計方法については学習と経験を積む必要があると感じました。

また、 Kafka は broker, consumer などの各コンポーネントのスケーラビリティが高いことがわかりましたが、
イベント駆動アーキテクチャを実現するための技術として、ユースケースに合わせて適切な技術を選んでいくことも必要だと感じました。

参考

ソースコード