kafka-consumer-api

2熱

2答えて

私はkafkaでメッセージを生成/消費するためにC++ kafka libを使用しています。ここで、切断や障害を処理するために消費者を監視したいと考えています。私はすべての消費者の接続リストの種類を探しています。カフカdocumentationから：消費イドレジストリグループ内のすべての消費者によって共有されるGROUP_IDに加えて、各消費者は、過渡、ユニークconsumer_idが与え

1熱

1答えて

KafkaはNoSuchElementExceptionを返します

以下に言及するJavaコードによって複数のパーティションを持つトピックを作成しました。 Javaコードを実行すると private void existOrCreateTopicsWithPartition(Properties prop) { String kafkaServer = prop.getProperty(KafkaProducerKey.BOOTSTRAP_SERVERS

0熱

1答えて

が、私はセットアップにAppAのように同じキューを有する三の同様のアプリケーションのためのカフカのクラスタ型のセットアップをしたいカフカクラスタ設定

用ソケットエラーを読んでタイムアウトしまし入門 - > {TopicX、TopicY、TopicZ}、AppBを - > {TopicX、TopicZ}、 AppC - > {TopicX、TopicY}。プロデューサーとコンシューマーはアプリ固有のものになります。 3つのブローカーが異なるポートを持つ三つの異なる設定ファイル内のパーティション1,2,3を有する Iセットアップカフカクラスタ。そし

1熱

2答えて

Kafka消費者グループオフセット保存

kafkaは、そのグループのすべての消費者が失敗した後、消費者のオフセットをどのくらい保存しますか？これには設定変数がありますか？

0熱

1答えて

なぜカフカ消費者が最初のメッセージを聞いていないのですか？

私のテストプログラムでは、私はリスナーを開始します。その後、ループでメッセージを送信します。 1つのメッセージを送信しても、そのメッセージをリスニングしていません。 2つのメッセージを送信すると、1つのメッセージを受信します.3を送信すると、2つのメッセージを受信します。なぜですか？プロデューサー KeyedMessage<String, byte[]> message = new KeyedM

2熱

1答えて

カフカ自動コミットインターバルのベストプラクティス

auto.commit.interval.msの値を選択する際のベストプラクティスはありますか？私がいることをhereを読む：一般にそれが大幅に飼育係に読み取り/書き込み速度を高め、それがその全体で強く一貫ですので、飼育係が鈍化しますので、小さすぎるこの間隔を維持することが推奨されていません定足数。何が小さすぎますか？これはまだカフカ> = 0.9.0バージョンの問題ですか？

0熱

1答えて

高レベルのカフカ消費者を一時停止する

私はある条件で一時停止したいと思っているカフカの消費者を持っていて、後でそれを再開して前のすべてのメッセージを消費します。 1つのアイデアは、他のスレッドによって更新され、消費する前に共有フラグを使用することです。すなわち、iterator.next().message()フラグの値をチェックします。それが本当であれば、他のものを消費しないでください。ちょうど私が正しい方向に考えているのか、それと

1熱

1答えて

カフカDSL - マニュアルメッセージでコミットする

以下のコードでSpring DSL Kafkaを使用して手動コミットを実装しようとしています。私たちはそれへの参照を見つけることができませんでした。「Auto.commit」というコンシューマの属性を「False」として追加できますが、メッセージを正常に処理したらコミットしたかったのです。これを助けることができる人はいますか？ @Bean IntegrationFlow consum