apache-kafka - 通过 Java API 获取和更新 Kafka 主题配置

Question

我正在编写一个应用程序，通过 Java API 对 kafka 主题执行一系列操作。我能够创建主题并添加分区。我需要有关获取主题元数据(例如分区、代理)和配置以及更新配置的帮助。

作为引用，我想更新此处提供的主题级别配置 - https://kafka.apache.org/documentation#configuration比如cleanup.policy, compression.type等

Answer 1

您可以使用下面的代码打印主题级别的配置。更新配置的用法类似。

String[] args = {"--zookeeper", "localhost:2181", "--entity-type", "topics", "--entity-name", "test", "--describe"};

ConfigCommand.main(args);

关于获取元数据，请引用https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example中的为主题和分区寻找Lead Broker :

添加:使用 AdminUtils 添加配置获取和更新示例:

ZkUtils zkUtils = ZkUtils.apply("localhost:2181/k1", 6000, 10000, JaasUtils.isZkSecurityEnabled());

    Properties pp = new Properties();
    pp.setProperty("delete.retention.ms", "3000000");
    pp.setProperty("file.delete.delay.ms", "40000");
    AdminUtils.changeTopicConfig(zkUtils, "test", pp);
    Properties p = AdminUtils.fetchEntityConfig(zkUtils, ConfigType.Topic(), "test");
    System.out.println(p);