はじめに
Parasoft Kafka Extensions は、SOAtest および Virtualize のカスタム拡張です。Parasoft Kafka Extensions には、トランスポート拡張とリスナー拡張が含まれています。
Parasoft Kafka Transport Extension は、SOAtestのメッセージクライアントツールに適切な Apache Kafkaトランスポートのサービスを追加します。Kafka を介して受け取るメッセージを設定、送信、検証するときに、SOAtest の豊かなインターフェイスをフルに活用できます。
Kafka トランスポートは、多くの場合、Apache Avro 拡張機能と組み合わせて使用されます。Kafka を Avro と併用する基本的な例を含む Avro 拡張機能の詳細については、「Apache Avro 拡張機能」を参照してください。
Parasoft Kafka Listener Extension は、Apache Kafka をサポートし、Virtualize のアセットがトピックからメッセージを読み取って別のトピックにメッセージで応答することを可能にします。
前提条件
- Apache Kafka brokers 0.10.0.0 以降
- Parasoft SOAtest および/または Virtualize 2023.2 以上
インストール手順
このアーティファクトは UI またはコマンドラインからインストールできます。
UI からのインストール
- [Parasoft ] > [設定] を選択し、[システム プロパティ] をクリックします。
- [JAR の追加] をクリックし、com.parasoft.soavirt.kafka-<VERSION>.jar ファイルを選択します。
- [適用] をクリックします。
- SOAtest/Virtualize を再起動します。
コマンドラインからのインストール
設定プロパティ ファイルの system.properties.classpath プロパティに com.parasoft.soavirt.kafka-<VERSION>.jar ファイルを追加します。例:
system.properties.classpath=<PATH_TO_JAR>/com.parasoft.soavirt.kafka-1.2.0.jar
Kafka Transport
使用方法
メッセージ クライアント ツール (たとえば SOAP Client、EDI Client、Messaging Client など) の [トランスポート ] タブで Kafka Transport の使用を有効にします。[トランスポート] メニューから [Kafka] を選択し、表示されるオプションを設定します。
システムに複数のカスタム拡張機能がある場合は、代わりに [トランスポート] メニューから [カスタム拡張] を選択し、表示される [実装の選択] メニューから [Kafka] を選択する必要があります。
設定
トランスポートは、ブロデュ―サー、コンシューマー、またはその両方として設定できます。トランスポートをプロデューサーとしてだけ設定するには、コンシューマー設定を空のままにします。トランスポートをコンシューマーとしてだけ設定するには、プロデューサー設定を空のままにします。
接続設定
| Brokers | Kafka が実行されているサーバーのリストをカンマ区切りで指定します。ポートを含めます。 |
|---|---|
| Security Protocol | セキュリティ プロトコルを指定します。次の値を指定できます:
Transport Layer Security (TLS) 設定は、この Security Protocol に SASL_PLAINTEXT および SASL_SSL
Producer Settings または Consumer Settings の Additional Settings File Path フィールドにファイルを指定します。 |
| Log Level | コンソールに表示する情報の量を指定します。デフォルト値は |
プロデューサー設定
| Topic | プロデューサーがメッセージを書き込むトピックの名前を指定します。 |
|---|---|
| Partition | トピックのパーティションの番号を整数で指定します (例: 空の場合、レコード キーを使用してパーティション番号が生成されます。 |
| Record Key Serializer | キーを文字列、バイナリ、Avro、または null として書き込み/送信するかどうかを指定します。オプションは デフォルト値は |
| Record Key | レコード キーを指定します。レコードキー シリアライザが string に設定されている場合、レコードキーは UTF-8 に変換されます。レコードキー シリアライザが bytes に設定されている場合、0x で始まるレコードキーは 16 進数でデコードされ、それ以外はすべて UTF-8 に変換されます。レコードキー シリアライザが avro に設定されている場合、レコードキーは XML としてモデル化された Avro メッセージからバイナリに変換されます。 |
| Idempotent | 1 つのメッセージだけがブローカーに送信され、重複がないようにするには、 |
| Request Timeout | 確認応答の送受信がタイムアウトするまでに待機する時間をミリ秒単位で指定します。デフォルト値は 30000 です。 |
| Additional Settings File Path | アプリケーションとの接続に必要になる可能性がある追加の設定を含むプロパティ ファイルを指定します。この設定はオプションです。 このフィールドが空の場合、UI には表示されないが、アプリケーションが要求する可能性がある追加の設定については、デフォルトの Kafka 設定が使用されます。 |
コンシューマー設定
| Topic | コンシューマーがサブスクライブする単一のトピック名または複数のトピック名をカンマで区切ったリストを指定します。 トピックにコロンとパーティションを表す整数値を付加することで、トピック内のパーティションを指定できます。例: |
|---|---|
| Group id | コンシューマーが所属するグループを指定します。トピックとともにパーティションが指定されていない場合、このプロパティは必須です。 |
| Offset | レコードの読み取りを開始するトピック/パーティション内の位置を指定します。 |
| Poll interval | ポーリングの間隔をミリ秒単位で指定します。デフォルト値は 100 です。 |
| Max Poll Records | 各回でポーリングするレコードの最大数を指定します。デフォルト値は 500 です。 |
| Commit Offsets | オフセットをコミットする際のモードを指定します。コミットは、コンシューマーがレコードの処理を再開できるパーティション内の位置です。
デフォルト値は |
| Timeout | コンシューマーが接続を切断するまでにサブスクライブ トピックにメッセージがパブリッシュされるのを待機する時間をミリ秒単位で指定します。デフォルト値は 30000 です。 |
| Additional Settings File Path | アプリケーションとの接続に必要になる可能性がある追加の設定を含むプロパティ ファイルを指定します。この設定はオプションです。 このフィールドが空の場合、UI には表示されないが、アプリケーションが要求する可能性がある追加の設定については、デフォルトの Kafka 設定が使用されます。 |
プロデューサー ヘッダー設定
プロデューサー ヘッダーが設定されている場合、それらはレコードメッセージとともに Kafka トピックにパブリッシュされます。カスタムヘッダーは、SOAtest トラフィック ビューアに 16 進数値で表示されます。最大 5 個のヘッダーをサポートできます。
| Header (1-5) Key | カスタム ヘッダーのテキスト文字列キーを指定します。 |
|---|---|
| Header (1-5) Value | カスタムヘッダーの値として、UTF-8 バイトのテキスト文字列、またはバイナリ値の場合は先頭に 0x が付いた 16 進文字列を指定します。null 値を指定するには、フィールドを空のままにします。値として空の文字列を指定するには、0x と入力します。 |
コンシューマー ヘッダー設定
コンシューマー ヘッダーが設定されている場合、SOAtest は指定されたすべてのヘッダー キーと値に一致するレコードを待機します。カスタムヘッダーは、SOAtest トラフィック ビューアに 16 進数値で表示されます。最大 5 個のヘッダーをサポートできます。
| Header (1-5) Key | カスタム ヘッダーのテキスト文字列キーを指定します。 |
|---|---|
| Header (1-5) Value | カスタムヘッダーの値として、UTF-8 バイトのテキスト文字列、またはバイナリ値の場合は先頭に 0x が付いた 16 進文字列を指定します。null 値を指定するには、フィールドを空のままにします。値として空の文字列を指定するには、0x と入力します。 |
Transport Layer Security 設定
| TLS Protocol | セキュアな接続を確立するために使用する トランスポート層プロトコルを定義します。 デフォルト: |
|---|---|
| Security Provider | TLS 接続に使用するセキュリティ プロバイダーの名前です。デフォルト値は JVM のデフォルトのセキュリティ プロバイダーです。 |
| Key Store File Location | アセット (.tst、.pva、.pvn) の [キーストア ファイル] への絶対パスまたは相対パスを定義します。 |
| Key Store Type | [キー ストア ファイル] の種類を定義します。JVM にロードされるセキュリティ プロバイダーに合わせて、異なるキーストア タイプを処理できます。JVM にセキュリティ プロバイダーを追加した場合は、そのセキュリティ プロバイダーのドキュメントを参照し、利用できるキーストア タイプ名を確認してください。 |
| Key Store Password | [キー ストア ファイル] からキー/証明書を抽出するために使用するパスワードを定義します。 空白にした場合、パスワードなしでキーの抽出を試みます。 |
| Key Password | キー ストア ファイルの秘密鍵のパスワードを指定します。クライアントに関しては、この設定はオプションです。 |
| Trust Store File Location | アセット (.tst、.pva、.pvn) の [トラスト ストア ファイル]への絶対パスまたは相対パスを定義します。 デフォルト: |
| Trust Store Type | [トラスト ストア ファイル] の種類を定義します。JVM にロードされるセキュリティ プロバイダーに合わせて、異なるトラスト ストア タイプを処理できます。JVM にセキュリティ プロバイダーを追加した場合は、そのセキュリティ プロバイダーのドキュメントを参照し、利用できるトラスト ストア タイプ名を確認してください。 |
| Trust Store Password | [トラスト ストア ファイル] からキー/証明書を抽出するために使用するパスワードを定義します。空白にした場合、パスワードなしでキーの抽出を試みます。 |
接続の管理設定
Keep connection alive | クライアント接続を持続して今後のパブリッシュで再利用するには、このオプションを有効化します。 |
|---|---|
Close connection after test execution | パブリッシュの直後にクライアント接続を切断するには、このオプションを有効化します。 |
Kafka Listener
使用方法
インストールすると、仮想アセットのデプロイメント設定に Kafka Listener オプションが表示されるようになります。Kafka の実装は、SQL レスポンダー以外のすべてのレスポンダーに使用できます。
- 仮想アセットの設定を開き、[トランスポート] タブをクリックします。
- [カスタム] タブをクリックし、リスナーを設定します (「Configuration」を参照)。システムに複数のカスタム拡張機能がある場合、[実装の選択] メニューから [Kafka Lister] を選択します。
設定
リスナーにはプロデューサーとコンシューマーの両方が設定されている必要があります。
Connection 設定
| Brokers | Kafka が実行されているサーバーのリストをカンマ区切りで指定します。ポートを含めます。 |
|---|---|
| Security Protocol | セキュリティ プロトコルを指定します。次の値を指定できます:
Transport Layer Security (TLS) 設定は、この Security Protocol に SASL_PLAINTEXT および SASL_SSL
Producer Settings または Consumer Settings の Additional Settings File Path フィールドに設定ファイルを指定します。 |
コンシューマー設定
| Topic | コンシューマーがサブスクライブする単一のトピック名または複数のトピック名をカンマで区切ったリストを指定します。 トピックにコロンとパーティションを表す整数値を付加することで、トピック内のパーティションを指定できます。例: |
|---|---|
| Group Id | コンシューマーが所属するグループを指定します。トピックとともにパーティションが指定されていない場合、このプロパティは必須です。 |
| Offset | レコードの読み取りを開始するトピック/パーティション内の位置を指定します。 |
| Max Poll Records | 各回でポーリングするレコードの最大数を指定します。デフォルト値は 500 です。 |
| Commit Offsets | オフセットをコミットする際のモードを指定します。コミットは、コンシューマーがレコードの処理を再開できるパーティション内の位置です。
デフォルト値は |
| Worker Threads | 各インスタンスで実行するワーカー スレッドの数を指定します。最少でも 1 が必要です。デフォルト値は 50 です。 |
| Additional Settings File Path | Virtualize UI には表示されない追加の Kafka コンシューマー設定を指定します。ファイルは各行に https://kafka.apache.org/0100/documentation.html#consumerconfigs |
プロデューサー設定
| Topic | プロデューサーがメッセージを書き込むトピックの名前を指定します。 |
|---|---|
| Partition | トピックのパーティションの番号を整数で指定します (例: 空の場合、レコード キーを使用してパーティション番号が生成されます。 |
| Record Key Serializer | キーを文字列、バイナリ、Avro、または null として書き込み/送信するかどうかを指定します。オプションは デフォルト値は |
| Record Key | レコード キーを指定します。レコードキー シリアライザが string に設定されている場合、レコードキーは UTF-8 に変換されます。レコードキー シリアライザが bytes に設定されている場合、0x で始まるレコードキーは 16 進数でデコードされ、それ以外はすべて UTF-8 に変換されます。レコードキー シリアライザが avro に設定されている場合、レコードキーは XML としてモデル化された Avro メッセージからバイナリに変換されます。 |
| Idempotent | 1 つのメッセージだけがブローカーに送信され、重複がないようにするには、 |
| Request Timeout | 確認応答の送受信がタイムアウトするまでに待機する時間をミリ秒単位で指定します。デフォルト値は 30000 です。 |
| Additional Settings File Path | Virtualize UI には表示されない追加の Kafka プロデューサー設定を指定します。ファイルは各行に https://kafka.apache.org/0100/documentation.html#producerconfigs |
Transport Layer Security
| TLS Protocol | セキュアな接続を確立するために使用する トランスポート層プロトコルを定義します。 デフォルト値は |
|---|---|
| Security Provider | TLS 接続に使用するセキュリティ プロバイダーの名前です。デフォルト値は JVM のデフォルトのセキュリティ プロバイダーです。 |
| Key Store File Location | アセット (.tst、.pva、.pvn) の [キーストア ファイル] への絶対パスまたは相対パスを定義します。 |
| Key Store Type | [キー ストア ファイル] の種類を定義します。JVM にロードされるセキュリティ プロバイダーに合わせて、異なるキーストア タイプを処理できます。JVM にセキュリティ プロバイダーを追加した場合は、そのセキュリティ プロバイダーのドキュメントを参照し、利用できるキーストア タイプ名を確認してください。 |
| Key Store Password | [キー ストア ファイル] からキー/証明書を抽出するために使用するパスワードを定義します。 空白にした場合、パスワードなしでキーの抽出を試みます。 |
| Key Password | キー ストア ファイルの秘密鍵のパスワードを指定します。クライアントに関しては、この設定はオプションです。 |
| Trust Store File Location | アセット (.tst、.pva、.pvn) の [トラスト ストア ファイル]への絶対パスまたは相対パスを定義します。 デフォルトは |
| Trust Store Type | [トラスト ストア ファイル] の種類を定義します。JVM にロードされるセキュリティ プロバイダーに合わせて、異なるトラスト ストア タイプを処理できます。JVM にセキュリティ プロバイダーを追加した場合は、そのセキュリティ プロバイダーのドキュメントを参照し、利用できるトラスト ストア タイプ名を確認してください。 |
| Trust Store Password | [トラスト ストア ファイル] からキー/証明書を抽出するために使用するパスワードを定義します。空白にした場合、パスワードなしでキーの抽出を試みます。 |
ロギング
Log Level | コンソールに表示する情報の量を指定します。デフォルト値は |
|---|
サード パーティのコンテンツ
この拡張機能は、以下のサード パーティのコンテンツを含みます。
- Apache Kafka クライアント (Apache License 2.0)
- SLF4J (MIT ライセンス)
その他のライセンスの詳細については、プラグインの licenses フォルダーを参照してください。
