Kafkaシンクコネクタで一意のクライアントIDを設定する

Question

分散ワーカで実行中のコネクタタスクを監視するためにJMXを使用しようとしています。使用できる唯一のJMX出力は、カフカコンシューマMBeanのようです。残念ながら、労働者の仕事で使用されているカフカ消費者の識別可能なclient.idを設定することは不可能です。

文書には、connector configurationにclient.idを使用するように指示されています。これをコネクタ設定に設定しても効果はありません。これをワーカー構成（プロパティファイル）に設定すると、ワーカーが設定およびステータスのトピックで使用するためにのみ設定されます。

the codeを見ると、シンクコンシューマは接頭辞がconsumer.のワーカー構成を使用しているようです。これは機能しますが、ワーカーのすべてのタスクに設定された単一のクライアントIDであり、単一のコンシューマ、または特定のコネクタまで監視することは不可能です。

デフォルトの動作では、コンシューマ内でデフォルトのclient.idが使用されます。これはconsumer-nを使用します（nは増分された数値です）。 JMX出力には、このコンシューマが含まれているgroupd.idが含まれていないため、コンシューマ接続をそのワーカで実行されているコネクタと相関させることができません。

追加のJMX出力（groupd.idやコネクタ名など）を使用しないか、より良いデフォルト値client.idを使用すると、JMXを使用してカフカコネクタを監視する方法がわかりません。

Answer 1

まず、設定がどこに行くのかについて少し混乱していると思います。設定には、コーカーとコネクタの2種類があります。 （たとえば、コネクタごとにデフォルトのワーカーレベルのコンフィグレーションをオーバーライドできるようにする）いくつかのコンフィグがありますが、ほとんどの場合、ワーカーレベルのコンフィグはプロセス全体の動作を制御しますが、コネクタコンフィグレーション個々のコネクタに固有です。

論理的には、1つのConnectクラスタで多数のクライアントが動作するため、独立して扱うことができます。これはあなたの質問が得意とするものです。クライアントのタイプには、ソースタスク（プロデューサ）、シンクタスク（コンシューマ）、フレームワークレベルのクライアント（Worker）などがあります。

あなたが最初に与えたのは、Kafka Connect workerのリンクです。従業員は消費者が使用するのと同じグループ調整プロトコルによって互いに調整するので、client.idが表示されます。したがって、client.idは、作業者の設定でmy-kafka-connect-cluster-workerのようなものになります。ワーカーに関連付けられたトラフィックは、Connectクラスタのワーカーに関連付けられています。

consumer.client.id（または実際にはプロデューサまたはコンシューマの設定）を上書きできることは間違いありません。ボード全体で、そのワーカーで作成されたすべてのクライアントに適用されます。しかし、あなたが正しく注意するように、これは統一されます。

簡単な答えは、現時点では、修正が簡単ではない（そして間違いなく価値がある）が、オーバーライドするのは簡単ではないということです。 creates producersとcreates consumersというコードは、デフォルトのクライアントIDを生成して設定に追加するには、少なくともワーカーグループIDとタスクIDを使用する必要があります（2の組み合わせは、複数の接続クラスタを同じにしても一意のIDを保証しますカフカクラスター）。

この問題に対処するためのJIRA filedがあります。