I want to delete a configuration (reset it to default) for a topic which was overridden before. This is possible with the provided script
$> ./kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics \
--entity-name test --delete-config my.overridden.config
Is there a way to do this with the KafkaAdminClient provided in kafka-clients-1.1.1.jar?
I just found the method org.apache.kafka.clients.admin.KafkaAdminClient.alterConfigs(Map<ConfigResource, Config>, AlterConfigsOptions)
, but when I call it with a configuration value set to null, I get a NullPointerException on the server:
[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)
An empty list won't work either.
I am using Kafka in version 2.11-1.1.0.
A lot of the function provided in Kafka admin jars are APIs and not a direct scripted function. You can change a topic's configuration using zookeeper AdminUtils in a java program as below. Send an empty property object to the function to clear out existing properties.
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);
}
If you need this function often, you can include a file reader to get new configurations and package into a jar for easy execution.