Event Streams と MQ連携 ～kafka connectorを使用してMQメッセージをKafkaに収集～

# はじめに

[IBM Event Streams](https://www.ibm.com/jp-ja/cloud/event-streams)は[Apache Kafka](https://kafka.apache.org/)をベースにしたイベント・ストリームプラットフォームです。Event Streamsでは様々なサービスとデータ連携するための[Connector](https://ibm.github.io/event-streams/connectors/)を提供されています。今回は、MQ source connectorを使用して、MQのキューにメッセージが到着したことをイベントとして検知し、Event Streamsのtopicに収集する方法を紹介します。

# 1. 事前準備

## 1.1 MQ環境

メッセージの収集元となるMQを設定します。
今回は、IBM Cloudのフリー環境を使用しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/ef53d5c0-31ee-b594-5638-040758ed50cd.png)

接続用の資格情報も登録しておきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/f7fd1bfd-dfab-494a-3519-dd8dbb412fc1.png)


## 1.2 Event Streams環境

メッセージの収集先となるEvent Streamsを設定します。
今回は、IBM CloudにCloud Pak for Integration 2022.4.1を導入して使用しています。

Event Streamsインスタンスに接続すると、以下のようなコンソールが表示されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/67e8a07d-b66d-21f4-39f0-6818eeca571f.png)


### 1.2.1 トピック作成

ホーム画面の「トピックの作成」からデータ収集に使用するトピックを作成します。
今回は「user」というトピックを作成しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/92c2a12a-ad6c-df41-aa22-bfd8d6eaf91f.png)


### 1.2.2 クラスター接続設定

ホーム画面の「このクラスターに接続」から接続に使用する資格情報を確認・登録します。
Kafkaリスナーおよび資格情報欄で内部に対する「TLS資格情報の生成」を実施します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/f87f48ce-70b0-7e4d-e1c5-f602f4666746.png)

以下のような流れで設定していきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/602fd84a-a9e6-efd9-3ca8-abb82df6e94d.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/ec6fb28d-b073-9e0d-82ac-c8aff59793e5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/2fa3e2e2-435d-c8fb-c88a-92a79f2e667b.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/6fabf11f-7068-b1ab-6190-cec48cd8a01c.png)


ここで設定した証明書は、OCPコンソールからも確認ができます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/8171bb33-5211-f386-26ac-1d1c5f7a93c1.png)

# 2. コネクター環境セットアップ

「ツールボックス」を選択すると、Event Streamsに接続するための様々なツールが示されています。
今回は、コネクターを使用していきます。以下の順序で実施していきます。

 2.1 Kafka Connect環境のセットアップ
 2.2 Kafka Connect環境へのコネクターの追加
 2.3 Kafka Connectとコネクターを開始

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/bb20a991-dd09-6475-d6ee-3a46a86e0a97.png)



## 2.1. Kafka Connect環境のセットアップ

ここで示されている手順に沿って準備していきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/0b861d52-7fe4-4d5a-9ef2-54b6c321dc88.png)

Kafka Connect ZIPをダウンロードすると以下のような中身になっています。

kafkaconnect:　
　Dockerfile
　kafka-connect.yaml
　my-plugins


まずは、kafka-connect.yamlに今回の環境用の認証情報を設定していきます。
環境に合わせて以下の値を設定していきます。証明書の詳細はOCPコンソールから確認ができます。
・bootstrapServers
・image
・tls
・authentication

```yaml:kafka-connect.yaml
apiVersion: eventstreams.ibm.com/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    eventstreams.ibm.com/use-connector-resources: "true"
spec:
  replicas: 1
  bootstrapServers: es-demo-kafka-bootstrap.cp4i.svc:9093
  image: image-registry.openshift-image-registry.svc:5000/cp4i/my-connect-cluster-image:latest
  template:
    pod:
      imagePullSecrets: []
      metadata:
        annotations:
          eventstreams.production.type: CloudPakForIntegrationNonProduction
          productID: 2a79e49111f44ec3acd89608e56138f5
          productName: IBM Event Streams for Non Production
          productVersion: 11.1.3
          productMetric: VIRTUAL_PROCESSOR_CORE
          productChargedContainers: my-connect-cluster-connect
          cloudpakId: c8b82d189e7545f0892db9ef2731b90d
          cloudpakName: IBM Cloud Pak for Integration
          cloudpakVersion: 2022.4.1
          productCloudpakRatio: "2:1"
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
  tls:
    trustedCertificates:
      - secretName: es-demo-cluster-ca-cert
        certificate: ca.crt
  authentication:
    type: tls
    certificateAndKey:
        certificate: user.crt
        key: user.key
        secretName: kafka-user
```

## 2.2 Kafka Connect環境へのコネクターの追加

次にmy-pluginsフォルダーに今回使用したいConnector用のプラグインファイルを配置します。
今回は、mq-source connectorを使用しますので、[こちらのGithub](https://github.com/ibm-messaging/kafka-connect-mq-source)で提供される情報をもとにjarファイルを作成し、配置します。


これで準備が終わりましたので、これらのファイルをDockerコンテナーとしてビルドし、OCP環境にPushしていきます。

```
>docker login cp.icr.io -u cp -p ***
WARNING! Using --password via the CLI is insecure. Use --password-stdin.
Login Succeeded

# dockerイメージビルド
>docker build -t my-connect-cluster-image:latest C:\kafkaconnect
[+] Building 6.3s (8/8) FINISHED
 => [internal] load build definition from Dockerfile                                                0.0s
 => => transferring dockerfile: 134B                                                                0.0s
 => [internal] load .dockerignore                                                                   0.0s
 => => transferring context: 2B                                                                     0.0s
 => [internal] load metadata for cp.icr.io/cp/ibm-eventstreams-kafka:11.1.3                         5.9s
 => [auth] cp/ibm-eventstreams-kafka:pull token for cp.icr.io                                       0.0s
 => [internal] load build context                                                                   0.2s
 => => transferring context: 13.54MB                                                                0.1s
 => CACHED [1/2] FROM cp.icr.io/cp/ibm-eventstreams-kafka:11.1.3@sha256:c914ec93fd1131a429f08b538e  0.0s
 => [2/2] COPY ./my-plugins/ /opt/kafka/plugins/                                                    0.1s
 => exporting to image                                                                              0.1s
 => => exporting layers                                                                             0.1s
 => => writing image sha256:c09376a99a08679869c2a24524bd9fd20a08730e88655868b202cd3b9c38029c        0.0s
 => => naming to docker.io/library/my-connect-cluster-image:latest                                  0.0s

# OCP環境レジストリーパスでタグ付け
> docker tag my-connect-cluster-image:latest default-route-openshift-image-registry.***.us-east.containers.appdomain.cloud/cp4i/my-connect-cluster-image:latest

# OCP環境にログイン
> docker login -u *** -p sha256~*** default-route-openshift-image-registry.***.us-east.containers.appdomain.cloudWARNING! Using --password via the CLI is insecure. Use --password-stdin.
Login Succeeded

# OCP環境のイメージレジストリーにプッシュ
> docker push default-route-openshift-image-registry.***.us-east.containers.appdomain.cloud/cp4i/my-connect-cluster-image:latest
The push refers to repository [default-route-openshift-image-registry.***.us-east.containers.appdomain.cloud/cp4i/my-connect-cluster-image]
9bb3680d1518: Pushed
afbaf216b515: Pushed
604ea310d3b8: Pushed
c885a267df03: Pushed
d41c9d214c08: Pushed
1d4211b8047a: Pushed
b6b5e54a9d61: Pushed
5b0e52ab72d6: Pushed
b9c8807c07f0: Pushed
abd5331ad63d: Pushed
1d36f4b724f1: Pushed
28647eedfe80: Pushed
28d152ef0c4c: Pushed
a5a3a5497180: Pushed
b642d9aa2f45: Pushed
8b86f8fd3a7d: Pushed
99c6ecde084d: Pushed
28a458282a1e: Pushed
b66994f69e49: Pushed
f536fae8f746: Pushed
cd1cd3ddb935: Pushed
61094c2c6010: Pushed
1ff1dbf9158b: Pushed
f922a018877b: Pushed
52cbfc36b72b: Pushed
latest: digest: sha256:ba6fd8d172c8e379c873d6dc9a811f8d390a8f0047923099ba211d06bd7f632e size: 5555
```

##  2.3 Kafka Connectとコネクターを開始

### 2.3.1 kafka connectビルド

以下の手順でKafkaConnectを開始します。

```
> oc apply -f kafka-connect.yaml
kafkaconnect.eventstreams.ibm.com/my-connect-cluster created
```

OCP環境からPodが開始され、エラーが出ていないことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/d56acc39-4840-73da-1424-8727168d5f79.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/f9578323-3c63-513c-1474-dbd5d1caafa0.png)

### 2.3.2 Kafka Connectorの開始

GitHubで提供しているConnectorのリソースファイルを接続先の環境に合わせて編集し、開始します。


```
> cat mq-source.yaml
apiVersion: eventstreams.ibm.com/v1beta2
kind: KafkaConnector
metadata:
  name: mq-source
  labels:
    eventstreams.ibm.com/cluster: my-connect-cluster
spec:
  class: com.ibm.eventstreams.connect.mqsource.MQSourceConnector
  tasksMax: 1
  config:
    mq.queue.manager: "USER.MGMT"
    mq.connection.name.list: "user***.appdomain.cloud(30684)"
    mq.channel.name: "CLOUD.APP.SVRCONN"
    mq.queue: "DEV.QUEUE.1"
    mq.user.name: "showroomapp"
    mq.password: "***"
    topic: "user"
    mq.connection.mode: client
    mq.record.builder: com.ibm.eventstreams.connect.mqsource.builders.DefaultRecordBuilder
    key.converter: org.apache.kafka.connect.storage.StringConverter
    value.converter: org.apache.kafka.connect.converters.ByteArrayConverter

> oc apply -f mq-source.yaml
kafkaconnector.eventstreams.ibm.com/mq-source created

```

OCPコンソールからkafkaconnectorのリソースが作成されていることを確認できます。
特にエラーが出ていないことを確認します。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/ba218f5c-c805-7361-761d-d830a39fbcb5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/eb725e76-0858-df53-0d0d-f10756617029.png)


また、Kafka ConnectのPodログにMQとの接続が確立さえたメッセージが表示されることも確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/5b325638-9b3a-2c53-1947-f86ded324ace.png)

# 3. 動作確認

以上でMQメッセージをKafkaに連携する準備が整いました。

MQ Consoleから今回接続しているキューに対してメッセージを登録します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/8e37d308-5b9c-2060-0ff5-63ebfa3485c7.png)

Event StreamsのConsoleから対象のトピックを確認すると、メッセージを受信できていることが確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/403443/dc9d9f47-38f0-e1bb-0d7a-29a97d499ed0.png)



以上です。