How do you programmatically create a topic with Kafka 1.1.0

Question

I recently upgraded to Kafka 1.1.0. I am trying to create unit tests for the kafka consumer. For this purpose it would be ideal if the unit test can create the topic it uses for the test. I found some code that looks like it should do what I want. However, when I run it it throws an exception: java.lang.NoSuchMethodError: org.apache.kafka.common.utils.Utils.closeQuietly(Ljava/lang/AutoCloseable;Ljava/lang/String;)V

Here is the code to create a topic which I found on line:

@BeforeClass
public static void createTopic() {
   try (final AdminClient adminClient = AdminClient.create(configure())) {
        try {
            // Define topic
            NewTopic newTopic = new NewTopic("test-orders", 1, (short)1);

            // Create topic, which is async call.
            final CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singleton(newTopic));

            // Since the call is Async, Lets wait for it to complete.
            createTopicsResult.values().get(ordersTopic).get();
        } catch (InterruptedException | ExecutionException e) {
            if (!(e.getCause() instanceof TopicExistsException)) {
                throw new RuntimeException(e.getMessage(), e);
            }
        }
    }
}

However it throws an exception when I run it.

java.lang.NoSuchMethodError: org.apache.kafka.common.utils.Utils.closeQuietly(Ljava/lang/AutoCloseable;Ljava/lang/String;)V
at org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:334)
at org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:52)
at com.sial.notifications.topics.OrdersTopicsTests.createTopic(OrdersTopicsTests.java:162)

The only configuration parameters I pass to it is the bootstrap servers and a client.id. What am I doing wrong? it seems simple enough

You'll get that error when your classpath isn't setup correctly — OneCricketeer, Jul 11 '18 at 03:45
have you checked this https://stackoverflow.com/questions/16946778/how-can-we-create-a-topic-in-kafka-from-the-ide-using-api — Ryuzaki L, Jul 11 '18 at 04:09

score 4 · Accepted Answer · answered Jul 10 '18 at 22:38

This slightly modified code worked for me when I ran it stand-alone against a 1.1.0 broker:

public static void main(String[] args) {
    final String ordersTopic = "test-orders";
    Properties props = new Properties();
    props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

    try (final AdminClient adminClient = AdminClient.create(props)) {
        try {
            // Define topic
            NewTopic newTopic = new NewTopic(ordersTopic, 1, (short)1);

            // Create topic, which is async call.
            final CreateTopicsResult createTopicsResult = adminClient.createTopics(Collections.singleton(newTopic));

            // Since the call is Async, Lets wait for it to complete.
            createTopicsResult.values().get(ordersTopic).get();
        } catch (InterruptedException | ExecutionException e) {
            if (!(e.getCause() instanceof TopicExistsException))
                throw new RuntimeException(e.getMessage(), e);
        }
    }
}

Since this is pretty similar to your code, and based on the error you're seeing, perhaps you haven't completely sorted out the dependencies to Kafka libraries? I used the Maven artifact org.apache.kafka:kafka_2.12:1.1.0.

thank you. The problem was that I use Eclipse, and it was holding onto the older kafka version's jars in its build path. If I had run the unit tests from the command line this would have worked. — whomer, Jul 11 '18 at 15:17
Following up with this comment, I too had recently upgraded from a Kafka-client that was in the 0.10 version to 1.0... version and was running into this error, the older version was still hanging around, once I removed it, it worked just fine. — NateH06, Aug 20 '18 at 21:33

score 0 · Answer 2 · answered Jul 13 '18 at 10:35

0

A much easier way is to just configure Kafka to automatically create any topic you use that does not exist already with the

auto.create.topics.enable

setting for Kafka. Doing this, there is no extra code needed for creating topics. You just use whatever topic name you want, and Kafka will create it for you when you use it.

answered Jul 13 '18 at 10:35

Marius Waldal

9,537
4
30
44

It's not possible due to restrictions on some environments. You also can't define important config settings like partitions and replications. – itstata Dec 02 '21 at 07:17

How do you programmatically create a topic with Kafka 1.1.0

2 Answers2