問題タブ [confluent-platform]

私はKafkaProducerテストケースで使用しており、プロデューサーは のschemaRegistryUrlローカルインスタンスを指す を使用していますSchema Registry。KafkaProducerスキーマ レジストリとの接続方法をモックする方法はありますか? つまりKafkaProducer/Consumer、スキーマ レジストリのインスタンスを実行せずにテストで動作させることです。

Confluent は、デフォルトでこの JMSSourceConnector for Kafka トピックを提供します。

それとも、このためにカスタム コネクタを作成する必要がありますか?

これに関する Confluent ページにドキュメントはありません。

私はカフカが初めてで、Confluent プラットフォームに興味があります。Confluent プラットフォームでのユーザー ストーリーはあまりないようです。Confluent プラットフォームと Apache Kafka の違いは何ですか?

2 つのクラスターがあります。

• コンフルエントな家の中の1つ（3.0.0-1）
• AWS に 1 つ、hadoop (hdp 2.4) を使用

hdfs コネクタを使用して、コンフルエントから Hadoop に書き込もうとしています。

簡単に言えば、コネクタは、ホスト名を使用する代わりに、hadoop クラスターのプライベート IP に接続しようとします。社内クラスターでは、/etc/hosts が更新され、内部 Hadoop ホスト名が関連するパブリック IP に解決されます。

私は分散コネクタを使用しています。次のようなコネクタ JSON ファイルがたくさんあります。

ワーカーは次のように定義されます。

いくつかのメモ:

• /kafka-connect は hdfs に存在し、誰でも書き込み可能
• 3 つのトピック (*.storage.topic) は存在します
• Kafkaブローカーを使用して各（3）サーバーで1つのワーカーを実行しています（すべてのブローカーにスキーマレジストリ、残りのAPI、およびzookeeperサーバーがあります）
• dfs.client.use.datanode.hostname を true に設定しました。このプロパティはクライアントの $HADOOP_HOME/hdfs-site.xml に設定されています。

/kafka-connect のサブディレクトリとハイブ メタデータが作成されていることがわかります。コネクタを起動すると、次のメッセージが表示されます。

createBlockOutputStream での INFO 例外 (org.apache.hadoop.hdfs.DFSClient:1471) org.apache.hadoop.net.ConnectTimeoutException: チャネルの接続準備が整うまでの待機中に 60000 ミリ秒のタイムアウトが発生しました。ch : java.nio.channels.SocketChannel[org.apache.hadoop.net.NetUtils.connect(NetUtils.java:533) の接続保留リモート org.apache.hadoop.hdfs.DFSOutputStream.createSocketForPipeline(DFSOutputStream.java: 1610) org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.createBlockOutputStream(DFSOutputStream.java:1408) org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.nextBlockOutputStream(DFSOutputStream.java:1361) org.apache.hadoop .hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:588) INFO 破棄 BP-429601535-10.0.0.167-1471011443948:blk_1073742319_1495 (org.apache.hadoop.hdfs.DFSClient:

これを修正する方法について何か考えはありますか? コンフルエントは、ホスト名ではなく IP を直接受け取るようです。

scala 2.10 パッケージが必要なので、confluent-2.0.0-2.10.5.tar.gz をダウンロードしました。

しかし、/share/java/schema-registry の kafka jar はまだ kafka_2.11-0.9.0.0-cp1.jar です。

とにかくクリーンな 2.10 scala confluent パッケージを入手できますか?

Windows で kafka レスト プロキシを実行する方法。

confluent-2.0.1-2.11.7.ta​​r.gz をダウンロードしました

Windows フォルダーに kafka-rest-start が表示されません。

次のシナリオがあります。

• Producer は、 http://docs.confluent.io/3.0.0/kafka-rest/docs/intro.html#produceで説明されているように、Confluent の REST プロキシ (Confluent のスキーマ レジストリにスキーマを登録する) を介して、Avro でエンコードされたメッセージを Kafka トピックに送信します。 -and-consume-avro-メッセージ
• Spring Cloud Stream 対応のメッセージは、トピックで新しいメッセージをリッスンします

私のアプリケーションは次のようになります。

一方、MyMessage は、Avro スキーマから Avro によって作成されたクラスです。

私のapplication.propertiesは次のようになります：

私の問題は、新しいメッセージを受信するたびに、次の例外がスローされることです。

私が理解していることから、問題は、Confluent スタックがメッセージ ペイロードの一部としてメッセージのスキーマの ID を含み、クライアントがスキーマ ID の後に実際の Avro メッセージの読み取りを開始することが期待されることです。Confluent の KafkaAvroDeserializer を使用するように Kafka バインディングを構成する必要があるようですが、これを実現する方法がわかりません。

(Confluent の avro コンソール コンシューマを使用してメッセージを完全に取得できるため、Avro エンコーディングの問題ではないようです)

また、@EnableSchemaRegistry アノテーションをいじって ConfluentSchemaRegistryClient Bean を構成しようとしましたが、これはスキーマが格納/取得される場所のみを制御し、実際の逆シリアル化は制御しないように見えます。

これはどういうわけか機能しているはずですか？

私はコンフルエントな kafka-rest を使用して avro メッセージを生成します。rest エンドポイントを使用してそれを消費すると、返される形式は次のようになります。

カフカのメッセージは実際にはこの形式なのか、それとも「キー」と「パーティション」が消費者エンドポイントによって追加されているのだろうか。前者の場合は奇妙になります。この場合、プロデュース リクエストを実行するときにレスポンス ペイロード avro スキーマが同じではないためです (ACTUAL_PAYLOAD に一致するスキーマ)。