Amazon MSKを試したい(構築編)

なかなか使う機会が少ないであろうMSKの構築を行ってみる。

## 手順概要
1 AWS上にMSKをデプロイする
2 IAM権限を設定する
3 設定用のEC2クライアントでブローカーに接続およびTOPIC作成
4 テスト
## 1 MSKのデプロイ
Amazon MSK>クラスター>クラスターの作成からKafkaクラスターを作成する。
今回は下記のようにProvisioned Standardにて構築する。

### クラスター設定
◯クラスターの作成方法：カスタム作成
　デフォルトのクイック作成はVPCもデフォルトのものが使われてしまうのでこちらを選択。
◯クラスタータイプ：プロビジョンド
◯Apache Kafkaバージョン：3.6.0(デフォルト)
　今回はで推奨の設定を利用する
◯ブローカータイプ：Standardブローカー
◯ゾーン数：3(デフォルト)
◯ゾーンあたりのブローカー：1(デフォルト)
◯クラスター設定：デフォルト
![2025-02-24 15.00.22 ap-northeast-1.console.aws.amazon.com 4990333477a4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3561522/3964e165-d836-40f7-8b7b-5d1a92c81711.png)


### ネットワーク設定
VPCやサブネットを指定
![2025-02-24 15.01.16 ap-northeast-1.console.aws.amazon.com 79b663a0758f.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3561522/ca28d719-a372-41bd-910e-53fd21289c23.png)

### セキュリティ設定
◯アクセスコントロール方法：IAMロールベースの認証(デフォルト)
　今回はクライアントをAWS内に持つため最も簡単なIAMを利用した認証とする。
◯クラスターとクライアント間の暗号化：TLS暗号化(デフォルト)
◯クラスター内の暗号化：TLS暗号化(デフォルト)
◯保管時のデータの暗号化：AWSマネージドCMKを使用(デフォルト)
![2025-02-25 20.21.48 ap-northeast-1.console.aws.amazon.com 2969187db299.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3561522/d6474d15-de8b-455a-ab45-90f37f593f45.png)

### モニタリングおよびタグ
必要に応じて設定
![2025-02-24 15.06.05 ap-northeast-1.console.aws.amazon.com 97ccedad9be4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3561522/46f3165c-2244-4001-a678-c362ee6952a1.png)

## 2 IAM権限設定
今回はIAMを使って認証するため下記のようなロールを作成する。
MSKへのアクセスは今回のケースではEC2から行うため下記のようなポリシーからロールを作成し、EC2に対してIAMプロファイルをアタッチする。
アカウントとTOPICは都度書き換える必要がある。
``` json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:AlterCluster",
                "kafka-cluster:DescribeCluster"
            ],
            "Resource": [
                "arn:aws:kafka:ap-northeast-1:Account-ID:cluster/MSKTutorialCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:*Topic*",
                "kafka-cluster:WriteData",
                "kafka-cluster:ReadData"
            ],
            "Resource": [
                "arn:aws:kafka:ap-northeast-1:Account-ID:topic/MSKTutorialCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": [
                "arn:aws:kafka:ap-northeast-1:Account-ID:group/MSKTutorialCluster/*"
            ]
        }
    ]
}
```

## 3 ブローカー接続
MSK接続用のクライアントを作成していく。
資材の用意のためインターネット接続可能なEC2を用意して下記を実施。
①Javaインストール
```
# sudo yum -y install java-11
```
②kafkaユーザを作成して作業ディレクトリを用意
```
# sudo -i
# mkdir /opt/kafka
```
③kafkaとMSK IAM認証用のjarをダウンロード
```
# cd /opt/kafka/
# wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgz
# tar -xzf kafka_2.13-3.6.0.tgz
# cd kafka_2.13-3.6.0/libs/
# wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar
```
④クライアントの設定を追加
```
# cd ../bin/
# touch client.properties
# cat << EOF >> touch 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
EOF
```

⑤TOPIC作成
下記コマンドはそれぞれの環境ごとに修正すること。
エンドポイントは下記の手順で確認できる。
クラスター＞<自身のクラスター名>＞クライアント情報の表示から確認できる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3561522/9dc78992-b8a7-4f9d-aff0-90a5e272e7de.png)
※例はブローカーが2つの構成。
```
./kafka-topics.sh --create --bootstrap-server b-1.msktest.XXXXXX.c3.kafka.ap-northeast-1.amazonaws.com:90XX  --command-config client.properties --replication-factor 3 --partitions 1 --topic MSKTutorialTopic
```

この時点でTOPICの作成まで完了している。
あとは実際に試す必要がある。

## 4 テスト
今回はプロデューサー(書き込み)とコンシューマー(読み込み)の動作を確認するため、2つのセッションでEC2のssh接続を行うかscreen等で切り替えることを推奨する。
①プロデューサー動作確認
下記コマンドはそれぞれの環境ごとに修正すること。
```
# ./kafka-console-producer.sh --broker-list b-1.msktest.XXXXXX.c3.kafka.ap-northeast-1.amazonaws.com:90XX --producer.config client.properties --topic MSKTutorialTopic
```
コマンドが通れば好きにメッセージを書き込むことができる。
1行ごとに1メッセージとして処理されEnterで送信される。
②コンシューマー動作確認
下記コマンドはそれぞれの環境ごとに修正すること。
```
./kafka-console-consumer.sh --bootstrap-server b-1.msktest.XXXXXX.c3.kafka.ap-northeast-1.amazonaws.com:90XX --consumer.config client.properties --topic MSKTutorialTopic --from-beginning
```
正常に動作するしていればプロデューサー側にて入力したメッセージがコンシューマー側でも1行ずつリアルタイムで反映される。


## 次回
次回はDRへのレプリケーションを扱う予定です。