第2章 AMQ Streams の評価

手順

1. サブスクリプションがアクティベートされ、システムが登録されていることを確認します。

   カスタマーポータルを使用して Red Hat サブスクリプションをアクティブ化し、システムを登録する方法は、付録A サブスクリプションの使用 を参照してください。

2. AMQ Streams のダウンロードサイト から amq-streams-x.y.z-ocp-install-examples.zip ファイルをダウンロードします。

3. ファイルを任意の場所で展開します。

   • Windows または Mac の場合: ZIP ファイルをダブルクリックして ZIP アーカイブの内容を展開します。
   • Red Hat Enterprise Linux の場合: ターゲットマシンでターミナルウィンドウを開き、ZIP ファイルがダウンロードされた場所に移動します。

   以下のコマンドを使用して、ZIP ファイルを展開します。

   unzip amq-streams-x.y.z-ocp-install-examples.zip

手順

1. クラスター管理者権限を持つアカウントを使用して OpenShift クラスターにログインします。

   例を以下に示します。

   oc login -u system:admin

2. AMQ Streams Kafka Cluster Operator に新規の kafka (プロジェクト) namespace を作成します。

   oc new-project kafka

3. AMQ Streams Kafka Cluster Operator をインストールする新しい kafka namespace を参照するインストールファイルを変更します。

   注記

   デフォルトでは、ファイルは myproject namespace で動作します。

   • Linux の場合は、以下を使用します。

   sed -i 's/namespace: .*/namespace: kafka/' install/cluster-operator/*RoleBinding*.yaml

   • Mac の場合は、以下を使用します。

   sed -i '' 's/namespace: .*/namespace: kafka/' install/cluster-operator/*RoleBinding*.yaml

4. CRD を管理するために、CRD およびロールベースアクセス制御 (RBAC) リソースをデプロイします。

   oc project kafka
   oc apply -f install/cluster-operator/

5. Kafka クラスターをデプロイする新しい my-kafka-project namespace を作成します。

   oc new-project my-kafka-project

6. 管理者でないユーザー developer に my-kafka-project へのアクセス権限を付与します。

   以下に例を示します。

   oc adm policy add-role-to-user admin developer -n my-kafka-project

7. my-kafka-project namespace を監視する権限を Cluster Operator に付与します。

   oc set env deploy/strimzi-cluster-operator STRIMZI_NAMESPACE=kafka,my-kafka-project -n kafka

   oc apply -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n my-kafka-project

   oc apply -f install/cluster-operator/032-RoleBinding-strimzi-cluster-operator-topic-operator-delegation.yaml -n my-kafka-project

   oc apply -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n my-kafka-project

   このコマンドによって、Cluster Operator が Kafka クラスターにアクセスするための権限を付与するロールバインディングが作成されます。

8. 新しいクラスターロール strimzi-admin を作成します。

   oc apply -f install/strimzi-admin

9. ロールを管理者でないユーザー developer に追加します。

   oc adm policy add-cluster-role-to-user strimzi-admin developer

手順

1. ユーザー developer として my-kafka-project namespace にログインします。

   以下に例を示します。

   oc login -u developer
   oc project my-kafka-project

   新規ユーザーが OpenShift Container Platform にログインした後に、そのユーザーのアカウントが作成されます。

2. 3 つの Zookeeper ノードと 3 つのブローカーノードで、新しい my-cluster Kafka クラスターを作成します。

   • ephemeral ストレージを使用します。

   • route を使用するよう設定された外部リスナーを使用して、OpenShift クラスター外部の Kafka クラスターを公開します。

     cat << EOF | oc create -f -
     apiVersion: kafka.strimzi.io/v1beta1
     kind: Kafka
     metadata:
       name: my-cluster
     spec:
       kafka:
         replicas: 3
         listeners:
           plain: {}
           tls: {}
           external:
             type: route
         storage:
           type: ephemeral
       zookeeper:
         replicas: 3
         storage:
           type: ephemeral
       entityOperator:
         topicOperator: {}
     EOF

3. クラスターがデプロイされるまで待機します。

   oc wait my-kafka-project/my-cluster --for=condition=Ready --timeout=300s -n kafka

4. クラスターの準備ができたら、パブリッシュするトピックを作成し、外部クライアントからサブスクライブします。

   my-cluster Kafka クラスターで 3 つのレプリカと 3 つのパーティションを使用して、以下の my-topic カスタムリソース定義を作成します。

   cat << EOF | oc create -f -
   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaTopic
   metadata:
     name: my-topic
     labels:
       strimzi.io/cluster: "my-cluster"
   spec:
     partitions: 3
     replicas: 3
   EOF

手順

1. ブートストラップ route のアドレスを見つけます。

   oc get routes my-cluster-kafka-bootstrap -o=jsonpath='{.status.ingress[0].host}{"\n"}'

   このアドレスと Kafkaクライアントの 443 番ポートをブートストラップアドレスとして使用します。

2. ブローカーの認証局の公開証明書を抽出します。

   oc extract secret/my-cluster-cluster-ca-cert --keys=ca.crt --to=- > ca.crt

3. 信頼できる証明書をトラストストアにインポートします。

   keytool -keystore client.truststore.jks -alias CARoot -import -file ca.crt

   メッセージの送受信を開始する準備が整いました。

手順

1. AMQ Streams のダウンロードサイト から最新バージョンの AMQ Stremas アーカイブ (amq-streams-x.y.z-bin.zip) をダウンロードします。

   ファイルを任意の場所で展開します。

2. ターミナルを開き、トピック my-topic と TLS の認証プロパティーを使用して Kafka コンソールプロデューサーを起動します。

   bin/kafka-console-producer.sh --broker-list <route-address>:443 --producer-property security.protocol=SSL --producer-property ssl.truststore.password=password --producer-property ssl.truststore.location=./client.truststore.jks --topic my-topic

3. プロデューサーが実行しているコンソールにメッセージを入力します。
4. Enter を押してメッセージを送信します。

5. 新しいターミナルタブまたはウィンドウを開き、Kafka コンソールコンシューマーを起動してメッセージを受信します。

   bin/kafka-console-consumer.sh --bootstrap-server <route-address>:443 --consumer-property security.protocol=SSL --consumer-property ssl.truststore.password=password --consumer-property ssl.truststore.location=./client.truststore.jks --topic my-topic --from-beginning

6. コンシューマーコンソールに受信メッセージが表示されることを確認します。
7. Crtl+C を押して、Kafka コンソールプロデューサーとコンシューマーを終了します。