問題タブ [confluent-kafka-dotnet]

Confluent Kafka lib for .NET を使用して、提供された資格情報がさまざまな Kafka トピックへの読み取りおよび書き込みアクセス権を持っているかどうかをプログラムで確認する方法があるかどうかを調べようとしています。

私がやりたいのは、基本的にシステムの起動時にスモーク テストを行い、指定された資格情報が正しいことを確認することです。たとえば、さまざまな設定でさまざまな環境にデプロイする場合などです。

コンシューマまたはプロデューサ全体を設定してから、実際にデータを読み書きするのは、ハッキリしていて費用がかかるように思えます。

たとえば、AdminClientこれを検証できる何かがあるのではないかと思いましたが、その方向を示唆するものは何もありません。

メッセージを消費するために C# で Confluent kafka を使用しています。これらのメッセージは、スキーマ フィンガープリントが抽出される 16 進文字列としてフォーマットされます。C#でスキーマフィンガープリントからスキーマを取得するには? 私は何かを逃していますか？

Apache Kafka を介して統合されたサービスが多数あり、各サービスにはコンシューマーとプロデューサーがありますが、トピックに多くの負荷がかかると、何かが消費を遅くするような消費速度の低下に直面しています。

私のカフカ消費者実装の例を次に示します。

これが私のプロデューサーの例です：

ご覧のとおり、コンシューマーは kafka トピックからメッセージを読み取り、メッセージを MediatR INotification オブジェクトに逆シリアル化し、ハンドラーに発行するだけです

ハンドラーは、データベース トランザクション、redis キャッシュの読み取り/書き込み、およびプッシュ通知を操作します。

私のハンドラの例:

しかし、15 秒のランプアップで 2000 リクエストの負荷テストを実行すると、コンシューマーが遅くなり始め、2000 リクエストすべてを消費するのに 2 ~ 5 分かかります。

MediatR レイヤーを削除して Consumer クラスでプロセスの処理を開始すると、パフォーマンスが向上するのではないかと考えていました

または、スループットを向上させる Kafka 構成がある場合は、In Sync トピックのレプリカの Ack を削除するか、しばらくしてからオフセットをコミットします。

最初に、MassTransit ライブラリを使用して kafka を実装しました。その後、この遅い消費速度を見つけた後、ライブラリを Confluet.Kafka に変更して、改善がある場合は MassTransit 抽象化レイヤーを削除しようとしましたが、それでも同じです:

<PackageReference Include="Confluent.Kafka" Version="1.7.0" />

すでに同じ問題に直面している人は誰でも私を助けることができますか?

OBS: 私の Kafka は Kubernetes の 3 つのブローカーを持つクラスターで実行されており、トピックごとに 3 つのレプリケーション ファクターを持つ 6 つのパーティションがあります。

AWS マネージド Kafka インスタンス (MSK) を使用して Kafka クライアント アプリ (プロデューサーとコンシューマーの両方) を作成しようとしています。また、ブローカーからブローカーへの通信およびクライアントからブローカーへの通信は、クラスター内の TLS を介して安全に構成されます。CA は AWS プライベート CA であり、クライアントが TLS 経由で認証を仲介するために MSK でサポートされている唯一の CA です。

問題のコンテキスト: AWS の公式ドキュメント ( https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html#msk-authentication-client ) の手順は Java の世界に傾倒しており、クライアントのトラストストアと取引を扱っています。 jksとしてのキーストア。ただし、.Net クライアントは Java JKS コンテナー形式 ( https://github.com/mhowlett/confluent-kafka-dotnet/tree/security/examples/Security )を使用しません。

• サーバー認証クライアント：この部分は解決できました。jks は単なるデータ ストアであるため、上記の aws doc に従ってキーストアを作成した後、追加の keytool および openssl コマンドをいくつか実行して、クライアント証明書とキーを明示的に抽出しました。これを使用して、メッセージを正常に生成および消費することができました。

• ただし、クライアントがサーバーを検証するには、ssl.ca.location を CA ルート証明書に設定する必要があります。プライベート CA (MSK インスタンスで CA として使用されるもの) からルート CA をダウンロードしました。これはデフォルトで pem 形式です (注: これには証明書の詳細のみが含まれ、キーの詳細は含まれません)。以下のコマンドを使用して .crt に変換しました。

  add pem to truststore : keytool -keystore kafka.client.truststore.jks -alias CARoot -importcert -file Certificate.pem

  get cert from truststore : keytool -export -alias CARoot -keystore kafka.client.truststore.jks -rfc -file ca-root.crt

上記の ca-root.crt を ca の場所に使用しても機能していないようです。ブローカの検証に失敗したというエラーがスローされます。クライアント マシンは Windows ですが、証明書の場所 ssl.ca.location を渡すこと以外は問題ありません。さらに、マシンに証明書をインストールしようとしましたが、まだ運がありません。以下のエラーが表示されます。

クライアント検証サーバー部分で何か不足していますか?

更新: ブローカーの検証も解決できました。上記の公式の AWS ドキュメントは、ブローカー検証認証のために Java 証明書に依存しています。jks から perm とさらに .crt を派生させるために少し拡張し、生成された .crt を ssl.ca.location の CA として使用したところ、機能しました。ただし、生成される証明書にはあまりにも多くの証明書があり、MSK ブローカー証明書の検証に使用される証明書が明確ではありません。また、プライベート AWS CA からダウンロードした CA 証明書が機能しないのはなぜですか? 誰かがこれらを手伝ってくれると本当に助かります。また、問題に対する代替の標準的な解決策があれば提案してください。