...

• Provide the Create Topic Policy implementation convenient access to querying some cluster with a convenient API for querying the cluster topics' metadata
• Introduce a Delete Topic Policy 
• A newer DeleteTopicRequest protocol message version that can include a boolean for verification only
  
• A newer DeleteTopicResponse message version that can include an error message to be returned to the user

NOTE:

Expecting community feedback on this KIP, the The initial interface proposed for querying the cluster metadata has intentionally been kept with somewhat narrow capabilities (what MessageHub actually required)
rather than being provided with speculative generality. We await major community feedback on this point. 

The Delete policy we had to implement had no requirements on querying cluster metadata, so but it should be easy to make it similar to the Create interface 

This KIP was created based on work done with Mickael Maison

...

DeleteTopics Request (Version: 1) => [topics] timeout validate_only 
  topics => STRING
  timeout => INT32
  validate_only => BOOLEAN
 
DeleteTopics Response (Version: 1) => [topic_error_codes] 
  topic_error_codes => topic error_code error_message
    topic => STRING
    error_code => INT16
    error_message => NULLABLE_STRING
 

Proposed Changes

kafka.server.AdminManager will be enhanced

• to instantiate a TopicDeletePolicy - in a similar fashion as what it currently does for the topic creation
• to invoke the valdate method of TopicDeletePolicy when handing a request to delete topics
• to skip performing the actual deletion (via AdminUtils.deleteTopic ) if the request is for validate only

• check if a topic creation policy implements CreateTopicPolicyV2 and in that case set a CreateTopicPolicyV2.TopicPartitionMetadataProvider on the policy

The TopicPartitionMetadataProvider interface could be implemented by the kafka.server.AdminManager itself or by another class, depending on how rich the Provider interface will turn out to be.

For the initial proposal, both provider methods 

  Map<String, Integer> topicsPartitionCount(); 
  boolean isTopicMarkedForDeletion(String topicName);

can be easily implemented by AdminManager

The metadata itself can be retrieved by enhancing the MetadataCache held by the AdminManager. 

Compatibility, Deprecation, and Migration Plan

...