[KAFKA-10033] AdminClient should throw UnknownTopicOrPartitionException instead of UnknownServerException if altering configs of non-existing topic

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 2.4.0
• Fix Version/s: 2.7.0
• Component/s: admin
• Labels: pull-request-available
• Assignee: Brian Byrne
• Reporter: Gregory Koshelev
• Created: 22/May/20 13:48
• Updated: 06/Jun/20 15:36
• Resolved: 06/Jun/20 15:36

Description

Currently, altering configs of non-existing topic leads to UnknownServerException:

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownServerException: Topic "kgn_test" does not exist.
	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
	at ru.kontur.vostok.hercules.stream.manager.kafka.KafkaManager.changeTtl(KafkaManager.java:130)
	... 10 common frames omitted
Caused by: org.apache.kafka.common.errors.UnknownServerException: Topic "kgn_test" does not exist.

The output above is produced due to AdminZkClient.validateTopicConfig method:

  def validateTopicConfig(topic: String, configs: Properties): Unit = {
    Topic.validate(topic)
    if (!zkClient.topicExists(topic))
      throw new AdminOperationException("Topic \"%s\" does not exist.".format(topic))
    // remove the topic overrides
    LogConfig.validate(configs)
  }

UnknownServerException is common exception but in this case cause is pretty clear. So this can be fixed easily by using UnknownTopicOrPartitionException.

Issue Links

• links to: GitHub Pull Request #8715
• links to: GitHub Pull Request #8717