Hi All, I am facing some strange issues, probably because I am making something wrong.
Basically, I need to create kafka topics dynamically, post some messages there and after some time, remove the topic. 1) For creating the topic private static final int KAFKA_CONNECT_TIMEOUT = 10000; private static final int KAFKA_DEFAULT_TOPIC_PARTITIONS = 10; private static final int KAFKA_DEFAULT_TOPIC_REPLICATION = 1; public static void createTopic(String brokerConnect, String topic) { ZkUtils utils = createZkUtils(brokerConnect); AdminUtils.createTopic(utils, topic, KAFKA_DEFAULT_TOPIC_PARTITIONS, KAFKA_DEFAULT_TOPIC_REPLICATION, new Properties()); } private static ZkUtils createZkUtils(String brokerConnect) { ZkClient zkClient = new ZkClient(brokerConnect, KAFKA_CONNECT_TIMEOUT, KAFKA_CONNECT_TIMEOUT, ZKStringSerializer$.MODULE$); ZkUtils utils = new ZkUtils(zkClient, new ZkConnection(brokerConnect), false); return utils; } 2) For creating the consumer: private static ConsumerConfig createConsumerConfig(String zookeeper, String groupId, long consumerTimeout) { Properties props = new Properties(); props.put("zookeeper.connect", zookeeper); props.put("group.id", groupId); props.put("zookeeper.session.timeout.ms", "6000"); props.put("zookeeper.sync.time.ms", "200"); props.put("auto.commit.interval.ms", "1000"); return new ConsumerConfig(props); } 3) For removing the topic public static void deleteTopic(String brokerConnect, String topic) { ZkUtils utils = createZkUtils(brokerConnect); AdminUtils.deleteTopic(utils, topic); } What I see is that after some not much time for creating/deleting topics, I start seeing timeouts creating the consumers, or creating the ZK utils, like this: Caused by: org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to zookeeper server within timeout: 10000 at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1223) ~[zkclient-0.7.jar:0.7] at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:155) ~[zkclient-0.7.jar:0.7] at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:129) ~[zkclient-0.7.jar:0.7] at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:96) ~[zkclient-0.7.jar:0.7] And I see it many times. Is this expected? Can I do something better to improve this? Appreciate any feedback, Demian