LibertyからEvent Streams(Kafka)を利用してみた

# はじめに
この記事は[WebSphere 25周年、DB2/MQ30周年 アニバーサリー記念 Qiita記事投稿キャンペーン](https://community.ibm.com/community/user/wasdevops/blogs/kaori-asada/2023/03/16/qiita?CommunityKey=d6c93aa2-6e10-48da-96dc-3831da8ee185)の参加記事です。


Libertyから、Event Streams(Kafka)へメッセージを送受信する簡単なアプリを作成してみました。

# 行ったこと
+ IBM Cloud 無料版Event Streamsを利用
+ Spring Bootを利用したWebアプリを作成
+ ローカル環境のLibertyからEvent Streamsに接続

## IBM Cloud 無料版Event Streamsを利用
IBM Cloudにあるフル・マネージドの Kafka サービスであるEvent Streamsの無料のライト・プランを利用します。ライト・プランはお試し用で、リージョンがダラスに限られる、Partition1個、Kafka Connect, Streamsが利用できないなど[制約事項](https://cloud.ibm.com/docs/EventStreams?topic=EventStreams-plan_choose&locale=ja)があります。
<img width="1000" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/f6f091c2-64c0-28a3-9b83-41687b043bee.png">

プロビジョニング後はすぐに利用することができます。
トピック "libertyTopic" を作成します。メッセージ保存期間を設定することができます。ライト・プランは、Partitionは1個です。
<img width="300" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/e62dadbc-06d4-ac0c-280d-95be29cbb94c.png">

サービス資格情報を作成します。表右のコピーをクリックして内容をテキスト・ファイルにコピーします。
<img width="500" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/fc525b18-a293-ae6a-05e6-15bedfc0b191.png">

接続に必要な情報になります。
```
{
  "api_key": "xxxxxxxxxxxxxxxxxxxxx",
  "apikey": "xxxxxxxxxxxxxxxxxxxxx",
  "bootstrap_endpoints": "broker-xxxx.ibm.com:9093,broker-yyyy.ibm.com:9093,broker-zzzz.ibm.com:9093,broker-aaaa.ibm.com:9093,broker-bbbb.ibm.com:9093,broker-cccc.ibm.com:9093",
  "iam_apikey_description": "Auto-generated for key xxxxxxxxx,
  "iam_apikey_name": "cre1",
  "iam_role_crn": "xxxxxxxxx",
  "iam_serviceid_crn": "xxxxxxxxxxx",
  "instance_id": "aaaaaa-aaaa-bbbb-cccc-xxxxxxxxxx",
  "kafka_admin_url": "https://xxxxxxxxxxx.ibm.com",
  "kafka_brokers_sasl": [
    "broker-xxxx.ibm.com:9093",
    "broker-yyyy.ibm.com:9093",
    "broker-zzzz.ibm.com:9093",
    "broker-aaaa.ibm.com:9093",
    "broker-bbbb.ibm.com:9093",
    "broker-cccc.ibm.com:9093"
  ],
  "kafka_http_url": "https://xxxxxxxxxxx.ibm.com",
  "password": "XXXXXXXXXXXXXXXXXXXXXX",
  "user": "token"
}
```


## Spring Bootを利用したWebアプリを作成
Spring Bootを利用した、以下のような簡単なWebアプリを作成します。
+ index.htmlからEvent Streamsにメッセージを送信
+ 送信されたメッセージは、KafkaListnerでログに出力

pom.xmlへ、Spring BootとSpring Kafkaを依存関係に追加します。

```pom.xml
-----略------
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-thymeleaf</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>
```

application.propertiesに先程コピーしたEvent Streamsの接続情報を記述します。Event Streamsに接続するにはsecurity設定が必要です。
```application.properties
# Kafkaブローカーリスト
spring.kafka.bootstrap-servers=broker-xxxx.ibm.com:9093,broker-yyyy.ibm.com:9093,broker-zzzz.ibm.com:9093,broker-aaaa.ibm.com:9093,broker-bbbb.ibm.com:9093,broker-cccc.ibm.com:9093

# Producer設定
spring.kafka.producer.acks=1
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

#Consumer設定
spring.kafka.consumer.group-id=group_id
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer


## Security options
spring.kafka.jaas.enabled=true
spring.kafka.properties.security.protocol=SASL_SSL
spring.kafka.properties.sasl.mechanism=PLAIN
spring.kafka.properties.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="token" password="XXXXXXXXXXXXXXXXXXXXXX";
```

メッセージ送信、送った内容を表示するHTMLファイルを2つ用意します。
```index.html
<!DOCTYPE html><html>
<head>
    <meta charset="UTF-8">
    <title>Kafka Sender</title>
</head>
<body>
	<h1>Hello Event Streams!</h1>

    <form method="post" action="/EventApp/send">
        <label for="message">Message:</label>
        <input type="text" name="message" id="message">
        <button type="submit">Send</button>
    </form>
</body></html>
```
```result.html
<!DOCTYPE html>
<html xmlns:th="http://www.thymeleaf.org">
<head>
    <meta charset="UTF-8">
    <title>Kafka Sender</title>
</head>
<body>
    <p>Message sent: [[${message}]]</p> 
</body>
</html>
```

JavaクラスはControllerクラスを1つ作成します。
+ コンテキスト・ルートへのアクセスをindex.htmlへ
+ Spring KafkaのKafkaTemplateを利用してメッセージを送信
+ spring KafkaのKafkaListenrを利用してメッセージを受信し、SystemOutに出力

```MessageController.java
package com.example.demo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.servlet.ModelAndView;

@Controller
public class MessageController {

	@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

	@RequestMapping(value = "/", method = RequestMethod.GET)
    public ModelAndView index(final ModelAndView mv) {
		mv.setViewName("index");
        return mv;
    }

    @RequestMapping(value = "/send", method = RequestMethod.POST)
    public ModelAndView send(@RequestParam("message") final String message, final ModelAndView mv) {
    	kafkaTemplate.send("libertyTopic", message);
    	mv.setViewName("result");
        mv.addObject("message", message);
        return mv;
    }

    @KafkaListener(topics = "libertyTopic", groupId = "")
    public void listen(String message) {
        System.out.println("Received message: " + message);
    }    
}
```


## ローカル環境のLibertyからEvent Streamsに接続
作成したアプリをLibertyにデプロイして稼働してみます。
<img width="400" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/5e59628d-f224-edb7-f0c7-6535c62b3ffd.png">

メッセージを送信すると、結果画面が表示されます。
<img width="400" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/46c47be4-aeb7-3e51-1304-f464b6f838df.png">

複数メッセージを送信しログを確認すると以下のようにメッセージが受信されているのも確認できました。
<img width="500" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/121209/ef38e069-ea0e-a8af-522c-1505ec8271fb.png">

# まとめ
LibertyからEvent Streams(Kafka)を利用する簡単なアプリケーションを作成しました。