こんにちは。Red Hatでソリューションアーキテクトをしている杉本です。今回のブログ記事では、Red Hatが提供しているManaged Kafkaをトライアルで実際に試してみる方法について書いていきます。
Red Hat OpenShift Streams for Apache Kafkaとは
Red Hat OpenShift Streams for Apache Kafka（RHOSAKと略され、「ロサック」と発音されることが多いようです）はRed Hatが提供するManaged Kafkaのサービスです。Kafkaの環境を自分で構築・運用することなく、ストリーム処理のアプリケーションの開発やリリースを簡単に行うことができるようになります。概要については、以前の赤帽ブログのApplication Services Newsletter 7月号やSpeakersdeck内のスライドでも紹介していますので、そちらも併せて読んで頂ければと思います。 今回のブログでは、RHOSAKを使うことでKafkaを利用できるところまでを実際に試す方法について説明していきます。なお、2021年7月現在、Red Hat OpenShift Streams for Apache Kafkaはベータ版としてのリリースとなっており、トライアルで作成したKafkaインスタンスは48時間で削除されます。
Red Hat Application Servicesのページ
まず、cloud.redhat.comにアクセスしてRed Hat アカウントでログインします。もしまだRed Hat アカウントをお持ちでなければアカウント作成のリンクからアカウントを作ってください。ログインできたらcloud.redhat.comのランディングページで、左のメニューからApplication Servicesを選択します。
Application Servicesのランディングページが表示されたら、左のメニューでStreams for Apache Kafka → Kafka Instancesを選択します。ベータ環境であることを確認するメッセージが表示されたら、[Use feature in beta] をクリックします。
Kafkaインスタンスの作成
Kafka Instancesのページで[Create Kafka Instance]をクリックしてKafkaインスタンスを作成していきます。Instacne名のフィールドに名前を入力し（ここでは "my-kafka-instance"）、[Create Instance]をクリックするとOpenShift上にKafkaクラスタの構築が開始されます。なお、ベータの時点で利用可能なクラウドはAWSのみでリージョンもUSのみですが、GA後は拡充される予定です。
数分でKafkaクラスターのインスタンスの作成が完了し、以下のような画面が表示されます。
作成されたKafkaインスタンスはOpenShift上にデプロイされており、KafkaのクライアントアプリケーションからKafkaインスタンスにアクセスするにはサービスアカウントが必要です。Kafka Instancesページに表示されている先ほど作成したmy-kafka-instanceのオプションのアイコン（3つのドット）をクリックし、View connection informationを選択します。
Kafkaインスタンスの接続情報が以下のように表示されます。
Kafkaインスタンスの接続情報画面に表示される接続情報からBootstrap serverのエンドポイントをどこか安全な場所にコピーします。
次に[Create service account]ボタンをクリックし、サービスアカウント名を入力して[Create]をクリックします（ここでは "my-service-account"）。
以下のようにClient IDとClient Secretが生成されるので、その値をどこか安全な場所にコピーし、チェックボックスにチェックを入れてクレデンシャルのウィンドウを閉じます。
Kafkaトピックの作成
Kafkaのインスタンスとそれにアクセスするためのサービスアカウントができたら、次にトピックを作成してみましょう。Kafka Instancesのページで、先ほど作成したKafkaインスタンスの名前をクリックします。Topicsタブで、[Create topic]をクリックし、ガイドに従ってトピックに関する情報を設定していきます。
- トピック名（ここでは "my-first-kafka-topic"）
- パーティション数（デフォルトは1）
- メッセージのリテンションポリシー（ここではリテンション期間は1週間、リテンションサイズはUnlimitedを指定）
- レプリカ数（現在のRHOSAKでは事前に設定された値を変更することはできません）
Kafkaトピックにメッセージを送信 (produce)
以上でKafkaを使う用意が整いましたので、トピックにメッセージを送信したり、トピックからメッセージを取り出すことができるようになります。ここでは、Kafkaとのメッセージの送受信をコマンドラインで行うことのできるkafkacatを使用してみます。kafkacatはhttps://github.com/edenhill/kafkacatからダウンロードできますので、手順に従ってインストールしてみてください。Macの場合には以下のようにbrewを使ってもインストール可能です。
$ brew install kafkacat
kafkacatがKafkaインスタンスにアクセスできるようにするためには、先ほど保存したKafkaインスタンスのBootstrap serverのエンドポイントとサービスアカウントのクレデンシャルが必要となります。以下のコマンドで環境変数として設定しておきましょう。それぞれの値は実際の値に置き換えてください。
$ export BOOTSTRAP_SERVER=<bootstrap_server> $ export USER=<client_id> $ export PASSWORD=<client_secret>
ターミナルで以下のコマンドを実行することでkafkacatをProducerモードで開始することができます。以下のコマンド例で接続するKafkaトピックは、先に作成したmy-first-kafka-topicを使用していますので必要に応じて書き換えてください。またこの例では、先ほど設定したBootstrap serverとサービスアカウントのクレデンシャルを使用してSASL/PLAINの認証を行っています。
$ kafkacat -t my-first-kafka-topic -b "$BOOTSTRAP_SERVER" \ -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \ -X sasl.username="$USER" \ -X sasl.password="$PASSWORD" -P
kafkacatをProducerモードで起動すると、ターミナルの標準入力 (stdin) から直接Kafkaトピックにメッセージを送信することができます。以下のように標準入力で入力してメッセージを送信してみましょう。一行ごとにトピックにメッセージが送信されます。
First message Second message Third message
このメッセージのproducerは後ほどまた使用しますのでターミナルはこのままにしておきましょう。
Kafkaトピックからメッセージを受信 (consume)
次にKafkaトピックからメッセージを取得するために、ターミナルをもう一つ立ち上げて、Producerの時と同様に環境変数をしたら、以下のコマンドを実行してkafkacatをConsumerモードで開始しましょう。Consumerモードで起動してトピックに接続したら（ここでは"my-first-kafka-topic"）、先ほどProducerとして送信したメッセージを受信して、標準出力にメッセージが出力されます。
$ kafkacat -t my-first-kafka-topic -b "$BOOTSTRAP_SERVER" \ -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \ -X sasl.username="$USER" \ -X sasl.password="$PASSWORD" -C First message Second message Third message % Reached end of topic my-first-kafka-topic [0] at offset 3
Producerモードのkafkacatのターミナルからメッセージを送信すると、Consumerモードのkafkacatのターミナルにメッセージが表示され、Kafkaのトピックでメッセージの送受信が行われているのを確認することができます。
まとめ
このようにRHOSAKを使用すると、Kafkaのクラスタ環境を簡単に構築し、利用することができるようになります。cloud.redhat.comにアクセスすればトライアルはすぐにできますので、ぜひRed HatによるManaged Kafka, RHOSAKを試してみていただければと思います。
