翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# ステップ 4: Amazon MSK クラスターにトピックを作成する
[Amazon MSK の開始方法](getting-started.md)のこのステップでは、CreateTopic API でネイティブ AWS ツールを使用するか、クライアントマシンで Apache Kafka AdminClient ツールを使用するという 2 つの方法のいずれかを使用してトピックを作成できます。
**警告**
CreateTopic API で AWS ツールを使用する場合は、クラスターが要件を満たしていることを確認します。詳細については、「 [トピック APIs](https://docs.aws.amazon.com/msk/latest/developerguide/msk-topic-operations-information.html#topic-operations-requirements)」を参照してください。
**警告**
AdminClient アプローチを使用する場合、このチュートリアルで使用される Apache Kafka バージョン番号は例にすぎません。MSK クラスターバージョンと同じバージョンのクライアントを使用することが推奨されます。旧クライアントバージョンでは、特定の機能や重大なバグ修正が失われている可能性があります。
**Topics**
+ [AWS ツールを使用したトピックの作成](#create-topic-aws-tools)
+ [MSK クラスターバージョンの確認](#find-msk-cluster-version)
+ [クライアントマシン上でトピックを作成する](#create-topic-client-machine)
## AWS ツールを使用したトピックの作成
CLI、SDK、 AWS マネジメントコンソールなどの AWS ツールを使用して、MSK AWS クラスターにトピックを作成できます。 AWS SDKs このアプローチは、Kafka クライアントツールに直接アクセスすることなく、トピックを管理するための効率的な方法を提供します。
AWS ツールを使用したトピックの作成の詳細については、[CreateTopic API デベロッパーガイド](https://docs.aws.amazon.com/msk/latest/developerguide/msk-create-topic.html)を参照してください。
## MSK クラスターバージョンの確認
1. [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/) で Amazon MSK コンソールを開きます。
1. ナビゲーションバーで、クラスターを作成したリージョンを選択します。
1. MSK クラスターを選択します。
1. クラスターで使用されている Apache Kafka のバージョンをメモします。
1. このチュートリアルの Amazon MSK バージョン番号のインスタンスは、ステップ 3 で取得したバージョンに置き換えてください。
## クライアントマシン上でトピックを作成する
1. **クライアントマシンへ接続します。**
1. Amazon EC2 コンソールの [https://console.aws.amazon.com/ec2/](https://console.aws.amazon.com/ec2/) を開いてください。
1. ナビゲーションペインで、[**インスタンス**] を選択してください。次に、「[ステップ 3: クライアントマシンを作成する](create-client-machine.md)」で作成したクライアントマシンの名前の横にあるチェックボックスを選択します。
1. [**Actions (アクション)**] を選択して、[**Connect (接続)**] を選択します。コンソールの指示に従ってクライアントマシンに接続します。
1. **Java をインストールし、Kafka バージョン環境変数を設定します。**
1. クライアントマシンに Java をインストールするには、次のコマンドを実行します。
```
sudo yum -y install java-11
```
1. 次のコマンドに示すように、MSK クラスターの [Kafka バージョン](#find-msk-cluster-version)を環境変数 `KAFKA_VERSION` に保存します。この情報は、セットアップ全体で必要になります。
```
export KAFKA_VERSION={KAFKA VERSION}
```
例えば、バージョン 3.6.0 を使用する場合、以下のコマンドを実行します。
```
export KAFKA_VERSION=3.6.0
```
1. **Apache Kafka をダウンロードして抽出します。**
1. 次のコマンドを実行して、Apache Kafka をダウンロードします。
```
wget https://archive.apache.org/dist/kafka/$KAFKA_VERSION/kafka_2.13-$KAFKA_VERSION.tgz
```
**注記**
以下のリストは、問題が発生した場合に使用できる代替 Kafka ダウンロード情報を示しています。
接続の問題が発生したり、ミラーサイトを使用したりする場合は、次のコマンドに示すように、Apache ミラーセレクタを試してください。
```
wget https://www.apache.org/dyn/closer.cgi?path=/kafka/$KAFKA_VERSION/kafka_2.13-$KAFKA_VERSION.tgz
```
[Apache Kafka ウェブサイト](https://kafka.apache.org/downloads)から直接適切なバージョンをダウンロードします。
1. 前のステップで TAR ファイルをダウンロードしたディレクトリで次のコマンドを実行します。
```
tar -xzf kafka_2.13-$KAFKA_VERSION.tgz
```
1. 新しく作成したディレクトリへのフルパスを`KAFKA_ROOT`環境変数内に保存します。
```
export KAFKA_ROOT=$(pwd)/kafka_2.13-$KAFKA_VERSION
```
1. **MSK クラスターの認証を設定します。**
1. Amazon [MSK IAM クライアントライブラリの最新バージョン](https://github.com/aws/aws-msk-iam-auth/releases/latest)を検索します。このライブラリを使用すると、クライアントマシンは IAM 認証を使用して MSK クラスターにアクセスできます。
1. 以下のコマンドを使用して、 `$KAFKA_ROOT/libs` ディレクトリに移動し、前のステップで見つけた関連する Amazon MSK IAM JAR をダウンロードします。必ず*\$1LATEST VERSION\$1*をダウンロードする実際のバージョン番号に置き換えてください。
```
cd $KAFKA_ROOT/libs
```
```
wget https://github.com/aws/aws-msk-iam-auth/releases/latest/download/aws-msk-iam-auth-{LATEST VERSION}-all.jar
```
**注記**
MSK クラスターとやり取りする Kafka コマンドを実行する前に、Java クラスパスに Amazon MSK IAM JAR ファイルを追加する必要がある場合があります。次の例に示すように、`CLASSPATH`環境変数を設定します。
```
export CLASSPATH=$KAFKA_ROOT/libs/aws-msk-iam-auth-{LATEST VERSION}-all.jar
```
これにより、セッション全体の `CLASSPATH` が設定され、以降のすべての Kafka コマンドで JAR を使用できるようになります。
1. `$KAFKA_ROOT/config` ディレクトリに移動して、クライアント設定ファイルを作成します。
```
cd $KAFKA_ROOT/config
```
1. 次のプロパティ設定をコピーして、新しいファイルに貼り付けます。**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
```
1. **(オプション) Kafka ツールの Java ヒープサイズを調整します。**
メモリ関連の問題が発生した場合、または多数のトピックやパーティションを扱う場合は、Java ヒープサイズを調整できます。これを行うには、Kafka コマンドを実行する前に `KAFKA_HEAP_OPTS`環境変数を設定します。
次の例では、最大ヒープサイズと初期ヒープサイズの両方を 512 メガバイトに設定します。これらの値は、特定の要件と利用可能なシステムリソースに応じて調整してください。
```
export KAFKA_HEAP_OPTS="-Xmx512M -Xms512M"
```
1. **クラスター接続情報を取得します。**
1. [https://console.aws.amazon.com/msk/](https://console.aws.amazon.com/msk/) で Amazon MSK コンソールを開きます。
1. クラスターのステータスが **[アクティブ]** になるまで待ちます。この処理には数分かかることがあります。ステータスが **[アクティブ]** になったら、クラスター名を選択します。これにより、そのクラスターの概要を含むページに移動します。
1. **View client information** (ライアント情報の表示) を選択します。
1. プライベートエンドポイントの接続文字列をコピーします。
ブローカーごとに 3 つのエンドポイントが提供されます。次のコマンドに示すように、これらの接続文字列の 1 つを環境変数 `BOOTSTRAP_SERVER` に保存します。** を接続文字列の実際の値に置き換えます。
```
export BOOTSTRAP_SERVER=
```
1. **次のコマンドを実行して、トピックを作成します。**
```
$KAFKA_ROOT/bin/kafka-topics.sh --create --bootstrap-server $BOOTSTRAP_SERVER --command-config $KAFKA_ROOT/config/client.properties --replication-factor 3 --partitions 1 --topic MSKTutorialTopic
```
`client.properties` ファイルの `NoSuchFileException` を取得した場合は、このファイルが Kafka bin ディレクトリ内の現在の作業ディレクトリに存在することを確認してください。
**注記**
セッション全体に`CLASSPATH`環境変数を設定しない場合は、代わりに各 Kafka コマンドに `CLASSPATH`変数をプレフィックスできます。このアプローチでは、クラスパスはその特定のコマンドにのみ適用されます。
```
CLASSPATH=$KAFKA_ROOT/libs/aws-msk-iam-auth-{LATEST VERSION}-all.jar \
$KAFKA_ROOT/bin/kafka-topics.sh --create \
--bootstrap-server $BOOTSTRAP_SERVER \
--command-config $KAFKA_ROOT/config/client.properties \
--replication-factor 3 \
--partitions 1 \
--topic MSKTutorialTopic
```
1. **(任意) トピックが正常に作成されたことを確認してください。**
1. コマンドが成功した場合、次のメッセージが表示されます。 `Created topic MSKTutorialTopic.`
1. トピックが存在することを確認するために、すべてのトピックを一覧表示します。
```
$KAFKA_ROOT/bin/kafka-topics.sh --list --bootstrap-server $BOOTSTRAP_SERVER --command-config $KAFKA_ROOT/config/client.properties
```
コマンドが失敗した場合、またはエラーが発生した場合は、トラブルシューティング情報について「[Amazon MSK クラスターをトラブルシューティングする](troubleshooting.md)」を参照してください。
1. **(オプション) この チュートリアル で使用した環境変数を削除します。**
このチュートリアルの次の手順で環境変数を維持したい場合は、この手順をスキップしてください。そうでない場合は、次の例に示すようにこれらの変数を解除できます。
```
unset KAFKA_VERSION KAFKA_ROOT BOOTSTRAP_SERVER CLASSPATH KAFKA_HEAP_OPTS
```
**次のステップ**
[ステップ 5: データを生成および消費する](produce-consume.md)