import kafka.admin.AdminOperationException; //导入依赖的package包/类
/**
 * Create a topic
 *
 * @param connection        Connection
 * @param topicName         Topic name
 * @param partitions        The number of partitions for the topic being created
 * @param replicationFactor The replication factor for each partition in the topic being created
 * @param topicProperties   A topic configuration override for an existing topic
 * @throws TopicOperationException if topic was not created.
 */
public void createTopic(final ZkUtils connection, final String topicName,
                        final int partitions,
                        final int replicationFactor,
                        final Properties topicProperties) {

    try {
        AdminUtils.createTopic(connection,
                topicName,
                partitions,
                replicationFactor,
                topicProperties);

    } catch (IllegalArgumentException | KafkaException | AdminOperationException e) {
        throw new TopicOperationException(topicName, e.getMessage(), e, this.getClass());
    }
}
 

import kafka.admin.AdminOperationException; //导入依赖的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.admin.AdminOperationException; //导入依赖的package包/类
public static void updatePartitionReassignmentData(ZkClient zkClient, Multimap<TopicAndPartition, Integer> partitionsToBeReassigned) {
    String zkPath = ZkUtils.ReassignPartitionsPath;
    int size = partitionsToBeReassigned.size();
    switch (size) {
        case 0: // need to delete the /admin/reassign_partitions path
            deletePath(zkClient, zkPath);
            logger.info("No more partitions need to be reassigned. Deleting zk path {}", zkPath);
            break;
        default:
            String jsonData = getPartitionReassignmentZkData(partitionsToBeReassigned);
            try {
                updatePersistentPath(zkClient, zkPath, jsonData);
                logger.info("Updated partition reassignment path with {}", jsonData);
            } catch (ZkNoNodeException nne) {
                ZkUtils.createPersistentPath(zkClient, zkPath, jsonData);
                logger.debug("Created path {} with {} for partition reassignment", zkPath, jsonData);
            } catch (Throwable e) {
                throw new AdminOperationException(e.toString());
            }
    }
}
 

import kafka.admin.AdminOperationException; //导入依赖的package包/类
/**
 * Override topic configuration
 *
 * @param connection zookeeper util API
 * @param topicName  topic name
 * @param configs    topic properties
 */
public void overrideTopicProperties(final ZkUtils connection, final String topicName, final Properties configs) {
    try {
        AdminUtils.changeTopicConfig(connection, topicName, configs);
    } catch (AdminOperationException | KafkaException e) {
        throw new TopicOperationException(topicName, e.getMessage(), e, this.getClass());
    }
}
 

import kafka.admin.AdminOperationException; //导入依赖的package包/类
@Override
public KafkaTopic createTopic(final KafkaTopic topic) throws RestException {
  if (!listTopics().contains(topic.getName())) {
    try {
      adminUtils.createTopic(zkUtils, topic.getName(), topic.getNumPartitions(), topic.getReplicationFactor(), topic.getProperties(), RackAwareMode.Disabled$.MODULE$);
    } catch (AdminOperationException e) {
      throw new RestException(e);
    }
  }
  return topic;
}
 

import kafka.admin.AdminOperationException; //导入依赖的package包/类
/**
 * Modify number of partition of a Kafka topic.
 *
 * @param topicName     name of topic.
 * @param numPartitions
 * @throws kafka.common.KafkaException
 */
public void modifyTopicPartitioning(String topicName, int numPartitions) throws KafkaException {
    if (AdminUtils.topicExists(zkUtils, topicName)) {
        logger.debug("Altering topic {}", topicName);
        try {
            AdminUtils.addPartitions(zkUtils, topicName, numPartitions, "", true, RackAwareMode.Enforced$.MODULE$);
            logger.debug("Topic {} altered with partitions : {}", topicName, partitions);
        } catch (AdminOperationException aoe) {
            logger.debug("Error while altering partitions for topic : {}", topicName, aoe);
        }
    } else {
        logger.debug("Topic {} doesn't exists", topicName);
    }
}
 

import kafka.admin.AdminOperationException; //导入依赖的package包/类
@Test
public void whenAdminUtilsThrowsAdminOperationExceptionCreateTopicShouldProperlyWrapExceptionInRestException() throws Exception {
  exception.expect(RestException.class);

  final Map<String, List<PartitionInfo>> topics = new HashMap<>();
  topics.put("1", new ArrayList<>());

  when(kafkaConsumer.listTopics()).thenReturn(topics);

  doThrow(AdminOperationException.class).when(adminUtils).createTopic(eq(zkUtils), eq("t"), eq(1), eq(2), eq(new Properties()), eq(RackAwareMode.Disabled$.MODULE$));

  kafkaService.createTopic(VALID_KAFKA_TOPIC);
}