【初心者】Amazon Managed Streaming for Apache Kafka (MSK) / MSK Serverless を使ってみる

# 1. 目的
- 知識0からKafkaの学習をしている。公式チュートリアルを実施して、基本構成やKafka用語についてざっくり理解できるようにする。

# 2. やったこと
- 以下2つの公式チュートリアルを実施し、構築手順や処理の流れを確認する。
    - [Getting Started Using Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html)
        - 通常版(Kafkaクラスタの構成時にインスタンスの数やサイズを指定するタイプ。ECSやAuroraのような感じ)
    - [Getting started using MSK Serverless clusters](https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html)
        - Serverless版(KafkaクラスタのインフラがAWS任せでスケールするタイプ。FargateやAurora Serverlessのような感じ)
- 内容としては以下が含まれる。
    - Kafkaクラスター作成
    - Topicの作成
    - Producerからのメッセージ送信
    - Consumerからのメッセージ取得

# 3. Amazon Managed Streaming for Apache Kafka (MSK) とは(自分の理解)
- Kafkaクラスター(自分で構築する場合、インストール、冗長化設定、バージョンアップなど煩雑)をAWS Managedで提供してくれるサービス。

# 4. 構成図

![構成図.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/d3e2c027-a6a0-aa3a-f293-1942558b1e23.png)



# 5. 手順
基本的にチュートリアルの手順通りに実施する。詰まった点や補足的に実施したコマンドなどを記載する。

## 5.1 事前準備
- 3つのsubnet(全て別AZ)を持つVPCを作成しておく。(通常版のKafkaクラスターは3AZでの冗長が基本構成のため)

## 5.2 [Getting Started Using Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/getting-started.html) (通常版Kafkaクラスター)

### [Step 1: Create an Amazon MSK Cluster](https://docs.aws.amazon.com/msk/latest/developerguide/create-cluster.html) 

- 「Amazon MSK - クラスター - クラスターの作成」　からKafkaクラスターを作成する。以下がデフォルト値から変更しているところ。

    - 作成方法:  「カスタム作成」を選択　(「クイック作成」だとDefaultVPCに作成されてしまう)
    - クラスタータイプ:  「プロビジョンド」を選択　(「プロビジョンド」が通常版、「サーバーレス」がServerless版)
    - ブローカータイプ:  「kafka.t3.small」を選択 (検証用のため一番小さいインスタンスタイプ)
    - VPC: あらかじめ作成した、subnetが3つあるVPCを選択
    - アクセスコントロール方法: 「認証されていないアクセス」を選択 (認証なしでシンプルにインスタンスから読み書きを行うため)
    - 暗号化: 「プレーンテキスト」を選択 (追加の設定をしなくてよいようにするため)


![kafka01a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/5727de76-4590-167e-4896-d906bf5dd85b.png)

![kafka02a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/5920fefe-4f38-7e4d-69af-4cb0686ff0a1.png)
![kafka03a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/ab8cac5c-7ec6-2954-bb6e-f3bea4b0c741.png)
![kafka04a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/b270ebfb-a7b3-116c-a210-2f484e2969d7.png)
![kafka05a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/c3e876f9-c4f0-56dd-a773-239c681e3e2b.png)
![kafka06a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/2eaf4fcb-02b3-81b0-8e5b-46ab3e7bf83c.png)
![kafka07a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/164e846f-b737-eab8-b382-465c12fa9f96.png)
![kafka08a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/692095d2-65aa-6a69-371a-896d031cf956.png)
![kafka09a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/77f5008e-fcf8-dd30-1ddf-b43a6d613188.png)

### [Step 2: Create a Client Machine](https://docs.aws.amazon.com/msk/latest/developerguide/create-client-machine.html)
- Kafkaクラスターと同一VPC内にAmazon Linux 2 のインスタンスを起動する。負荷をかける作業は行わないため、インスタンスタイプはt2.microにする。

### [Step 3: Create a Topic](https://docs.aws.amazon.com/msk/latest/developerguide/create-topic.html)
- Kafkaのアプリケーションをインストールし、topicを作成する。

```
[ec2-user@ip-10-0-0-54 ~]$ sudo yum install java-1.8.0
[ec2-user@ip-10-0-0-54 ~]$ wget https://archive.apache.org/dist/kafka/2.6.2/kafka_2.12-2.6.2.tgz
[ec2-user@ip-10-0-0-54 ~]$ tar -xzf kafka_2.12-2.6.2.tgz

# Topicの作成
[ec2-user@ip-10-0-0-54 ~]$ bin/kafka-topics.sh --create --zookeeper [ZookeeperConnectString] --replication-factor 3 --partitions 1 --topic MSKTutorialTopic
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
Created topic MSKTutorialTopic.

# Topicのリスト表示
[ec2-user@ip-10-0-0-54 kafka_2.12-2.6.2]$ bin/kafka-topics.sh --list --zookeeper [ZookeeperConnectString]
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
MSKTutorialTopic
__amazon_msk_canary
__consumer_offsets
[ec2-user@ip-10-0-0-54 kafka_2.12-2.6.2]$
```
- [ZookeeperConnectString]は、マネージメントコンソールの以下の箇所を参照する。

![kafka10b.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/ba5ba20b-7065-6c17-582e-62b86b931fef.png)

### [Step 4: Produce and Consume Data](https://docs.aws.amazon.com/msk/latest/developerguide/produce-consume.html)

- 通信方式をPLAINTEXTに設定する。(Kafkaクラスター作成時も暗号化なしにしているため)
```
[ec2-user@ip-10-0-0-54 bin]$ cat client.properties
security.protocol=PLAINTEXT
```

- bootstrap-server を指定して、Producerとしてメッセージの送信を行う。(「aaaaa」などの文字列をメッセージとして手入力する)
```
[ec2-user@ip-10-0-0-54 bin]$ ./kafka-console-producer.sh --broker-list [endpoint] --producer.config client.properties --topic MSKTutorialTopic
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
>aaaaa
>bbbbb
>ccccc
>^C
```
- EC2インスタンスでもう1つ別のターミナルを開き、Consumerとしてメッセージの取得を行う。Producer側で入力した文字列がリアルタイムに取得できる。

```
[ec2-user@ip-10-0-0-54 bin]$ ./kafka-console-consumer.sh --bootstrap-server [endpoint] --consumer.config client.properties --topic MSKTutorialTopic --from-beginning
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
aaaaa
bbbbb
ccccc
^CProcessed a total of 3 messages
[ec2-user@ip-10-0-0-54 bin]$
```
- broker-list/bootstrap-serverのendpointは、マネージメントコンソールの以下の場所を参照。

![kafka10a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/f3881563-f518-751b-abff-8e0d7ffed13d.png)

## 5.3 [Getting started using MSK Serverless clusters](https://docs.aws.amazon.com/msk/latest/developerguide/serverless-getting-started.html) (Serverless版Kafkaクラスター)

### [Step 1: Create an MSK Serverless cluster](https://docs.aws.amazon.com/msk/latest/developerguide/create-serverless-cluster.html)

- 「Amazon MSK - クラスター - クラスターの作成」　からKafkaクラスターを作成する。以下がデフォルト値から変更しているところ。
    - 作成方法:  「カスタム作成」を選択　(「クイック作成」だとDefaultVPCに作成されてしまう)
    - クラスタータイプ:  「サーバーレス」を選択　(「プロビジョンド」が通常版、「サーバーレス」がServerless版)
    - VPC: あらかじめ作成したVPCを使用する。Subnetは2つのみ選択する。

![kafka12a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/0582b0ae-7dfd-de28-1fc7-73e309253b9b.png)
![kafka13a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/8879f123-b208-1e0e-54a6-993e961c86ec.png)
![kafka14a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/2fb14be2-2647-7ce0-5673-1e5e2380d096.png)
![kafka15a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/27019b4c-36dc-1a49-13f0-531c492458b8.png)
![kafka16a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/e180abd2-65df-1888-0276-75352a7973c6.png)

### [Step 2: Create an IAM role](https://docs.aws.amazon.com/msk/latest/developerguide/create-iam-role.html)

- EC2インスタンス(Producer/Consumer)からKafkaクラスターへアクセスするためのIAMポリシー/IAMロールを作成する。
- 今回は作業簡略化のため、チュートリアルのIAMポリシー/IAMロールは作成せず、既存のPowerUser権限のあるIAMロールを使用する。

### [Step 3: Create a client machine](https://docs.aws.amazon.com/msk/latest/developerguide/create-serverless-cluster-client.html)

- Kafkaクラスターと同一VPC内にAmazon Linux 2 のインスタンスを起動する。負荷をかける作業は行わないため、インスタンスタイプはt2.microにする。
- PowerUser権限のあるIAMロールをインスタンスに付与する。
- アプリケーションのインストール、設定を行う。

```
[ec2-user@ip-10-0-0-130 ~]$ sudo yum -y install java-11
[ec2-user@ip-10-0-0-130 ~]$ wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
[ec2-user@ip-10-0-0-130 ~]$ tar -xzf kafka_2.12-2.8.1.tgz
[ec2-user@ip-10-0-0-130 libs]$ wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar
[ec2-user@ip-10-0-0-130 bin]$ cat client.properties
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
```

### [Step 4: Create an Apache Kafka topic](https://docs.aws.amazon.com/msk/latest/developerguide/msk-serverless-create-topic.html)

- Topicを作成する。

```
# Topicを作成
[ec2-user@ip-10-0-0-130 bin]$ ./kafka-topics.sh --bootstrap-server [endpoint] --command-config client.properties --create --topic msk-serverless-tutorial --partitions 6
Created topic msk-serverless-tutorial.

# Topicのリスト表示
[ec2-user@ip-10-0-0-130 bin]$ ./kafka-topics.sh --bootstrap-server [endpoint] --command-config client.properties --list
msk-serverless-tutorial

# (参考) IAM権限が不足している場合のTopic作成時のエラー
[ec2-user@ip-10-0-0-130 bin]$ ./kafka-topics.sh --bootstrap-server [endpoint] --command-config client.properties --create --topic msk-serverless-tutorial --partitions 6
Error while executing topic command : Authorization failed.
[2022-06-20 01:10:42,909] ERROR org.apache.kafka.common.errors.TopicAuthorizationException: Authorization failed.
 (kafka.admin.TopicCommand$)

```

- bootstrap-serverのendpointは、マネージメントコンソールの以下を参照する。

![kafka17a.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/189302/439e6aef-182a-8522-8531-ca4fd662bc94.png)


### [Step 5: Produce and consume data](https://docs.aws.amazon.com/msk/latest/developerguide/msk-serverless-produce-consume.html)

- Producerとしてメッセージの送信を行う。
```
[ec2-user@ip-10-0-0-130 bin]$ ./kafka-console-producer.sh --broker-list [endpoint] --producer.config client.properties --topic msk-serverless-tutorial
>11111
>22222
>33333
>^C
```

- ターミナルをもう1個開き、Consumerとしてメッセージの受信を行う。
```
[ec2-user@ip-10-0-0-130 bin]$ ./kafka-console-consumer.sh --bootstrap-server [endpoint] --consumer.config client.properties --topic msk-serverless-tutorial --from-beginning
11111
22222
33333
^CProcessed a total of 3 messages
```

- broker-list/bootstrap-server のendpointの値は、Topic作成時のものと同じものを使用する。

# 6. 所感
- Producer --> Kafkaクラスター --> Consumer とメッセージを伝送する、Kafkaの最も基本的なところの動作確認ができた。引き続き、使いい方など深堀していきたい。