在恢复所有主题的分区时,我不断收到调试消息。像下面。此消息每毫秒在我的服务器上连续打印一次。
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-7
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-6
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-9
08:44:34.850 [default-akka.kafka.default-dispatcher-10] DEBUG o.a.k.clients.consumer.KafkaConsumer - Resuming partition test222-8
这个
这是代码
val zookeeperHost = "localhost"
val zookeeperPort = "9092"
// Kafka queue settings
val consumerSettings = ConsumerSettings(system, new ByteArrayDeserializer, new StringDeserializer)
.withBootstrapServers(zookeeperHost + ":" + zookeeperPort)
.withGroupId((groupName))
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
// Streaming the Messages from Kafka queue
Consumer.committableSource(consumerSettings, Subscriptions.topics(topicName))
.map(msg => {
consumed(msg.record.value)
})
.runWith(Sink.ignore)
请帮助正确进行分区以停止 DEBUG 消息。
最佳答案
好像是reactive-kafka code在开始获取之前恢复每个分区:
consumer.assignment().asScala.foreach { tp =>
if (partitionsToFetch.contains(tp)) consumer.resume(java.util.Collections.singleton(tp))
else consumer.pause(java.util.Collections.singleton(tp))
}
def tryPoll{...}
checkNoResult(tryPoll(0))
如果分区之前没有暂停,KafkaConsumer.resume 方法是无操作的。
关于scala - Akka.Kafka - 警告消息 - 恢复分区,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41019323/