apache-kafka-streams

2熱

1答えて

特定のメッセージがトピックに既に存在するかどうかを確認する必要がある状況があります。トピックに重複は絶対に必要ありません。すべてのメッセージを消費してそれらをチェックするのではなく、これを行うための優雅な方法を提案できますか？

3熱

1答えて

ユニットテスト私はProcessorTopologyTestDriverに出くわしたカフカストリーム、残念ながらこのクラスはKTableために利用可能な回避策がありますバージョン0.10.1.0（KAFKA-4408）で破壊し得ているように思われる方法を模索している間問題？ "Mocked Streams"プロジェクトを見たことがありますが、最初はバージョン0.10.2.0を使用していましたが、

0熱

1答えて

カフカの状態ストアロックDBはフォールトトレラントですか？

カフカの状態ストアRock DBはフォールトトレラントですが、変更ログから機能していないデータをどのように復元できますか？

3熱

1答えて

なぜKafka KTableにエントリがないのですか？

私は、Kafka StreamsのKTableを使用する単一インスタンスのJavaアプリケーションを持っています。最近まで私はKTableを使ってすべてのデータを取り出すことができましたが、突然メッセージの一部が消えてしまったように見えました。そこに固有のキーを持つ〜33kのメッセージがあるはずです。私はキーでメッセージを取得したいとき、私はいくつかのメッセージを取得しません。私はReadOnl

0熱

1答えて

どのようにカフカの低レベルのプロセッサのAPIで、MS

そう仮定する断続時間をセッション時間を決定するX分/秒 { props.put("group.max.session.timeout.ms", X*2); props.put("session.timeout.ms", x); props.put("request.timeout.ms", X*2); } を超えていることはカフカのセッションタイムアウトを設定する正しい方法では、低レベ

1熱

1答えて

カフカ出力ストリーム

私は今カフカの消費者とプロデューサーのAPIを使用していますが、ストリームAPIで手を入れたいと思っています。私はオンラインで数多くの参考文献を見てきましたが、この単純なことを理解することはできません。出力トピックにのみメッセージを送信するKStreamを作成するにはどうすればよいですか。たとえば、githubリポジトリにあるこの最も基本的な例を考えてみましょう。 https://github

1熱

1答えて

Kafka State Store

同じカフカトピックから2つのカフカストリームを使用している場合、フォールトトレランスのケースでは、1つのカフカストリームがダウンした場合、カフカの状態ストアは2番目のカフカストリームに複製されます。最初に2番目のカフカストリームの州の店に何が起こるか戻ってきたら。

5熱

1答えて

カフカストリームで手作業でコミットするには？

カフカストリームで手動でコミットする方法はありますか？私は手動でコミット呼んでいる while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records){ // process

1熱

1答えて

複数のインスタンスでKafkaストリーミングが動作しない

Kafka Streamsアプリケーションの複数のインスタンスを実行すると、最初のインスタンスだけが正しくメッセージを受信しています。しかし、新しいインスタンスを開始すると、メッセージは受信されません。この問題を修正するための提案はありますか？ここでは、私はあなたがこの問題を持っていると信じて、私のカフカ・ストリーミング・アプリ package test.kafkastream; im

0熱

2答えて

GenericData.Record用のKafkaAvro Serdeの書き方

私はキーと値の両方のデータのシリアル化にKafka 0.10.2とAvroを使用しています。今私はKafkaストリームを使用したいと思いますが、GenericData.RecordクラスのSerdeクラスを作成しようとしています。 import org.apache.avro.generic.GenericData.Record; import io.confluent.kafka.schem