小编Nan*_*dia的帖子

我已经在node.js中实现了Avro模式，该模式随消息有效负载一起发送。而且工作正常。我在寻找是否可以通过Kafka-node模块使用架构注册表。我进行了探索，但未成功找到任何东西。

在每个消息中发送模式会增加消息大小吗？与使用架构注册表相比，它会影响性能吗？

任何帮助，将不胜感激。

我正在运行一个Kafka Stream应用程序,该应用程序使用来自2个主题的数据并将加入/合并的结果输出到3个主题中.kafka主题有15个分区和3个复制因子.我们有5个卡夫卡经纪人和5个动物园管理员.我正在运行15个Kafka Stream应用程序实例,因此每个应用程序可以有1个分区.Kafka版本 - 0.11.0.0

我在我的kafka流应用程序中得到以下异常:

org.apache.kafka.clients.consumer.CommitFailedException:由于该组已经重新平衡并将分区分配给另一个成员,因此无法完成提交.这意味着后续调用poll()的时间长于配置的max.poll.interval.ms,这通常意味着轮询循环花费了太多时间进行消息处理.您可以通过增加会话超时或通过max.poll.records减少poll()中返回的批量的最大大小来解决此问题.在org.apache的org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.sendOffsetCommitRequest(ConsumerCoordinator.java:725)org.apache.kacheka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:604)org.apache .kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1173)位于org.apache.kafka.streams.processor的org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:307) .internals.StreamTask.access $ 000(StreamTask.java:49)位于org.apache.kafka.streams.processor.internals的org.apache.kafka.streams.processor.internals.StreamTask $ 1.run(StreamTask.java:268) .StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)org.apache.kafka.streams. (StreamTask.java:362)org.apache.kafka.stream.process.Stat..processor.internals.StreamThread $ 3.apply(StreamThread.java:1118)位于org.apache.kafka.streams.processor的org.apache.kafka.streams.processor.internals.StreamThread.performOnStreamTasks(StreamThread.java:1448). ingals.StreamThread.suspendTasksAndState(StreamThread.java:1110)atg.apache.kafka.streams.processor.internals.StreamThread.access $ 1800(StreamThread.java:73)org.apache.kafka.streams.processor.internals.StreamThread $ RebalanceListener.onPartitionsRevoked(StreamThread.java:218)org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinPrepare(ConsumerCoordinator.java:422)org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded (AbstractCoordinator.java:353)org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310)at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java) :297)org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(Ka)fkaConsumer.java:1078)org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:582) org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:553)at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)2017-08- 09 14:50:49 - [ERROR] [click-live-StreamThread-1] [org.apache.kafka.streams.processor.internals.StreamThread.performOnStreamTasks:1453]:

有人可以帮助解决原因和解决方案吗？

此外,当我的一个kafka经纪人失败时,我的kafka流应用程序没有连接到其他经纪人？我已经设定brokers.list=broker1:9092,broker2:9092,broker3:9092,broker4:9092,broker5:9092

我有一个 Kafka Streams 应用程序，它从几个主题获取数据并连接数据并将其放入另一个主题中。

卡夫卡配置：

5 kafka brokers
Kafka Topics - 15 partitions and 3 replication factor. 

注意：我在运行 Kafka Brokers 的同一台机器上运行 Kafka Streams 应用程序。

每小时消耗/产生几百万条记录。每当我关闭任何卡夫卡经纪人时，它就会进入重新平衡状态，并且需要大约。重新平衡需要 30 分钟，有时甚至更长。

有人知道如何解决 kafka 消费者的重新平衡问题吗？此外，很多时候它在重新平衡时会抛出异常。

这导致我们无法使用此设置在生产环境中运行。任何帮助，将不胜感激。

Caused by: org.apache.kafka.clients.consumer.CommitFailedException: ?
Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time …

我想知道什么最适合我：Kafka流或Kafka消费者api或Kafka connect？

我想从主题中读取数据，然后进行一些处理并写入数据库。因此，我已经编写了使用者，但我觉得我可以编写Kafka流应用程序，并使用其有状态处理器执行任何更改并将其写入数据库，从而消除了我的使用者代码，而只需要编写db代码。

我要插入记录的数据库是：HDFS-（插入原始JSON）MSSQL-（已处理json）

另一个选择是Kafka connect，但我发现hdfs sink和jdbc sink连接器目前没有json支持（我不想用avro编写），创建架构对于复杂的嵌套消息也很痛苦。

还是我应该编写自定义Kafka connect来做到这一点。

那么，您是否需要考虑是否应该编写Kafka使用者或Kafka流或Kafka connect？在性能方面有哪些更好的选择，而开销却更少呢？