Как мы можем создать тему в Кафке из IDE, используя API

Question

Как мы можем создать тему в Kafka из IDE, используя API, потому что, когда я делаю это:

bin/kafka-create-topic.sh --topic mytopic --replica 3 --zookeeper localhost:2181

Я получаю ошибку:

bash: bin/kafka-create-topic.sh: No such file or directory

И я следовал за настройкой разработчика как есть.

Answer 1

В Kafka 0.8.1+ - последней на сегодня версии Kafka - вы можете программно создать новую тему через AdminCommand, Функциональность CreateTopicCommand (часть старой Kafka 0.8.0), которая упоминалась в одном из предыдущих ответов на этот вопрос, была перемещена в AdminCommand,

Пример Scala для Кафки 0.8.1:

import kafka.admin.AdminUtils
import kafka.utils.ZKStringSerializer
import org.I0Itec.zkclient.ZkClient

// Create a ZooKeeper client
val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
// Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
// createTopic() will only seem to work (it will return without error).  The topic will exist in
// only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
// topic.
val zkClient = new ZkClient("zookeeper1:2181", sessionTimeoutMs, connectionTimeoutMs,
    ZKStringSerializer)

// Create a topic named "myTopic" with 8 partitions and a replication factor of 3
val topicName = "myTopic"
val numPartitions = 8
val replicationFactor = 3
val topicConfig = new Properties
AdminUtils.createTopic(zkClient, topicName, numPartitions, replicationFactor, topicConfig)

Постройте зависимости, используя sbt в качестве примера:

libraryDependencies ++= Seq(
  "com.101tec" % "zkclient" % "0.4",
  "org.apache.kafka" % "kafka_2.10" % "0.8.1.1"
    exclude("javax.jms", "jms")
    exclude("com.sun.jdmk", "jmxtools")
    exclude("com.sun.jmx", "jmxri"),
  ...
)

РЕДАКТИРОВАТЬ: Добавлен пример Java для Kafka 0.9.0.0 (последняя версия от января 2016 года).

Maven зависимости:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.9.0.0</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.7</version>
</dependency>

Код:

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

import java.util.Properties;

import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

public class KafkaJavaExample {

  public static void main(String[] args) {
    String zookeeperConnect = "zkserver1:2181,zkserver2:2181";
    int sessionTimeoutMs = 10 * 1000;
    int connectionTimeoutMs = 8 * 1000;
    // Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
    // createTopic() will only seem to work (it will return without error).  The topic will exist in
    // only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
    // topic.
    ZkClient zkClient = new ZkClient(
        zookeeperConnect,
        sessionTimeoutMs,
        connectionTimeoutMs,
        ZKStringSerializer$.MODULE$);

    // Security for Kafka was added in Kafka 0.9.0.0
    boolean isSecureKafkaCluster = false;
    ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperConnect), isSecureKafkaCluster);

    String topic = "my-topic";
    int partitions = 2;
    int replication = 3;
    Properties topicConfig = new Properties(); // add per-topic configurations settings here
    AdminUtils.createTopic(zkUtils, topic, partitions, replication, topicConfig);
    zkClient.close();
  }

}

РЕДАКТИРОВАТЬ 2: Добавлен пример Java для Kafka 0.10.2.0 (последняя версия от апреля 2017 года).

Maven зависимости:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.11</artifactId>
    <version>0.10.2.0</version>
</dependency>
<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.9</version>
</dependency>

Код:

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;

import java.util.Properties;

import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

public class KafkaJavaExample {

  public static void main(String[] args) {
    String zookeeperConnect = "zkserver1:2181,zkserver2:2181";
    int sessionTimeoutMs = 10 * 1000;
    int connectionTimeoutMs = 8 * 1000;

    String topic = "my-topic";
    int partitions = 2;
    int replication = 3;
    Properties topicConfig = new Properties(); // add per-topic configurations settings here

    // Note: You must initialize the ZkClient with ZKStringSerializer.  If you don't, then
    // createTopic() will only seem to work (it will return without error).  The topic will exist in
    // only ZooKeeper and will be returned when listing topics, but Kafka itself does not create the
    // topic.
    ZkClient zkClient = new ZkClient(
        zookeeperConnect,
        sessionTimeoutMs,
        connectionTimeoutMs,
        ZKStringSerializer$.MODULE$);

    // Security for Kafka was added in Kafka 0.9.0.0
    boolean isSecureKafkaCluster = false;

    ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeperConnect), isSecureKafkaCluster);
    AdminUtils.createTopic(zkUtils, topic, partitions, replication, topicConfig, RackAwareMode.Enforced$.MODULE$);
    zkClient.close();
  }

}

Answer 2

Начиная с версии 0.11.0.0 все, что вам нужно:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version>
</dependency>

Этот артефакт теперь содержит AdminClient (org.apache.kafka.clients.admin).

AdminClient может справиться со многими задачами администратора Kafka, включая создание темы:

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");

AdminClient admin = AdminClient.create(config);

Map<String, String> configs = new HashMap<>();
int partitions = 1;
int replication = 1;

admin.createTopics(asList(new NewTopic("topic", partitions, replication).configs(configs)));

Результатом этой команды является CreateTopicsResult, который вы можете использовать, чтобы получить Future для всей операции или для каждой отдельной темы создания:

• чтобы получить будущее для всей операции, используйте CreateTopicsResult#all(),
• получить Futures для всех тем индивидуально, используйте CreateTopicsResult#values(),

Например:

CreateTopicsResult result = ...
KafkaFuture<Void> all = result.all();

или же:

CreateTopicsResult result = ...
for (Map.Entry<String, KafkaFuture<Void>> entry : result.values().entrySet()) {
    try {
        entry.getValue().get();
        log.info("topic {} created", entry.getKey());
    } catch (InterruptedException | ExecutionException e) {
        if (Throwables.getRootCause(e) instanceof TopicExistsException) {
            log.info("topic {} existed", entry.getKey());
        }
    }
}

KafkaFuture является "гибким будущим, которое поддерживает цепочку вызовов и другие шаблоны асинхронного программирования", и "в конечном итоге станет тонкой прокладкой над Java 8 CompletebleFuture".

Answer 3

Для создания темы через java api и Kafka 0.8+ попробуйте следующее:

Первый импорт под заявлением

import kafka.utils.ZKStringSerializer$;

Создайте объект для ZkClient следующим образом,

ZkClient zkClient = new ZkClient("localhost:2181", 10000, 10000, ZKStringSerializer$.MODULE$);
AdminUtils.createTopic(zkClient, myTopic, 10, 1, new Properties());

Answer 4

Вы можете попробовать с помощью класса scala kafka.admin.CreateTopicCommand создать тему из кода Java..., предоставив необходимые аргументы.

String [] arguments = new String[8];
arguments[0] = "--zookeeper";
arguments[1] = "10.***.***.***:2181";
arguments[2] = "--replica";
arguments[3] = "1";
arguments[4] = "--partition";
arguments[5] = "1";
arguments[6] = "--topic";
arguments[7] = "test-topic-Biks";

CreateTopicCommand.main(arguments);

NB: Вы должны добавить зависимости maven для jopt-simple-4.5 & zkclient-0.1

Answer 5

Основываясь на последних API-версиях kafka-client и Kafka 1.1.1, рабочая версия кода выглядит следующим образом:

Импортируйте последние kafka-клиенты, используя sbt.

// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
libraryDependencies += Seq("org.apache.kafka" % "kafka-clients" % "2.1.1",
"org.apache.kafka" %% "kafka" % "1.0.0")

Код для создания темы в Scala:

import java.util.Arrays
import java.util.Properties

import org.apache.kafka.clients.admin.NewTopic
import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig}

class CreateKafkaTopic {
  def create(): Unit = {
    val config = new Properties()
    config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "192.30.1.5:9092")

    val localKafkaAdmin = AdminClient.create(config)

    val partitions = 3
    val replication = 1.toShort
    val topic = new NewTopic("integration-02", partitions, replication)
    val topics = Arrays.asList(topic)

    val topicStatus = localKafkaAdmin.createTopics(topics).values()
    //topicStatus.values()
    println(topicStatus.keySet())
  }

}

Подтвердите новую тему, используя:

./kafka-topics.sh --zookeeper 192.30.1.5:2181 --list

Надеюсь, это кому-нибудь пригодится. Ссылка: http://kafka.apache.org/21/javadoc/index.html?org/apache/kafka/clients/admin/AdminClient.html

Answer 6

Если вы используете Kafka 0.10.0.0+, создание темы из Java требует передачи параметра типа RackAwareMode. Это кейс-объект Scala, и получить его экземпляр из Java сложно (доказательство: например, как "получить" кейс-объект Scala из Java? Например. Но это не применимо для нашего случая).

К счастью, rackAwareMode является необязательным параметром. Тем не менее, Java не поддерживает необязательные параметры. Как мы решаем это? Вот решение:

AdminUtils.createTopic(zkUtils, topic, 1, 1, 
    AdminUtils.createTopic$default$5(),
    AdminUtils.createTopic$default$6());

Используйте это с ответом Мигуно, и вы готовы идти.

Answer 7

Существует новый API AdminZkClient, который мы можем использовать для управления темами на сервере Kafka.

String zookeeperHost = "127.0.0.1:2181";
Boolean isSucre = false;
int sessionTimeoutMs = 200000;
int connectionTimeoutMs = 15000;
int maxInFlightRequests = 10;
Time time = Time.SYSTEM;
String metricGroup = "myGroup";
String metricType = "myType";
KafkaZkClient zkClient = KafkaZkClient.apply(zookeeperHost,isSucre,sessionTimeoutMs,
                connectionTimeoutMs,maxInFlightRequests,time,metricGroup,metricType);

AdminZkClient adminZkClient = new AdminZkClient(zkClient);

String topicName1 = "myTopic";
int partitions = 3;
int replication = 1;
Properties topicConfig = new Properties();

adminZkClient.createTopic(topicName1,partitions,replication,
            topicConfig,RackAwareMode.Disabled$.MODULE$);

Вы можете ссылаться на эту ссылку для получения подробной информации https://www.analyticshut.com/streaming-services/kafka/create-and-list-kafka-topics-in-java/

Answer 8

Несколько способов, которыми ваш звонок не будет работать.

1. Если в вашем кластере Kafka недостаточно узлов для поддержки значения репликации 3.

2. Если есть префикс пути chroot, вы должны добавить его после порта zookeeper

3. Вы не находитесь в каталоге установки Kafka при запуске (это наиболее вероятно)

Answer 9

Из примера производителя Kafka 0.8 в приведенном ниже примере будет создана тема с именем page_visits а также начать производство, если auto.create.topics.enable атрибут установлен в true (по умолчанию) в конфигурационном файле Kafka Broker

import java.util.*;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer {
    public static void main(String[] args) {
        long events = Long.parseLong(args[0]);
        Random rnd = new Random();

        Properties props = new Properties();
        props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("partitioner.class", "example.producer.SimplePartitioner");
        props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, String> producer = new Producer<String, String>(config);

        for (long nEvents = 0; nEvents < events; nEvents++) { 
            long runtime = new Date().getTime();  
            String ip = “192.168.2.” + rnd.nextInt(255); 
            String msg = runtime + “,www.example.com,” + ip; 
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);
            producer.send(data);
        }
        producer.close();
   }
}

Answer 10

Создание темы в Scala. Если вы работаете в режиме seudo в конфигурационном файле брокера

auto.create.topics.enable= true

Включит автоматическое создание темы на сервере. Если для этого параметра задано значение true, то попытки создать, использовать или извлечь метаданные для несуществующего раздела автоматически создадут его с коэффициентом репликации по умолчанию и числом разделов.

приведенный ниже фрагмент кода.

import scala.util.Random
import java.util.Properties
import kafka.producer.ProducerConfig
import kafka.producer.Producer
import kafka.producer.KeyedMessage
import java.util.Date

class SimpleProducer {


  def sendmessages(){
  val rnd = new Random();

  val props = new Properties();  
  props.put("metadata.broker.list", "192.1.1.1:6667"); 
  props.put("serializer.class", "kafka.serializer.StringEncoder");
  //props.put("partitioner.class", "rtbi.dis.producers.SimplePartitioner")

  val config = new ProducerConfig(props);
  val producer = new Producer[String, String](config);
    for (event<-1 to 5000) { 
               val runtime = new Date().getTime;  
               val ip = "192.1.1.1" + rnd.nextInt(255); 
               val msg = runtime + ",www.example.com," + ip; 
               val data = new KeyedMessage[String, String]("mytopic", ip, msg); //here mytopic is a topic 
               producer.send(data);
        }
        producer.close();
  }
 }
  object SimpleProducer extends App{   
    val s= new SimpleProducer().sendmessages();
  }

Answer 11

Из какой IDE вы пытаетесь?

Пожалуйста, укажите полный путь, ниже приведена команда из терминала, которая создаст тему

1. cd kafka/bin
2. ./kafka-create-topic.sh --topic test --zookeeper localhost:2181

Answer 12

Начиная с Kafka 0.10.1 ZKStringSerializer, упомянутый Майклом, является частным (для Scala). Вы можете использовать фабричные методы createZkClient или createZkClientAndConnection в ZkUtils.

Пример Scala для Кафки 0.10.1:

import kafka.utils.ZkUtils

val sessionTimeoutMs = 10000
val connectionTimeoutMs = 10000
val (zkClient, zkConnection) = ZkUtils.createZkClientAndConnection(
  "localhost:2181", sessionTimeoutMs, connectionTimeoutMs) 

Тогда просто создайте тему, как предложил Майкл:

import kafka.admin.AdminUtils

val zkUtils = new ZkUtils(zkClient, zkConnection, false)
val numPartitions = 4
val replicationFactor = 1
val topicConfig = new Properties
val topic = "my-topic"
AdminUtils.createTopic(zkUtils, topic, numPartitions, replicationFactor, topicConfig)