2

我想删除之前被覆盖的主题的配置（将其重置为默认值）。使用提供的脚本可以做到这一点 

$> ./kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics \
--entity-name test --delete-config my.overridden.config

有没有办法使用kafka-clients-1.1.1.jar中提供的KafkaAdminClient来做到这一点？ 我刚刚找到了方法，但是当我使用设置为 null 的配置值调用它时，我在服务器上得到了 NullPointerException：
org.apache.kafka.clients.admin.KafkaAdminClient.alterConfigs(Map<ConfigResource, Config>, AlterConfigsOptions)

[2018-07-31 11:24:01,658] ERROR [Admin Manager on Broker 0]: Error processing alter configs request for resource Resource(type=TOPIC, name='test'}, config org.apache.kafka.common.requests.AlterConfigsRequest$Config@5d4fef59 (kafka.server.AdminManager)
java.lang.NullPointerException
    at java.util.Hashtable.put(Hashtable.java:459)
    at java.util.Properties.setProperty(Properties.java:166)
    at kafka.server.AdminManager$$anonfun$alterConfigs$1$$anonfun$apply$18.apply(AdminManager.scala:357)
    at kafka.server.AdminManager$$anonfun$alterConfigs$1$$anonfun$apply$18.apply(AdminManager.scala:356)
    at scala.collection.Iterator$class.foreach(Iterator.scala:891)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at kafka.server.AdminManager$$anonfun$alterConfigs$1.apply(AdminManager.scala:356)
    at kafka.server.AdminManager$$anonfun$alterConfigs$1.apply(AdminManager.scala:339)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
    at scala.collection.Iterator$class.foreach(Iterator.scala:891)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
    at scala.collection.AbstractTraversable.map(Traversable.scala:104)
    at kafka.server.AdminManager.alterConfigs(AdminManager.scala:339)
    at kafka.server.KafkaApis.handleAlterConfigsRequest(KafkaApis.scala:1987)
    at kafka.server.KafkaApis.handle(KafkaApis.scala:136)
    at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69)
    at java.lang.Thread.run(Thread.java:745)

空列表也不起作用。

我在 2.11-1.1.0 版本中使用 Kafka。

4

1 回答 1

1

Kafka 管理 jar 中提供的许多功能都是 API，而不是直接的脚本函数。您可以在 java 程序中使用 zookeeper AdminUtils 更改主题的配置，如下所示。向函数发送一个空属性对象以清除现有属性。

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import kafka.admin.AdminUtils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

public static void changeConfig(String topic) {
    ZkClient zkClient = new ZkClient("your_zkHost", 5000, 5000, ZKStringSerializer$.MODULE$);
    ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection("your_zkHost"), false);
    Properties prop = new Properties();
    prop.setProperty("retention.ms", "3600000");
    AdminUtils.changeTopicConfig(zkUtils, topic, prop);
}

如果你经常需要这个功能，你可以包含一个文件阅读器来获取新的配置并打包到一个 jar 中以便于执行。

于 2018-07-31T19:16:48.370 回答