問題タブ [kafka-producer-api]

Java オブジェクトをトピックに公開するために、独自のシリアライザーを作成しました。serializer.class プロパティをカスタム シリアライザーに設定しました。プロデューサーを実行すると、次の例外が発生します。誰かが私を助けることができますか？

例外

私のプロデューサー

複数のパーティションを持つ単一のトピックでカフカ プロデューサーを実装しています。メッセージ内の特定の値 (メッセージ json の feedName プロパティ値) によってメッセージが移動するパーティションを選択しています。feedName - partitionId マッピングの SQL テーブルを維持しています。私の質問は、パーティション ID はリーダーとレプリカで同じになりますか? 異なる場合、すべてのブローカーでパーティションを一意に識別するにはどうすればよいですか?

Storm トポロジを作成し、Kerberos 対応の環境にデプロイしました。HBase と HDFS にデータを書き込むボルトがありますが、HBase と HDFS の資格情報を取得できません。

以前に同じエラーが発生しましたが、クラスターを再起動することで解決しました。それが正しい解決策だったかどうかは疑問です。

また、ボルトが正常に機能していたのに、クラスターが数日間使用されなかった後に突然例外が発生したという同様のシナリオにも直面しました。

これらの認証情報は一時ディレクトリにキャッシュされていますか?

同じのワーカー ログの下を見つけてください。

メッセージを Kafka にプッシュする Java ベースの Web アプリケーション用の kafka プロデューサーがあります。ドキュメントによると、kafka プロデューサーはスレッドセーフであることがわかりました。Kafkaプロデューサーの単一のインスタンスを持ち、それを異なるスレッド（Webリクエスト）で使用できるということですか？私の場合、それぞれがプロデューサーを開いたり閉じたりします。これにより問題が発生しますか? または、リクエストごとにプロデューサーを開始する方が良いですか?

(新しい Producer APIを使用して) Kafka クラスターに複数のメッセージを連続して発行すると、Futureメッセージごとにプロデューサーから が取得されます。

ここで、プロデューサを構成したと仮定して、最後の未来を待って、以前のすべてが (そして順番に) 配信されたことを確認できますかmax.in.flight.requests.per.connection = 1? retries > 0それとも、すべての先物を待つ必要がありますか? コードでは、これを行うことができますか:

これの代わりに：

ここで何もキャッチされない場合 (最初のスニペットから):

次に、すべてのメッセージがクラスターに順番に格納され(プロデューサーが内部で再試行を実行したかどうかに関係なく)、何か問題が発生した場合、最後の未来ではなくても例外が発生します (私が問題が最初に発生したのは?

他に注意すべき奇妙なコーナー ケースはありますか?

私は、kafka 0.9.0.1 を使用してコンシューマーで上記の例外を取得します。このスレッド [1] によると、server.properties ファイルに「group.max.session.timeout.ms」プロパティが表示されません。

どんな手掛かり？

[1] http://comments.gmane.org/gmane.comp.apache.kafka.user/12426