kafka-consumer-api

0熱

1答えて

Spring Kafkaクライアントを使用してKafkaサーバーに接続しようとしています&ログに以下のエラーが表示されます。私は特定のkafkaブローカーIPに接続しています。私は以下のエラーログにローカル参照が表示されているようです。これをさらにデバッグするのを手伝ってください。私はlogback.xmlを使用してプッシュしています。以下はその構成です。私はテストの前にカフカホストとポートをハ

0熱

1答えて

4つのパーティションを持つ単一プロデューサ用の複数のコンシューマセットアップKafka Java

4つのパーティションを持つ単純なプロデューサを作成し、コンシューマグループに4つのコンシューマを作成して各パーティションのデータを消費するようにしました。どうすればいいですか？コンシューマコードコードの public class KafkaConsumer { static List<String> list = new ArrayList<String>(); publi

0熱

1答えて

Kafka：バージョン0.11のConsumerConnectorの新しいAPI

私は0.8.2.0から0.11.0.0にKafkaクライアントを更新しています。私の古いコードでは、createMessageStreamsメソッドを使用してメッセージストリームを取得し、各トピックのストリームを繰り返し処理するのには、ConsumerConnectorを使用します。しかし、ConsumerConnectorは新しいAPIで償却されているようです。 package kafka.c

1熱

2答えて

削除の10秒後にカフカの話題が再現されています

私はカフェの話題の削除に関する問題に直面しています。トピックを作成/削除し、メッセージを消費する&を生成するためのkafka rest APIの使用。トピックが削除されるトピックを削除しようとしましたが、しばらくしてから10秒後にトピックが再表示されます。消費者グループのオフセットをチェックしており、LAGが負の値としてリストされています。ドッキングウィンドウの実行--net =ホスト--r

0熱

1答えて

kafka接続がポーリング中にコンシューマグループスレッドから拒否されました

2つのパーティションと1つのレプリカでそれぞれカフカトピックXとYが作成されました。上記のkafkaコンテナと9092とzookeeperで動作する2つのJavaプロセス（mavenモジュール）がkafka-clientの2つあります。 JVMは両方ともkafkaに正常に接続しており、両方ともトピックが存在するかどうかを確認できます。しかし、一方のJVMはコンシューマ・グループを正常に登録できま

0熱

2答えて

Kafkaは多数のパーティション（64k）でトピックを作成できません

WindowsでKafka 0.10.1.0を実行しています。 kafka-topics.batを使用して64kパーティションの大きなトピックを作成しようとしています。しかし、エラーで作成できません - [2017-10-26 15:35:01,232] WARN Session 0x25f59be4474000f for server <ZK server name>/<ZK Server IP

3熱

1答えて

kafkalistenerでオフセットを設定する方法

私はkafkaサーバーからデータを消費しようとしていますが、私は@Kafkalistener注釈を使用しています。問題は、アプリケーションを再起動するたびにすべてのメッセージが表示されてしまうことです。アプリケーションで最後に消費されたオフセットを保存して、次のメッセージを消費するにはどうすればよいですか？

2熱

1答えて

Kafka Streams：データ型の変換方法は？

私はトポロジービルダーでKafkaのStreams APIを使用しています。あるデータ型を別のデータ型に変換できるプロセッサを、パイプライン内の次のプロセッサが使用できるようにするには、どうすればいいのか教えてください。 [topic]--(string)-->[processor: parse json]--(object)-->[processor 2]--(object)-->[sink]

0熱

2答えて

今後のKafkaバージョンでSimpleConsumerは減価償却されますか？高レベルの消費者をマニュアルオフセット制御に使用することは良いでしょうか？

kafka 0.10と0.11の最新バージョンでSimpleConsumer Supportが見つかりません。使用例：各トピックのオフセットを手動で制御し、外部ソースに保存したいと考えています。私は世論調査を行うと 1）私はメッセージを取得することができる午前、が、私は読み方多くのメッセージを制御できません。ハイレベルの消費者との問題は、上記のユースケースのためにあるのです。 2）ポーリング

0熱

1答えて

承認後のカフカ消費者の回復.ACKnowledge（）;は呼び出されていません

私はAcknowledgingMessageListenerを実装するシンプルなKafkaコンシューマを実装しました。 acknowledgment.acknowledge（）があります。のonMessageメソッドで呼び出され、すべてがうまくいったり、回復可能な例外があったときに呼び出されます。この場合、すべてが機能します。しかし、物事が間違っていて、エラーがスローされる（私は捕まえない）かも