kafka-consumer-api

0熱

1答えて

CKANのリソースをKafka consume-APIにリンクすることはできますか？私はリアルタイムでKafkaで公開されているライブリソースをCKAN APIでアクセスしたいと考えています。

0熱

1答えて

を読み取りますプロデューサー： object ProducerApp extends App { val topic = "topicTest" val props = new Properties() props.put("bootstrap.servers", "localhost:9092") props.put(ConsumerConfig.GROUP_ID_CONFIG, "

1熱

1答えて

カフカリバランス。重複した処理の問題

私はコンシューマワーカーアプリケーションを持っています。これは内部でX個のスレッドを起動しています。各スレッドはそれをKafkaCosnumerにしています。 Cosnumersは同じgroupIdを持ち、同じトピックを購読しています。そのため、各消費者はパーティションの公平な分配を得ることができます。処理の性質上、メッセージを失うことはできず、重複を許可することもできません。私が走っているカフ

0熱

2答えて

カフカJavaコンシューマーは既に閉店

私はカフカを使い始めました。私は消費者に小さな問題に直面しています。私はJavaで消費者を書いた。この例外が発生します - IllegalStateExceptionこのコンシューマは既に閉じられています。私は次の行に例外を取得： ConsumerRecords<String,String> consumerRecords = consumer.poll(1000); 私の消費者は、いくつ

0熱

1答えて

Kafka - 空になるまで消費する

KafkaConsumerのすべてのコンシューマーレコードがフェッチされるまで続けることが重要なユースケースがあります。このユースケースでは、パイプラインに入ることはありません。絶対的に確実にフェッチすることは何もないことを保証するための適切な方法は何ですか？

1熱

1答えて

最後のメッセージをカフカのトピックに送信

私はカフカを初め、独自のストリーミングサービスをカフカに接続するためのプロトタイプを作成しています。社内のストリームコンシューマは、接続時に最後に受信したメッセージのIDでログオンする必要があるため、トピックに送信された最後のメッセージのキーを取得したいと考えています。 KafkaProducerまたはKafkaConsumerを使用してこれを行うことは可能ですか？コンシューマを使用して次の操

0熱

1答えて

カフカコンシューマ用にいくつのパーティションを作成する必要がありますか？

現在、トピックのパーティションは1つしかなく、消費者は1つしかありません。しかし、私の消費者は低迷しているので、新しい消費者を同じトピックに追加したいと考えています。しかし、私は同時に、すべての消費者が排他的なメッセージセットを処理したいと思っています。スループットを増やして排他メッセージを処理するたびにコンシューマを追加できるように、作成するパーティション数を決定できません。

0熱

1答えて

KafkaConsumerがオフセットから読み取らない0

私はカフカの例をテストしたいと思います。私はプロデューサーカフカ0.10.0.1を使用しています： object ProducerApp extends App { val topic = "topicTest" val props = new Properties() props.put("bootstrap.servers", "localhost:9092") props.put

0熱

1答えて

カフカパーティションラグを増やす

私はキューとしてKafka 1.0を使用するアプリケーションを持っています。カフカのトピックには80のパーティションと80のコンシューマが走っています。（Kafka-pythonの消費者）。コマンドを実行することにより：私はパーティションの1つがオフセットで立ち往生し、新しいレコードが追加されているようタイムラグが連続的に増加していることを見 ./bin/kafka-consumer-grou

0熱

1答えて

Spring Kafka： `onMessage`を使用して通知するのではなく、新しいメッセージをポーリングする

私はカフカのメッセージを消費するためにSpringベースのプロジェクトで自然な選択と思われたので、プロジェクトでSpring Kafkaを使用しています。メッセージを消費するには、MessageListenerインターフェイスを使用できます。 Spring Kafkaは内部的に、新しいメッセージごとに私のonMessageメソッドを呼び出すように気を付けます。私の設定では、新しいメッセージを明示