将 Kafka 升级到 2.5.0 后，不推荐使用 ZkUtils [重复]

Question

Kafka 从 2.3.0 升级到 2.5.0 的一项新变化是删除 ZkUtils（请参阅 https://issues.apache.org/jira/browse/KAFKA-8545）

删除使用的最佳做法是什么以及我应该改用哪个包

public void createTopic(String topicName, int partitions, int replicationFactor) {
    DataExportConfig conf = ApplicationContextProvider.getApplicationContext().getBean("dataExportConfig", DataExportConfig.class);
    String zKaddress = conf.getZkHost();
    boolean isSecureKafkaCluster = false;

    ZkUtils zkUtils ZkUtils.apply(zKaddress, zkSessionTimeoutMs, zkConnectionTimeOutInMs, isSecureKafkaCluster);
    try {
        if (!AdminUtils.topicExists(zkUtils, topicName)) {
            AdminUtils.createTopic(zkUtils, topicName, partitions, replicationFactor, new Properties(), RackAwareMode.Enforced$.MODULE$);
            //log
        } else {
            //log
        }

        List<String> topicList = Arrays.asList(topicName);
        // checking real partition size for topic
        Integer topicPartitionSizeInZooKeeper = JavaConversions.mapAsJavaMap(zkUtils.getPartitionAssignmentForTopics(JavaConversions.asScalaBuffer(topicList))).get(topicName).size();
        if (topicPartitionSizeInZooKeeper != partitions) {
            //log
    } catch (Exception ex) {
            //log
        throw ex;
    } finally {
        //close zookeeper client after all topics are created
        zkUtils.zkClient().close();
    }
}

Answer 1

您现在需要使用 Admin API createTopics() 方法来创建主题：

int partitions = 1;
short replicationFactor = 3;

Properties props = new Properties();
props.put("bootstrap.servers", "localshot:9092");
AdminClient admin = AdminClient.create(props);

NewTopic topic = new NewTopic("mytopic", partitions, replicationFactor);
Collection<NewTopic> newTopics = Arrays.asList(topic);
CreateTopicsResult ctr = admin.createTopics(newTopics);
ctr.all().get();