Kubernetes サービスを使用した IBM® MQ の Event Streams への接続

IBM Cloud® Kubernetes Service クラスターで実行する Kafka Connect ランタイムを取得します。次に、 IBM MQ Source Connector を開始して、 IBM MQ ソース・キューから Event Streamsの宛先 Kafka トピックにメッセージをコピーします。

IBM MQ Source Connector は、 IBM MQ キュー・マネージャーに接続し、 MQ キューから MQ メッセージ・データを取り込みます。コネクターは、各 MQ メッセージを Kafka レコードに変換し、メッセージを Event Streams Kafka トピックに送信します。

ステップ 1. 前提条件のインストール

以下のソフトウェアおよびサービスがインストールされていることを確認します。

Event Streams インスタンス - 標準プランまたはエンタープライズ・プラン。
IBM MQ on Cloud または IBM MQ バージョン 8以降の 1 つのインスタンス。

ユーザー ID とパスワードを使用して IBM MQ で認証するように、 IBM MQ コネクターを構成できます。 MQ ブリッジのインスタンスと関連付けられた ID にのみ以下の許可を付与することをお勧めします。
- CONNECT (接続) 権限。 IBM MQ コネクターは、 MQ キュー・マネージャーに接続できる必要があります。
- IBM MQ コネクターのコンシューム元として構成されているキューに対する GET 権限。
Kubernetes Service クラスター。テスト目的で 1 つを無料でプロビジョンできます。

クラスターへの CLI アクセスも必要です。詳しくは、CLI および API のセットアップ (Setting up the CLI and API) を参照してください。
最新バージョンの kubectl
Git

ステップ 2 kafka-connect リポジトリーの複製

必要なファイルを含んでいる以下の 2 つのリポジトリーを複製します。

ステップ 3。 Kafka Connect 構成の作成

この構成をセットアップする必要があるのは 1 回のみです。Event Streams は、将来の使用のために保管します。

event-streams-samples プロジェクトから、 kafka-connect/IKS directory にナビゲートし、 connect-distributed.properties ファイルを編集し、 <BOOTSTRAP_SERVERS> を 1 つの場所で、 <APIKEY> を 3 つの場所で Event Streams 資格情報に置き換えます。

<BOOTSTRAP_SERVERS> をコンマ区切りリストとして指定します。有効でない場合は、エラーになります。

<APIKEY> はマシン上に平文で表示されますが、IBM Cloud® Kubernetes Service にプッシュされると秘密情報になります。

Kafka Connect では、信頼性とスケーラビリティーの理由から、複数のワーカーを実行できます。 Kubernetes Service クラスターに複数のノードがあり、複数の Connect ワーカーが必要な場合は、kafka-connect.yamlファイルを編集し、エントリーreplicas: 1を編集します。
次に、以下のコマンドを実行します。

シークレットを作成するには、以下のコマンドを実行します。
```
kubectl create secret generic connect-distributed-config --from-file=connect-distributed.properties
```
configmap を作成するには、以下のコマンドを実行します。
```
kubectl create configmap connect-log4j-config --from-file=connect-log4j.properties
```

ステップ 4: Kafka Connect のデプロイ

以下のコマンドを実行して、kafka-connect.yamlファイル内の構成を適用します。

kubectl apply -f ./kafka-connect.yaml

ステップ 5. Kafka Connect が実行中であることの検証

Kafka Connect が実行されていることを確認するには、以下の例のように、ポート 8083 上の kafkaconnect-service にポート転送します。

kubectl port-forward service/kafkaconnect-service 8083

ポート転送に使用した端末を開いたままにし、次のステップに別の端末を使用します。

Connect REST API がhttp://localhost:8083で使用可能になります。 API について詳しくは、以下を参照してください。 Kafka Connect REST インターフェース。

これで、Kubernetes Service にデプロイされ、実行されている Kafka Connect ランタイムが作成されました。次に、 IBM MQ コネクターを構成して開始します。

ステップ 6. mq-source json ファイルの構成

kafka-connect-mq-source/configにあるmq-source.jsonファイルを編集して、少なくとも必要なプロパティーが情報で完成するようにします。

mq-source.json ファイルのプロパティー

mq-source.jsonファイル内のプレースホルダーを独自の値に置き換えます。

トピック: 必須。宛先 Kafka トピックの名前。
QUEUE_MANAGER: 必須。ソース IBM MQ キュー・マネージャーの名前。
キュー: 必須。ソース IBM MQ キューの名前。
CHANNEL_NAME: 必須 (バインディングまたは CCDT ファイルを使用する場合は除外)。サーバー接続チャネルの名前。
CONNECTION_NAME_LIST: 必須 (バインディングまたは CCDT ファイルを使用する場合は除外)。キュー・マネージャーに接続するための 1 つ以上のホスト (ポート) ペアのリスト。各項目はコンマで区切ります。

手順 7. コネクターをその構成で開始します

以下のコマンドを実行して、前のステップで指定した構成で IBM MQ コネクターを開始します。

curl -X POST -H "Content-Type: application/json" http://localhost:8083/connectors --data "@./mq-source.json"

ステップ 8. コネクターのモニター

コネクターを確認するには、以下のアドレスに移動します。

http://localhost:8083/connectors/mq-source/status

コネクターの状態が未実行の場合は、コネクターを再始動してください。