How Can we create a topic in Kafka from the IDE using API

JavaApache Kafka

Java Problem Overview

How Can we create a topic in Kafka from the IDE using API because when I do this:

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

I get the error:

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

And I followed the developer setup as it is.

Java Solutions

Solution 1 - Java

In Kafka 0.8.1+ -- the latest version of Kafka as of today -- you can programmatically create a new topic via AdminCommand. The functionality of CreateTopicCommand (part of the older Kafka 0.8.0) that was mentioned in one of the previous answers to this question was moved to AdminCommand.

Scala example for Kafka 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)

Build dependencies, using sbt as example:

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"),
  ...
)

EDIT: Added Java example for Kafka 0.9.0.0 (latest version as of Jan 2016).

Maven dependencies:

<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>

Code:

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

}

EDIT 2: Added Java example for Kafka 0.10.2.0 (latest version as of April 2017).

Maven dependencies:

<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>

Code:

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

}

Solution 2 - Java

As of 0.11.0.0 all you need is:

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

This artifact now contains the AdminClient (org.apache.kafka.clients.admin).

AdminClient can handle many Kafka admin tasks, including topic creation:

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

The output of this command is a CreateTopicsResult, which you can use to get a Future for the whole operation or for each individual topic creation:

  • to get a future for the whole operation, use CreateTopicsResult#all().
  • to get Futures for all the topics individually, use CreateTopicsResult#values().

For example:

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

or:

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 is "a flexible future which supports call chaining and other asynchronous programming patterns," and "will eventually become a thin shim on top of Java 8's CompletebleFuture."

Solution 3 - Java

For creating a topic through java api and Kafka 0.8+ try the following,

First import below statement

import kafka.utils.ZKStringSerializer$;

Create object for ZkClient in the following way,

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

Solution 4 - Java

You can try with kafka.admin.CreateTopicCommand scala class to create Topic from Java code...providng the necessary arguments.

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: You should add the maven dependencies for jopt-simple-4.5 & zkclient-0.1

Solution 5 - Java

Based on the latest kafka-client api and Kafka 2.1.1, the working version of code follows:

Import the latest kafka-clients using sbt.

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

The code for topic creation in 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())
  }

}

Validate the new topic using:

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

Hope it is helpful to someone. Reference: http://kafka.apache.org/21/javadoc/index.html?org/apache/kafka/clients/admin/AdminClient.html

Solution 6 - Java

If you are using Kafka 0.10.0.0+, creating topic from Java requires passing parameter of RackAwareMode type. It's a Scala case object, and getting it's instance from Java is tricky (proof: https://stackoverflow.com/questions/2561415/how-do-i-get-a-scala-case-object-from-java for example. But it is not applicable for our case).

Luckily, rackAwareMode is an optional parameter. Yet Java does not support optional parameters. How do we solve that? Here is a solution:

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

Use it with miguno's answer, and you are good to go.

Solution 7 - Java

From Kafka 0.8 Producer Example the sample below will create a topic named page_visits and also start producing if the auto.create.topics.enable attribute is set to true (default) in the Kafka Broker config file

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

Solution 8 - Java

A few ways your call wouldn't work.

  1. If your Kafka cluster didn't have enough nodes to support a replication value of 3.

  2. If there is a chroot path prefix you have to append it after the zookeeper port

  3. You arent in the Kafka install directory when running (This is the most likely)

Solution 9 - Java

There is AdminZkClient which we can use to manage topics in Kafka server.

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$);

You can refer this link for details https://www.analyticshut.com/streaming-services/kafka/create-and-list-kafka-topics-in-java/

Solution 10 - Java

From which IDE are your trying ?

Please provide complete path , below are the command from terminal which will create a topic

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

Solution 11 - Java

As of Kafka 0.10.1 the ZKStringSerializer mentioned by Michael is private (for Scala). You can use the factory methods createZkClient or createZkClientAndConnection in ZkUtils.

Scala example for Kafka 0.10.1:

import kafka.utils.ZkUtils

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

Then just create the topic as Michael suggested:

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)
Categories
Recommended Java Solutions
Recommended Apache Kafka Solutions

Previous Solution:

A public action method '..' was not found on controller '..'

asp.net Mvcasp.net Mvc-Partialview

Next Solution:

String termination - char c=0 vs char c='\0'

CStringC StringsNull TerminatedAnsi C

Attributions

All content for this solution is sourced from the original question on Stackoverflow.

The content on this page is licensed under the Attribution-ShareAlike 4.0 International (CC BY-SA 4.0) license.

Content TypeOriginal AuthorOriginal Content on Stackoverflow
QuestionramuView Question on Stackoverflow
Solution 1 - JavamigunoView Answer on Stackoverflow
Solution 2 - JavaDmitry MinkovskyView Answer on Stackoverflow
Solution 3 - JavaJaya AnanthramView Answer on Stackoverflow
Solution 4 - JavaBiksView Answer on Stackoverflow
Solution 5 - JavaForeverLearnerView Answer on Stackoverflow
Solution 6 - JavaDmitriusanView Answer on Stackoverflow
Solution 7 - JavaHildView Answer on Stackoverflow
Solution 8 - JavaGregory PatmoreView Answer on Stackoverflow
Solution 9 - JavaMahesh MogalView Answer on Stackoverflow
Solution 10 - JavaSanketView Answer on Stackoverflow
Solution 11 - Javaits_a_paddoView Answer on Stackoverflow