Replacement for SimpleConsumer.send in kafka?



simpleconsumer已在kafka中弃用,org.apache.kafka.clients.consumer.kafkaconsumer是替代品。但是,它没有发送(...)函数。如何使用新的kafkaconsumer重写以下代码?

import scala.concurrent.duration._
import kafka.api.TopicMetadataRequest
import kafka.consumer.SimpleConsumer
....
val consumer = new SimpleConsumer(
    host = "127.0.0.1",
    port = 9092,
    soTimeout = 2.seconds.toMillis.toInt,
    bufferSize = 1024,
    clientId = "health-check")
// this will fail if Kafka is unavailable
consumer.send(new TopicMetadataRequest(Nil, 1))

您可以使用.partitionsFor.listTopics

获得主题元数据

没有直接替换方法,这取决于您要做什么。如果您需要所有分区信息,则该方法有用于消费者的方法。

最新更新