KafkaStoreUtils { public static void checkAndCreateTopic(String zkConnect, String topic, int replicas) { ZkClient zkClient = new ZkClient(zkConnect, SESSION_TIMEOUT_MS, CONNECTION_TIMEOUT_MS, ZKStringSerializer$.MODULE$); ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zkConnect), IS_SECURE_KAFKA_CLUSTER); if (AdminUtils.topicExists(zkUtils, topic)) { verifyTopic(zkUtils, topic); return; } int partitions = 1; Properties topicConfig = new Properties(); topicConfig.put(LogConfig.CleanupPolicyProp(), "compact"); AdminUtils.createTopic(zkUtils, topic, partitions, replicas, topicConfig, RackAwareMode.Enforced$.MODULE$); zkClient.close(); zkUtils.close(); } @SuppressWarnings({ "rawtypes", "unchecked" }) private static void verifyTopic(ZkUtils zkUtils, String topic) { Set topics = new HashSet(); topics.add(topic); scala.collection.mutable.Map partitionAssignmentForTopics = zkUtils .getPartitionAssignmentForTopics(JavaConversions.asScalaSet(topics).toSeq()); scala.collection.Map partitionAssignment = (scala.collection.Map) partitionAssignmentForTopics.get(topic).get(); if (partitionAssignment.size() != 1) { throw new RuntimeException(String.format("The schema topic %s should have only 1 partition.", topic)); } Properties prop = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), topic); String retentionPolicy = prop.getProperty(LogConfig.CleanupPolicyProp()); if (retentionPolicy == null || "compact".compareTo(retentionPolicy) != 0) { throw new RuntimeException(String.format("The retention policy of the schema topic %s must be compact.", topic)); } } } private  KafkaStoreUtils(); static void checkAndCreateTopic(String zkConnect, String topic, int replicas);  }
@Test public void topicIsCreated() { String topic = "topic1"; KafkaStoreUtils.checkAndCreateTopic(cluster.zKConnectString(), topic, REPLICAS); assertThat(topicExists(topic), is(true)); }
@Test public void topicAlreadyExists() throws Exception { String topic = "topic2"; int partitions = 1; Properties props = new Properties(); props.setProperty(LogConfig.CleanupPolicyProp(), "compact"); cluster.createTopic(topic, partitions, 1, props); try { KafkaStoreUtils.checkAndCreateTopic(cluster.zKConnectString(), topic, REPLICAS); } catch (RuntimeException e) { fail(); } }
@Test(expected = RuntimeException.class) public void topicAlreadyExistsButWrongPartitions() throws Exception { String topic = "topic3"; int partitions = 2; Properties props = new Properties(); props.setProperty(LogConfig.CleanupPolicyProp(), "compact"); cluster.createTopic(topic, partitions, 1, props); KafkaStoreUtils.checkAndCreateTopic(cluster.zKConnectString(), topic, REPLICAS); }
@Test(expected = RuntimeException.class) public void topicAlreadyExistsButNotCompact() throws Exception { String topic = "topic4"; int partitions = 1; cluster.createTopic(topic, partitions, 1); KafkaStoreUtils.checkAndCreateTopic(cluster.zKConnectString(), topic, REPLICAS); }