AWS MSKのディザスタリカバリ構築手順

1. S3バケット2つ（1つは現行のMSKクラスターと同じリージョン、もう1つはDRリージョン）

2. S3 VPCゲートウェイエンドポイント

3. 2つのS3バケット間のレプリケーション

4. Confluent Amazon S3 Sink ConnectorのZipファイル

5. AWS MSK Connectorのログを監視するためのAWS CloudWatchロググループ

6. 後述するS3ポリシーを付与したKafka Connector用ロール

ステップ1: S3バケットの作成

本番リージョンとDRリージョンに、それぞれS3バケットを1つずつ作成します。本記事では本番リージョンとして米国北部バージニア（us-east-1）、DRリージョンとして米国オレゴン（us-west-2）を使用します。

新しいMSKクラスターを起動します（すでに稼働中のMSKクラスターがある場合は、本ステップをスキップしても構いません）。

セキュリティ設定では、赤くハイライトされた以下の2つのチェックボックスのみを選択してください。この2つを選んでいる理由は、暗号化を有効にする場合にプロデューサー側とコンシューマー側で暗号化証明書を作成する必要があり、その内容は本記事の対象外であるためです。

続いて、MSKがトピックデータを宛先バケットに書き込むために使うロールを用意します。IAMコンソールで「Roles」>「Create Roles」をクリックし、以下のオプションを選択します。本記事では「msk-blog-role」という名前でロールを作成しました。

本番クラスターで使用するS3バケットへアクセスできるように、MSK IAMロール用のカスタムポリシーを作成します。IAMロールにアタッチするIAMポリシーのJSONは以下のとおりです。本記事ではポリシー名を「msk-blog-s3-policy」としました。

{
  "Version": "2012-10-17",
  "Statement": [\
    {\
      "Effect": "Allow",\
      "Action": [\
        "s3:ListAllMyBuckets"\
      ],\
      "Resource": "arn:aws:s3:::*"\
    },\
    {\
      "Effect": "Allow",\
      "Action": [\
        "s3:ListBucket",\
        "s3:GetBucketLocation",\
        "s3:DeleteObject"\
      ],\
      "Resource": "arn:aws:s3:::<name-of-your-bucket>"\
    },\
    {\
      "Effect": "Allow",\
      "Action": [\
        "s3:PutObject",\
        "s3:GetObject",\
        "s3:AbortMultipartUpload",\
        "s3:ListMultipartUploadParts",\
        "s3:ListBucketMultipartUploads"\
      ],\
      "Resource": "*"\
    }\
  ]
}

ポリシーを作成したら、ロールにアタッチします。

アタッチが完了したらロールを保存し、「Trust relationships」をクリックして以下のJSONに置き換えます。

{
  "Version": "2012-10-17",
  "Statement": [\
    {\
      "Effect": "Allow",\
      "Principal": {\
        "Service": "kafkaconnect.amazonaws.com"\
      },\
      "Action": "sts:AssumeRole"\
    }\
  ]
}

MSK Connectorがインターネットを経由せずに宛先S3バケットへデータを書き込むには、VPCエンドポイントが必要です。本記事では、ゲートウェイエンドポイントmsk-blog-vpc-endpointを作成しました。

Confluent S3 Sink Connector [LINK] をダウンロードし、S3バケットにアップロードします。次にAWS MSKコンソールへ移動し、左側の「Custom plugins」を選択して「Create custom plugin」をクリックします。Confluent S3 Sink ConnectorのZipファイルをアップロードしたS3バケットを参照してファイルを選び、プラグイン名を入力したうえで「Create custom plugin」をクリックします。

Amazon Linux 2 AMIのEC2インスタンスを起動します。次に、新規作成したEC2インスタンスからMSKクラスターへデータを送信できるよう、MSKクラスターに紐づいたセキュリティグループを、新規EC2インスタンスのインバウンドルールに「All Traffic」で追加します（下図参照）。

Kafkaトピックを作成するには、EC2インスタンス上で以下のコマンドを実行します。

#Javaのインストール
sudo yum install java-1.8.0

#Apache Kafkaのダウンロード
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz

#ファイルの展開
tar -xzf kafka_2.12-2.2.1.tgz

続いて、Apache ZooKeeper接続用のクライアント情報を取得します。MSKクラスターでの確認場所は下図のとおりです。

取得したプレーンテキストの接続文字列を、以下のコマンド実行時に使用します。

#コマンドのサンプル

<path-to-your-kafka-installation>/bin/kafka-topics.sh --create --bootstrap-server bootstrapServerString --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic

#実際のコマンドは以下のような形になります。

/home/ec2-user/kafka_2.12-2.2.1/bin/kafka-topics.sh --create --bootstrap-server z-3.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181,z-2.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181,z-1.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181 --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic

先ほど作成したカスタムConfluentプラグインを使ってMSK Connectorを作成します。

「Next」をクリックし、命名規則に沿ってコネクタ名を入力します。続いてMSKクラスターを選択します。

コネクタ設定フィールドに、以下のコードスニペットを追加します。S3バケット名を実際の宛先バケット名に置き換えるのを忘れないでください。

connector.class=io.confluent.connect.s3.S3SinkConnector
s3.region=us-east-1
format.class=io.confluent.connect.s3.format.json.JsonFormat
flush.size=1
schema.compatibility=NONE
tasks.max=2
topics.regex=name.(.*)
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
storage.class=io.confluent.connect.s3.storage.S3Storage
s3.bucket.name=<my-destination-bucket>
topics.dir=tutorial

その他に必要な設定は以下のとおりです。本記事はPOCのためcapacityとworkersはデフォルトのままとしていますが、実運用ではMSKクラスターが扱うトピック数に応じてcapacityを調整してください。

最後に「Create connector」をクリックします。上記のとおり設定が正しければ、コネクタのステータス欄に緑色の「Running」が表示されます。「Failed」と表示された場合は、上記の手順のいずれかで抜けや誤りがある可能性が高いです。原因を特定するには、本コネクタ用に作成したCloudWatchロググループの作成ログを確認するとよいでしょう。

Connectorの連携が正しく動作しているかを確認するため、プロデューサー用EC2サーバーでテストメッセージを作成してみましょう。

プロデューサー用EC2サーバーで、以下のコマンドを実行します。

/home/ec2-user/kafka_2.12-2.2.1/bin/kafka-console-producer.sh --broker-list z-3.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181,z-2.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181,z-1.xxxxxxxxxxxxxxx.xxxxxxxx.c4.kafka.us-east-1.amazonaws.com:2181 --replication-factor 2 --partitions 1 --topic mkc-tutorial-topic

EC2のコマンドラインで任意のメッセージを入力します。1〜2分待ってから宛先バケットを開き、メッセージがバケット内にバックアップされているかを確認します。

これで、トピックデータが先ほど作成した宛先バケットへバックアップされる状態になりました。ただし、AWSリージョン障害が発生するとそのデータにはアクセスできなくなるため、本番側の宛先バケット（例: us-east-1）からDRリージョン（us-west-2）のバケットへデータレプリケーションを設定しておく必要があります。設定方法は、こちらのAWS公式ドキュメントをご参照ください。

[1] https://aws.amazon.com/msk/

[2] https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect.html

[3] https://docs.aws.amazon.com/msk/latest/developerguide/msk-connect-getting-started.html