...

Users will have to ensure that the policy implementation code is in the broker's classpath. The constructor of TopicDetails is public to make testing convenient for users. Under normal circumstances, it should only be instantiated by Kafka. We chose to create separate API classes instead of reusing request classes to make it easier to evolve the latter.

Proposed Changes

AdminManager will instantiate a CreateTopicPolicy instance if create.topics.policy.class.name is defined. When a create topics request is received, it will process each topic in sequence. For each topic, it will first perform the existing hardcoded request parameters validation (numPartitions and replicationFactor cannot be used at the same time as replicaAssignments) followed by CreateTopicPolicy.validate (if defined). If validation fails, INVALID_REQUEST error will be returned for that topic in the response. If validation succeeds, topic creation will be attempted as usual. Note that validation failure only affects the relevant topic, other topics in the request will still be processed.

...