检查系统是否连接到Kafka

Question

但是它太复杂了，无法从Java代码中完成，我认为这不是我应该使用的方向。

提前谢了。

Answer 1

我有同样的问题，我不想离开这个问题没有任何答案。我读了很多关于如何检查连接的文章，我找到的大多数答案都是用Zk检查连接，但我真的想直接用Kafka服务器检查连接。

我所做的是创建一个简单的KafkaConsumer，并用listTopics（）列出所有主题。如果连接是成功的，那么你会得到一些东西作为回报。否则，将获得TimeoutException。

  def validateKafkaConnection(kafkaParams : mutable.Map[String, Object]) : Unit = {
    val props = new Properties()
    props.put("bootstrap.servers", kafkaParams.get("bootstrap.servers").get.toString)
    props.put("group.id", kafkaParams.get("group.id").get.toString)
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val simpleConsumer = new KafkaConsumer[String, String](props)
    simpleConsumer.listTopics()
  }

然后，可以将此方法包装在try-catch语句中以捕获异常。