Создать тему кафки на Java

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();
        }
    }
}

но когда я пытаюсь это сделать, я получаю ошибку, как показано ниже

Исключение в потоке "main" java.lang.NoSuchMethodError: scala.Product.$ init $ (Lscala / Product;) V в kafka.admin.RackAwareMode $ Disabled $. (RackAwareMode.scala: 27) в kafka.admin.RackAwareMode $ Disabled $. (RackAwareMode.scala) в com.OTMProducer.CreateKafkaTopic (OTT).java: 243)

Пожалуйста, предложите мне причину этого.Я сослался на некоторые уже доступные ответы, такие как Как создать тему kafka из java для KAFKA-2.1.1-1.2.1.1?

Answer 1

Эта ошибка обычно означает, что вам не хватает библиотеки Scala.Добавление Scala 2.12 к вашему пути должно решить проблему.

---

2.12 - не версия Кафки, а версия Scala.Версия Кафки - это следующий номер в имени файла.Например, в kafka_2.11-1.1.0 версия Kafka - 1.1.0, а версия 2.11 - версия Scala.

Если вы используете последнюю версию Kafka> = 0.11, рекомендуемый способ создания тем программным способом - черезAPI администратора Kafka.Это позволяет избежать прямой записи в Zookeeper и также не зависит от Scala (поэтому вам не нужно добавлять его).

См. AdminClient.createTopics() в javadoc AdminClient