はじめに

Amazon Managed Streaming for Apache Kafka（以下、Amazon MSK）の設定方法や挙動を確認した記録です。

これまで、OSSのApache Kafka、IBM Event streams、Red Hat Streams for Apache kafka、Confluent Platform などを使用したシステム構築や検証に携わってきましたが、AWSを採用されているお客様が多い中で、Amazon MSKが有力な選択肢となり得るのかを確認することを目的に確認しました。

Kafkaを多少知っている＆AWSフレンドリーな自分としては、 かなり「アリ」 だと感じています。
AWSならではの設定や構築にあたってのつまづきポイントもありましたが、今回のトライで基本的な方法を確立できました。ここで確認した方法をベースに、さらに知見を広げてお客様や誰かの何かのお役に立てたらと思います。

前提

Apache Kafkaのバージョン：3.5.1
Zookeeper構成（KRaft構成ではない）
トライした時期：2024/9/13-17
AWS環境は個人利用のもの（自費）

実施した概要

① Amazon MSKのクラスターを構築する（本稿で紹介）
② AWS外からJavaプログラムを使用して以下を実施（別途紹介）
- シンプルな文字列形式のメッセージのProduce/Consume
- AWS Glue Schema Registryの設定
- Avro形式のメッセージのProduce/Consume
③ AWS外からIBM Infosphere Replication CDCを使用して以下を実施（別途紹介）
- JSON形式でキャプチャー対象テーブルのデータをAmazon MSKに送信
- Avro形式でキャプチャー対象テーブルのデータをAmazon MSKに送信

Amazon MSKのクラスターを構築する

タイトルにある「できるだけ普通のKafkaっぽく使えるように」は、Kafkaを知っている人にとっての導入・設定のハードルを下げたい、製品・サービス選定時の先入観をなくしたいという意図です。

Amazon MSKならではの設定は、見たところ、セキュリティや運用効率を高めるものになっています。しかしながら、ライトに始めたい方や簡易に製品・サービスを見比べたい人には「面倒くささ」のようにも感じられると思いますので、はじめの一歩として、シンプルな用途での構築方法を確認していきます。

クラスターを作成・設定する

クラスターの作成を選択する
クラスターの作成方法を選択する
- ここでは「クイック作成」ではなく「カスタム作成」を選択する
- 「クイック作成」だとクラスタータイプを「サーバーレス」「プロビジョンド」のどちらにしても、クラスターを配置するVPCがデフォルトVPCになるため、基本的には「カスタム作成」になるかと思う
  - （クイック作成でサーバーレスの場合）
  - （クイック作成でプロビジョンドの場合）
  - デフォルトVPCがない場合は自動的にVPCが追加になる
クラスターの設定内容を入力していく
- ブローカータイプで指定するインスタンスのサイズは今回は試行用として最も小さいものを選択
- ブローカーやストレージの構成も指定
ネットワークの設定内容を入力していく
- クラスターを配置するVPCを指定
- ゾーン（AZ）を指定、パブリックアクセスを許可する場合はパブリックサブネットを指定する必要がある
- 但し、パブリックアクセスはクラスター作成時には設定できず、クラスター作成後に設定変更する形で実現するものとなっている
- セキュリティグループを指定（後でも修正が可能）
- クラスターに対するインバウンド／アウトバウンド通信をあらかじめ整理しておく必要がある
セキュリティの設定内容を入力していく
- パブリックアクセスを許可する場合はいずれかの認証方法を有効化する必要がある
- 今回はパブリックアクセスを許可し、さらに従来のKafkaっぽい設定とするためにIAMロールベースの認証は無効としSASL/SCLAM認証のみ有効とする
- アクセスコントロール方法でいずれかの認証を有効化した場合、TLS暗号化が有効になる
- なお、後の確認でわかったことだが、ここでのTLS通信に使用されるのはパブリックCAであるAmazon Root CAである
モニタリングおよびタグの設定内容を入力していく
- 今回は試行なのでここはもうデフォルトでいく
クラスターの設定内容のおさらい

クラスターの認証設定をする

公式ドキュメントの日本語訳ページのままだと問題がある箇所があります。
シークレット名が「Amazon MSK_」で始まる必要があると書かれていますが、
「AmazonMSK_」とする必要があります（半角スペースは不可）。

まずはAWS KMSキーを作成する
- キーを設定
- ラベルを追加
- キーの管理アクセス許可を定義
- キーの設定内容のおさらい（キーポリシーは編集しなくてOK）
AWS Secrets Managerでシークレットを作成する
- 「その他のシークレットのタイプ」を選択
- キー/値のペアは以下の通り入力する（キーは以下の通りで、値は任意です）
```
{
  "username": "my-user",
  "password": "my-password"
}
```
- シークレットを設定（シークレットの名前の先頭は「Amazon_MSK_」とするのが必須です）
- ローテーションを設定
- シークレットの設定内容のおさらい
シークレットをクラスターに関連づける
- シークレットの選択
- 先ほど作成したシークレットが選べるようになっているので選択
- シークレットを選択したら「シークレットの関連づけ」を選択
- シークレットの関連付け完了

パグリックアクセスの設定をする

パブリックアクセスとは、VPC外からのAmazon MSKクラスターへのアクセスのことです。
セキュリティ上の理由から、Amazon MSKクラスターの作成中にパブリックアクセスを有効にすることはできず、作成済みのクラスターを更新する形で設定できるようになっています。
また、これもセキュリティ上の理由からですが、パブリックアクセスを有効化するには以下の様な条件があります。

クラスターに関連づけられているサブネットがパブリックサブネットであること
クラスターのアクセスコントロール設定でSASL/IAM、SASL/SCRAM、mTLSのいずれかの認証方法が設定されていること（今回はSASL/SCRAMのみ選択しています）
クラスター内の暗号化が有効化されていること
ブローカーとクライアント間のプレーンテキストでのアクセスが無効化されていること
クラスターのアクセスコントロール設定でSASL/SCRAMまたはmTLS認証を採用している場合、Kafka ACLsの設定が適切に設定されていること（具体的には、クラスターの設定でallow.everyone.if.no.acl.foundがfalseになっていること）

Kafka ACLsはKafka クラスター上のリソース（トピック、消費者グループ、ブローカー、Zookeeper など）へのアクセスを制御するための仕組みです。
allow.everyone.if.no.acl.foundがtrueの場合、ACLが存在しないリソースに対して誰でもアクセスができます。
allow.everyone.if.no.acl.foundがfalseの場合、ACLが存在しないリソースに対してアクセスが拒否されます。例えば、誰がどのトピックにアクセスできる、といったACLがないとトピックへのアクセスができません。

以下はAmazon MSKクラスターをパブリックアクセスができるようにしたときの手順です。

クラスター設定の変更
- クラスター表示画面でプロパティのタブを選択し、クラスター設定の「編集」を選択する
- 「構成の作成」を選択する
- 構成の作成の画面でallow.everyone.if.no.acl.found=falseを設定する（下記画面上の14行目）
- 構成が作成されたことの確認
- クラスター表示画面でプロパティのタブを選択し、クラスター設定の「編集」を選択した先で「クラスター設定」「設定リビジョン」を入力して「変更を保存」を選択する
- クラスター設定変更が開始したことを確認
バブリックアクセス設定
- クラスター表示画面でプロパティのタブを選択し、ネットワーク設定から「パブリックアクセスを編集」を選択する
- 「有効にする」をチェックして「変更を保存」を選択する
- パブリックアクセス設定が開始したことを確認
- パブリックアクセスが「オン」になったことが確認できる

Amazon MSKクラスターに対してパブリックアクセスを利用して接続する接続元のIPアドレスまたはセキュリティグループとポートが、Amazon MSKクラスターに関連づけているセキュリティグループのインバウンドルールに設定されている必要があります。
セキュリティの都合上、こちらではセキュリティグループの設定内容は載せていませんが、必要な設定・手順ですのでご注意ください。

クラスターへの接続情報の確認

パブリックエンドポイントを確認する
- 「クライアント情報の表示」を選択する
- 各種エンドポイントの確認ができる

ブートストラップサーバーのパブリックエンドポイント（参考）

b-2-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,
b-3-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,
b-1-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196

Apache Zookeeper接続のエンドポイントは、Amazon MSKクラスターと同じVPC内からの接続に限定されます。
Kafka ACLzの設定に必要なエンドポイントなので、これも後の手順で必要になります。

VPC外からAmazon MSKクラスターに接続する（Kafkaユーティリティを使用したProduce／Consume）

Kafka ACLsの設定

Kafka ACLsの設定は、構築したAmazon MSKクラスターと同VPCに設置したEC2からZookeeperに対して実施することで実施可能です。
実施内容は以下の通りです。
※ いろいろ試しましたが、この方法で上手く行った、というもので他にも方法はあるかもしれません。

構築したAmazon MSKクラスターと同VPCに設置したEC2上で、Amazon MSKクラスターと同バージョンのApache Kafkaのモジュールをダウンロード

# wgetのインストール
yum install wget
# Apache Kafkaのモジュールをwget
wget https://archive.apache.org/dist/kafka/3.5.1/kafka_2.13-3.5.1.tgz
# Apache Kafkaのモジュールを解凍
tar -zxvf kafka_2.13-3.5.1.tgz

Javaをインストール

yum install java

アクセス権の付与（my-userに全トピック、全グループへのフルアクセス権限を与える）

アクセス権の付与（zookeeper接続）

./kafka-acls.sh \ 
  --authorizer-properties zookeeper.connect=z-1.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:2181 \
  --add \
  --operation All \
  --group=* \
  --topic '*' \
  --allow-principal "User:my-user"

設定したアクセス権の確認

アクセス権確認（zookeeper接続）

./kafka-acls.sh \
  --authorizer-properties zookeeper.connect=z-1.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:2181 \
  --list

設定ファイルを作成する

Amazon MSKクラスターとの認証方法で指定したSASL/SCRAMの認証情報を記述したプロパティファイルを用意する

vi /mskconfig-ssk.properties

プロパティファイルの内容

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="my-user" password="my-password";

kafka-console-producerを使用したメッセージのProduce

トピックの作成

kafka-topics.shを使用したトピックの作成

./kafka-topics.sh \
  --bootstrap-server b-2-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-3-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-1-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196 \
  --create \
  --topic my-test-topic \
  --replication-factor 3 \
  --partitions 1 \
  --command-config /mskconfig-ssk.properties

auto.create.topics.enableがtrueの場合、メッセージをProduceしたときに指定した名称のトピックがない場合は、指定した名称のトピック名が自動で作成されますが、falseの場合は自動で作成されずエラーになります。
Amazon MSKクラスターのデフォルトではfalseとなっていたので、ここではProduceする前にトピックを作成しています。

トピックのリスト

kafka-topics.shを使用したトピックのリスト表示

./kafka-topics.sh \
  --bootstrap-server b-2-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-3-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-1-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196 \
  --command-config /mskconfig-ssk.properties \
  --list

作成したmy-test-topicというトピックが表示されました。

メッセージのProduce

kafka-console-producer.shを使用したメッセージのProduce

./kafka-console-producer.sh \
  --broker-list b-2-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-3-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-1-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196 \
  --topic my-test-topic \
  --producer.config /mskconfig-ssk.properties

kafka-console-producerを使用したメッセージのConsume

メッセージのConsume

kafka-console-consumer.shを使用したメッセージのConsume

./kafka-console-consumer.sh \
  --bootstrap-server b-2-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-3-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196,b-1-public.mymskcluster.zxrapj.c4.kafka.ap-northeast-1.amazonaws.com:9196 \
  --topic my-test-topic \
  --consumer.config /mskconfig-ssk.properties \
  --from-beginning

おわりに

ここまでの記載の通り、KafkaとAWSに多少慣れている人であれば、AWS上にAmazon MSKを利用したKafka環境を構築するのは難しいことではありませんでした。
Kafkaの製品・サービス選定をする上では、Amazon MSKのSLAや非機能面での制約なども考慮すべきですが、Kafkaを使い慣れたAWS環境でSaaSとして利用できることはアドバンテージになると考えられます。

Kafkaの製品・サービス選定やクラスターの構築・設定の方法について、取っ掛かりとして参考になればと思います。

引き続き、以下の②③を実施していきます。

① Amazon MSKのクラスターを構築する（本稿で紹介）
② AWS外からJavaプログラムを使用して以下を実施（別途紹介）
- シンプルな文字列形式のメッセージのProduce/Consume
- AWS Glue Schema Registryの設定
- Avro形式のメッセージのProduce/Consume
③ AWS外からIBM Infosphere Replication CDCを使用して以下を実施（別途紹介）
- JSON形式でキャプチャー対象テーブルのデータをAmazon MSKに送信
- Avro形式でキャプチャー対象テーブルのデータをAmazon MSKに送信