問題タブ [kafka-consumer-api]

私は kafka-0.8.1.1 バージョンを使用していますが、このバージョンでは自動再接続は Java コード経由では機能しません。プロパティファイルを次のように持っています

コンシューマースレッドを開始している間、それはカフカサーバーに接続されています。サーバーを実行しているスレッドの間にカフカサーバーへの接続が2時間失われます。再接続すると、カフカコンシューマーはメッセージを受信せず、エラーもスローしませんフロー。

接続が失われたことを検出して、消費者スレッドを自動的に再接続するにはどうすればよいでしょうか。

前もって感謝します

嵐のトポロジを書きました。私は基本的に avro スキーマのタプルをバイト配列の形式で kafka トピックに送信したいと考えています。

これは私がボルトを設定する方法です：

そして、これは私がバイト配列に変換する方法です

次の方法でタプルを発行すると、kafka トピックに何も表示されません (バイトストリームを kafka に送信します):

しかし、代わりに、バイト配列を文字列に変換してからkafkaトピックに変換すると、機能します:

以下のようなもの：

私は何を間違っていますか？ストームカフカボルトを使用してバイトストリームをカフカトピックに送信するにはどうすればよいですか?

kafka-net プラグインを備えた Red Hat VM で kafka 0.8.1.1 を使用しています。カフカからの以前のメッセージの受信を停止するようにコンシューマーを構成するにはどうすればよいですか?

私の消費者コード:

シンプルなコンシューマーを使用して kafka トピックから読み取り、フェッチ サイズがトピック内のメッセージのサイズにどのように関連しているかについて質問があります。

たとえば、トピック内の各メッセージが 10 KB であるとします。

オフセット 0 から 16 kb のフェッチ サイズで消費する場合 (この質問のために)、1 つの完全なメッセージと 6kb の別の部分的なメッセージを読み取りますか?

リンクには次のように記載されていますが、それが起こっているとは思いません

最適化として、サーバーはメッセージ セットの最後に部分的なメッセージを返すことができます。クライアントはこのケースを処理する必要があります。

単一のノードで kafka をセットアップし、zookeeper と kafka サーバーを開始しました。コンソールで内部プロデューサーとコンシューマーをテストしましたが、うまく動作します。しかし、コンソールで内部 kafka コンシューマーとカスタム プロデューサーを実行しているとき動作しません。

以下は私のプロデューサークラスです

コントロールが producer.send() に到達すると、次の例外を除いて 3 回試行した後に停止します

次のように、kafka-console-consumer.sh を使用してメッセージを消費している場所から Kafka アプリケーションがあります。

これにより、私が Kafka コンシューマーを介して Kafka ブローカーに書き込むすべてのメッセージが見落とされずに提供されます。

最近、zookeeperhost にアクセスできない (何らかの理由で) 別の環境にアプリケーションをデプロイしました。したがって、代わりに以下のように kafka-simple-consumer-shell.sh を使用しています。

しかし、これを使用すると、ほとんどのメッセージ (5000 件中 2 ～ 4 件) が見落とされます。誰かが kafka-simple-consumer-shell.sh がメッセージを読み取る方法を説明してください。

おそらくいくつかのメッセージが別のパーティションに送られているのではないかと疑っています.パーティション0から読み取っているだけなので、毎回すべてのメッセージを取得しているわけではありません. しかし、パーティションの数を確認する方法がわかりませんか? 他のパーティションのIDは何ですか？1でやってみましたがダメでした。

誰か助けてください。

私は単純なカフカ プロデューサー コンシューマー クライアント API を試しています。コンソールからコンシューマーにメッセージが表示されるため、私のプロデューサー クラスは正常に動作しますが、コンシューマー コードを実行しているときに何も表示されず、何が問題なのか、どこにあるのかわかりません。間違いをしている

これはプロデューサーコードです -

消費者クラスは -

確認のために、testConsumer() メソッドで sysout を使用して 2 つのチェックを適用したため、実行中に check1 のみが表示されます。つまり、コードが check2 に到達していません。何らかの問題があると思います。consumer.createMessageStreams(topicCount);その理由と解決方法を教えてください。 ?