在 Java 中创建 kafka 主题

Question

我正在尝试动态创建 kafka 主题，因为我创建了一个将输入作为主题名称并应该创建主题的方法。

我正在使用 kafka 2.12 和 Java 1.8 以下是我尝试过的......

private static void CreateKafkaTopic(String topicName) {
    ZkClient zkClient = null;
    ZkUtils zkUtils = null;
    try {
        String zookeeperConnect = "localhost:2181";
        int sessionTimeOutInMs = 15 * 1000; // 15 secs
        int connectionTimeOutInMs = 10 * 1000; // 10 secs
        zkClient = new ZkClient(zookeeperConnect, sessionTimeOutInMs, connectionTimeOutInMs, ZKStringSerializer$.MODULE$);
        boolean isSecureKafkaCluster = false;
        zkUtils = new ZkUtils(zkClient, new  ZkConnection(zookeeperConnect), isSecureKafkaCluster);
        Properties topicConfig = new Properties();
        AdminUtils.createTopic(zkUtils, topicName, 1, 1, topicConfig,RackAwareMode.Disabled$.MODULE$);
    } catch (Exception ex) {
        ex.printStackTrace();
    } finally {
        if (zkClient != null) {
            zkClient.close();
        }
    }
}

但是当我尝试这个时，我收到如下错误

线程“主”java.lang.NoSuchMethodError 中的异常： scala.Product.$init$(Lscala/Product;)V at kafka.admin.RackAwareMode$Disabled$.(RackAwareMode.scala:27​​) 在 kafka.admin.RackAwareMode$Disabled$.(RackAwareMode.scala) 在 com.OTMProducer.CreateKafkaTopic(OTMProducer.java:243)

请告诉我这是什么原因。我提到了一些已经可用的答案，例如How To create a kafka topic from java for KAFKA-2.1.1-1.2.1.1?

Answer 1

此错误通常意味着您缺少 Scala 库。将Scala 2.12 添加到您的路径应该可以解决问题。

---

2.12 不是 Kafka 的版本，而是 Scala 的版本。 Kafka 的版本是文件名中的下一个数字。比如kafka_2.11-1.1.0中，Kafka版本是1.1.0，2.11是Scala版本。

假设您运行的是最新的 Kafka 版本 >= 0.11，以编程方式创建主题的推荐方法是通过 Kafka Admin API。这避免了直接写入 Zookeeper 并且也不依赖于 Scala（因此您不必添加它）。

在 AdminClient 的 javadoc 中查看 AdminClient.createTopics()