I am trying to create a topic in kafka 0.8.2 by using :
AdminUtils.createTopic(zkClient, myTopic, 2, 1, properties);
If I run the code more than once locally for testing, this fails as the topic was already created. Is there a way to check if the topic exists before creating the topic? The TopicCommand
api doesn't seem to return anything for listTopics
or describeTopic
.
For this purpose, you can use the method AdminUtils.topicExists(ZkUtils zkClient, String topic)
, it will return true
if the topic already exists, false
otherwise.
Your code would then be something like this:
if (!AdminUtils.topicExists(zkClient, myTopic)){
AdminUtils.createTopic(zkClient, myTopic, 2, 1, properties);
}
You can use AdminClient from kakfa-client version 0.11.0.0
Sample code:
Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhist:9091");
AdminClient admin = AdminClient.create(config);
ListTopicsResult listTopics = admin.listTopics();
Set<String> names = listTopics.names().get();
boolean contains = names.contains("TEST_6");
if (!contains) {
List<NewTopic> topicList = new ArrayList<NewTopic>();
Map<String, String> configs = new HashMap<String, String>();
int partitions = 5;
Short replication = 1;
NewTopic newTopic = new NewTopic("TEST_6", partitions, replication).configs(configs);
topicList.add(newTopic);
admin.createTopics(topicList);
}