import kafka.common.TopicExistsException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Before
public void setup() throws Exception {
  kafkaSource = new KafkaSource();
  kafkaServer = new KafkaSourceEmbeddedKafka(null);
  try {
    kafkaServer.createTopic(topic0, 1);
    usedTopics.add(topic0);
    kafkaServer.createTopic(topic1, 3);
    usedTopics.add(topic1);
  } catch (TopicExistsException e) {
    //do nothing
    e.printStackTrace();
  }
  context = prepareDefaultContext("flume-group");
  kafkaSource.setChannelProcessor(createGoodChannel());
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * 
 * @param topic
 */
@GET
@Path("/ui/topics/{topic}")
public void getTopic(@PathParam("topic") String topic) {
	try {
		LOGGER.info("Fetching details of topic: " + topic);

		uiService.getTopic(getDmaapContext(), topic);

		LOGGER.info("Returning details of topic: " + topic);
	} catch (ConfigDbException | IOException | TopicExistsException excp) {
		LOGGER.error("Error while fetching details of topic: " + topic, excp);
		DMaaPResponseBuilder.respondWithError(getDmaapContext(), HttpStatus.SC_NOT_FOUND,
				"Error while fetching details of topic: " + topic);
	}
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * @param dmaapContext
 * @param topicName
 * @throws ConfigDbException
 * @throws IOException
 * @throws TopicExistsException
 */
@Override
public void getTopic(DMaaPContext dmaapContext, String topicName)
		throws ConfigDbException, IOException, TopicExistsException {
	// TODO - We need to work on the templates and how data will be set in
	// the template
	LOGGER.info("Fetching detials of apikey: " + topicName);
	Topic topic = getMetaBroker(dmaapContext).getTopic(topicName);

	if (null == topic) {
		LOGGER.error("Topic [" + topicName + "] does not exist.");
		throw new TopicExistsException("Topic [" + topicName + "] does not exist.");
	}

	JSONObject json = new JSONObject();
	json.put("topicName", topic.getName());
	json.put("description", topic.getDescription());
	json.put("owner", topic.getOwner());

	LOGGER.info("Returning details of topic [" + topicName + "]. Sending response.");
	DMaaPResponseBuilder.respondOk(dmaapContext, json);

}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * Creates a Topic.
 *
 * @param topicName             Topic name.
 * @param partitions        Number of partitions for the topic.
 * @param replicationFactor Replication factor.
 * @param curatorFramework CuratorFramework.
 */
public static void createTopic(String topicName, int partitions, int replicationFactor, CuratorFramework curatorFramework) {
    if (partitions <= 0)
        throw new AdminOperationException("number of partitions must be larger than 0");

    if (replicationFactor <= 0)
        throw new AdminOperationException("replication factor must be larger than 0");

    if (!topicExists(topicName, curatorFramework)) {
        m_logger.info(String.format("Topic %s not found, creating...", topicName));
        ZkClient zkClient = fromCurator(curatorFramework);
        try {
            AdminUtils.createTopic(zkClient, topicName, partitions, replicationFactor, new Properties());
            m_logger.info("Topic created. name: {}, partitions: {}, replicationFactor: {}", topicName,
                    partitions, replicationFactor);
        } catch (TopicExistsException ignore) {
            m_logger.info("Topic exists. name: {}", topicName);
        } finally {
            if (zkClient != null) {
                zkClient.close();
            }
        }
    } else {
        m_logger.info(String.format("Topic %s found!", topicName));
    }
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
public void createTopic(String name, int numPartitions, boolean waitUntilMetadataIsPropagated) throws InterruptedException {
  ZkUtils zkUtils = null;
  Level oldLevel = UnitTestHelper.getJavaLoggingLevel();
  try {
    UnitTestHelper.setJavaLoggingLevel(Level.OFF);
    zkUtils = ZkUtils.apply(zookeeperConnectString, 30000, 30000, false);
    AdminUtilsWrapper.createTopic(zkUtils, name, numPartitions, 1, new Properties());
    if (waitUntilMetadataIsPropagated) {
      waitUntilMetadataIsPropagated(name, numPartitions);
    }
  }catch(TopicExistsException tee) {
  }finally {
    if(zkUtils != null){
      zkUtils.close();
    }
    UnitTestHelper.setJavaLoggingLevel(oldLevel);
  }
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
public static boolean createKafkaTopicIfNecessary(String brokerUri, int replFactor, int numPartitions, String topic)
    throws IOException {
  URI zkUri = URI.create(brokerUri);
  Preconditions.checkArgument("zk".equals(zkUri.getScheme()));
  String zkServerList = zkUri.getAuthority() + zkUri.getPath();

  ZkUtils zkUtils = ZkUtils.apply(zkServerList, ZK_SESSION_TIMEOUT_MS,
      ZK_CONNECTION_TIMEOUT_MS, JaasUtils.isZkSecurityEnabled());
  try {
    if (AdminUtils.topicExists(zkUtils, topic)) {
      return false;
    }

    try {
      AdminUtils.createTopic(zkUtils, topic, numPartitions, replFactor, new Properties());
    } catch (TopicExistsException ignored) {
      return false;
    } catch (RuntimeException e) {
      throw new IOException(e);
    }
  } finally {
    if (zkUtils != null) {
      zkUtils.close();
    }
  }
  return true;
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * Creates a new topic
 * @param topicName The topic name
 * @param partitionCount The partition count
 * @param replicaCount The replica count
 * @param topicProperties The optional topic properties
 * @throws TopicExistsException thrown if the requested topic already exists
 */
public void createTopic(final String topicName, final int partitionCount, final int replicaCount, final Properties topicProperties) throws TopicExistsException {
	if(!running.get()) throw new IllegalStateException("The KafkaTestServer is not running");
	if(topicName==null || topicName.trim().isEmpty()) throw new IllegalArgumentException("The passed topic name was null or empty");
	if(partitionCount < 1) throw new IllegalArgumentException("Invalid topic partition count: " + partitionCount);
	if(replicaCount < 1) throw new IllegalArgumentException("Invalid topic replica count: " + replicaCount);
	final ZkUtils z = getZkUtils();
	AdminUtils.createTopic(z, topicName, partitionCount, replicaCount, topicProperties==null ? new Properties() : topicProperties, new RackAwareMode.Disabled$());
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
public void create(final String... topics) {
    final ZkClient zkClient = createZkClient();
    final List<String> existingTopics = getTopics(zkClient);

    try {
        for (final String topic:topics) {
            if ( ! (existingTopics.contains(topic) || exists(zkClient, topic)) ) {

                String[] args = {
                        "--zookeeper", zkConnect,
                        "--partitions", "1",
                        "--replication-factor", "1",
                        "--create",
                        "--topic", topic
                };

                final TopicCommand.TopicCommandOptions opts = new TopicCommand.TopicCommandOptions(args);
                opts.checkArgs();

                try {
                    TopicCommand.createTopic(zkClient, opts);
                } catch (TopicExistsException e) {
                    LOGGER.warn("unable to create topic, the topic '{}' is already defined", topic, e);
                }
            }
        }
    } finally {
        zkClient.close();
    }
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * Creates a new topic
 * @param topicName The topic name
 * @param partitionCount The partition count
 * @param replicaCount The replica count
 * @param topicProperties The optional topic properties
 * @throws TopicExistsException thrown if the requested topic already exists
 */
public void createTopic(final String topicName, final int partitionCount, final int replicaCount, final Properties topicProperties) throws TopicExistsException {
	if(!connected.get()) throw new IllegalStateException("The KafkaTestServer is not running");
	if(topicName==null || topicName.trim().isEmpty()) throw new IllegalArgumentException("The passed topic name was null or empty");
	if(partitionCount < 1) throw new IllegalArgumentException("Invalid topic partition count: " + partitionCount);
	if(replicaCount < 1) throw new IllegalArgumentException("Invalid topic replica count: " + replicaCount);
	AdminUtils.createTopic(zkUtils, topicName, partitionCount, replicaCount, topicProperties==null ? new Properties() : topicProperties, new RackAwareMode.Disabled$());
}
 

import kafka.common.TopicExistsException; //导入依赖的package包/类
/**
 * Fetching detials of topic in a templated form for display
 * 
 * @param dmaapContext
 * @param topic
 * @throws ConfigDbException
 * @throws IOException
 * @throws TopicExistsException
 */
void getTopic(DMaaPContext dmaapContext, final String topic)
		throws ConfigDbException, IOException, TopicExistsException;