我想为以前被覆盖的主题删除配置(将其重置为默认值)。这可以通过提供的脚本实现
$> ./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来实现这一点?
我刚找到方法 org.apache.kafka.clients.admin.KafkaAdminClient.alterConfigs(Map<ConfigResource, Config>, AlterConfigsOptions)
,但当我使用设置为null的配置值调用它时,服务器上会出现一个nullpointerexception:
[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。
1条答案
按热度按时间dvtswwa31#
kafka管理jar中提供的许多函数都是api,而不是直接编写脚本的函数。您可以在java程序中使用zookeeper adminutils更改主题的配置,如下所示。将空属性对象发送到函数以清除现有属性。
如果您经常需要这个函数,您可以包含一个文件读取器来获取新的配置,并将其打包到jar中以便于执行。