kafka-producer-api

0熱

1答えて

私はカフカにはとても新しいです。だからこの質問は非常に基本的かもしれない。私が達成しようとしているのは、ブローカが失敗し、その後に追加されたときに再調整にかかる時間を調べることです。私の文書を読む（http://kafka.apache.org/documentation/#basic_ops_restarting）。ブローカーが失敗した場合やmigrate any partitions th

0熱

1答えて

Apache KafkaプロデューサコンシューマAPIの問題

私はカフカプロデューサとコンシューマをテストするための簡単なJava APIを作成しようとしています。 Macマシン上の別の端末でプロデューサとコンシューマを実行しているときにうまく動作します。しかし、私はこのエラーを取得し、Java APIのコードを使用してカフカサーバーに接続しようとしたとき：スレッドで例外kafkatest2.ProducerTest.mainで「メイン」のjava.lan

0熱

1答えて

SSL/TLSリスナーで実行中のKafkaでBlockingChannelを読み取るときに

Kafkaブローカーがポート9093でSSLをリッスンしているときに、自分のコードの1つが例外EOFをスローします。プレーンテキストリスナーでコードスニペットがうまくいきます。何が間違っている可能性がありますか？私は取得しています public KafkaMetadataHelper(String kafkaConnect) throws Exception { // use low

2熱

2答えて

kafka-client 0.8.2.1はsslをサポートしています

バージョンが0.10.2.1であるkafkaブローカーがありますが、Kafkaクライアント0.8.2.1を使用してプロデューサーを作成する必要があります。このクライアントバージョンはSSLセキュリティをサポートしていますか？

-1熱

1答えて

kafkaのファイルからログを読み込む方法は？

私はkafkaのApacheログを読んでから、Spark Streamingにさらに処理したいと思います。私はkafkaを初めて使用しています。私が理解している限り、ログファイルを読み込むプロデューサクラスを作成する必要があります。

1熱

1答えて

カフカは生成時にオフセットを0に戻す

confluent-kafka-goとコンフルエントを介して設定された単一パーティションの単一ブローカーkafkaを使用してgolangに基本チャネルベースのプロデューサを作成しました。 acksを通して反復中に時々生成すると、正常に生成されたメッセージのオフセットはとして受信されます。サンプルログ： { "レベル"： "情報"、 "MSG"： "トピックテストに正常に公開されたが、オフセット0

0熱

1答えて

エラー：「文字列」

コード props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, io

1熱

1答えて

Kafka Consumer - リカバリメソッドでリスナーが受け取ったパラメータを取得します。

私はスプリングカフカコンシューマを構築しています。私は再試行の仕組みを設定しました。再試行が終わったら、失敗したメッセージをデッドレターのトピックにプッシュしたいと思います。を聞く方法は、以下の回復方法の一環として public void listen(@Payload Map<String, Object> conciseMap, @Header(KafkaHeaders.RECE

1熱

1答えて

カフカのプロデューサーは非常に遅い

私はカフカで新しく、私は解決できないという質問があります。私はKafkaとZookeeperをWindows（Linuxではなく）の自分のコンピュータにインストールしました。いくつかのパーティション（6と12のパーティション間で再生）を持つブローカーを作成しました。コンシューマーを作成すると、完璧に動作して読みやすくなりますが、プロデューサーを参照して、多くのWebサイトで見ることができるシン

3熱

1答えて

Kafka Streams：レコードのタイムスタンプ（0.11.0）を変更する方法は？

私はKafkaにメッセージを送るためにFluentD（最後の安定版バージョン12）を使用しています。しかし、FluentDは古いKafkaProducerを使用しているので、レコードのタイムスタンプは常に-1に設定されます。したがって、WallclockTimestampExtractorを使用して、メッセージがカフカに到着した時点のレコードのタイムスタンプを設定する必要があります。私が本当に