問題タブ [kafka-producer-api]

私は apache camel と apache kafka を初めて使用し、プロジェクトの小さな POC を行っています。Camel-kafka コンポーネントを使用して kafka から読み取ろうとすると、次の問題のエラー ログが表示されます。

私のJavaコードは次のとおりです。

}

kafka プロデューサー コンソール ツールから txt をいくつか入れて、kafka のキャメル コンポーネントを使用して読み取ろうとしました。

複雑な C 構造をカフカにダンプする必要があるカフカに基づくメッセージ システムを作成しています。まず、単純な構造を作成して送信することで、単純なことをしようとしています。

プロデューサーで:

コンシューマーでも同じ構造を定義しますが、次を使用します。

しかし、それはゴミを印刷します。

i、jは1、1でした

% メッセージ (オフセット 0、8 バイト):

メッセージ ペイロード hexdump (8 バイト):

00000000: 01 00 00 00 01 00 00 00

構造体 p: -469758844 32530

i,j は 256,256 でした

% メッセージ (オフセット 1、8 バイト):

メッセージ ペイロード hexdump (8 バイト):

00000000: 80 00 00 00 80 00 00 00

構造体 p: -469758810 32530

i、j は 4294967295(最大整数)、4294967295(最大整数)

% メッセージ (オフセット 2、8 バイト):

メッセージ ペイロード hexdump (8 バイト):

00000000: ff ff ff ff ff ff ff

構造体 p: -469758776 32530

i,j は 0,0 でした

% メッセージ (オフセット 3、8 バイト):

メッセージ ペイロード hexdump (8 バイト):

00000000: 00 00 00 00 00 00 00 00

構造体 p: -469758742 32530

どうすれば構造を正しくダンプして取得できますか? データをシリアル化する可​​能性があることはわかっていますが、将来的には、より複雑な不透明な構造を送信する必要があります。さらに、シリアル化することで、より多くのメモリを使用する可能性があります。

ありがとう

Scala で 0.9 Kafka Java クライアントを使用しています。

ProducerRecordキーやパーティションを含めたり含めなかったりできるコンストラクタがいくつかあります。

問題ないはずです。

ただし、キーなしProducerRecordでは型エラーが発生します。

これは Kafka の規則に違反しているのでしょうか、それとも、Scala でこの Java API を使用することによる不必要な予防措置でしょうか?

より基本的には、キー付きメッセージとキーなしメッセージを同じ Kafka トピックに入れるのは不適切な形式ですか?

ありがとうございました

Javadoc: http://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/package-summary.html

編集

パラメータの分散を変更すると、これKをKafkaProducer修正できますか?

ファイルからデータを読み取るために公式ドキュメントに従って実装しています。kakf-connect

私はカフカを完全に実行しています。メッセージを送受信するプロデューサーとコンシューマー。

ただし、次のコマンドを実行すると:

次のエラーが表示されます。

エラー: メイン クラス org.apache.kafka.connect.cli.ConnectStandalone が見つからないか、読み込めませんでした

チェックを入れたところ、ファイルConnectStandalone hereがありますconnect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectStandalone。

私のconnect-file-source.propertiesは次のとおりです。

何か足りないものはありますか？このエラーを取り除くにはどうすればよいですか?

メッセージのバッチを取得するソースがあります。これらのメッセージは確実に Kafka に追加する必要があります。

aync プロデューサーを使用すると、多くのメッセージを送信するときに、パーティションがしばらくダウンしていないかどうか疑問に思います。そのメッセージをスキップして次のメッセージを送信するだけで、メッセージの順序が正しくありません。

Kafka に指示できる方法はありますか?一連のメッセージをバッチで生成し、アトミックにすべてを渡すか、すべてを失敗させるか?

*スループットに深刻な影響を与えるため、同期プロデュースは実行したくありません。

Kafka Async プロデューサーを使用している場合、バッファーに X 個のメッセージがあると仮定します。それらがクライアントで実際に処理され、ブローカーまたは特定のパーティションがしばらくダウンしている場合、kafka クライアントは再試行し、メッセージが失敗した場合、特定のメッセージを失敗としてマークし、次のメッセージに移動します (これは可能性があります)。順不同のメッセージにつながる）？または、順序を維持するために、バッチ内の残りのメッセージは失敗しますか?

次に順序を維持するので、理想的には、失敗した場所からバッチを失敗させて、失敗したポイントから再試行できるようにしたいのですが、どうすればそれを達成できますか?