...

package org.apache.kafka.server.policy;


public interface CreateTopicPolicy {
  void validate(TopicDetails topicDetails) throws InvalidRequestException;
}

public class TopicDetails {
    private final String topic;
    private final int numPartitions;
    private final short replicationFactor;
    private final Map<Integer, List<Integer>> replicasAssignments;
    private final Map<String, String> configs;

    /** replicasAssignment is a map from partition id to broker ids */
    public TopicDetails(String topic, int numPartitions, short replicationFactor, Map<Integer, List<Integer>> replicasAssignments, Map<String, String> configs) {
        this.topic = topic;
        this.numPartitions = numPartitions;
        this.replicationFactor = replicationFactor;
        this.replicasAssignments = replicasAssignments;
        this.configs = configs;
	}
}

Users will have to ensure that their the policy implementation code is in the broker's classpath.

...

AdminManager will instantiate a CreateTopicsPolicy CreateTopicPolicy instance if create.topics.policy.class.name is defined and will call the validate method after existing checks are done (i.e. numPartitions and replicationFactor cannot be used with replicaAssignments). If a request with does not pass validation, an InvalidRequestException is thrown (in the same way as existing checks). One . 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.

An existing limitation is that there is no mechanism to pass the error message back to the client. In order to do this, we would have to modify the response to include an error string, which means that the user may not know the exact reason why validation failed for a given topic. Protocol changes would be required to improve this.

As described in the previous section, we are proposing one policy config/interface per supported request type. The main advantage is that we can add additional configs in a compatible manner. Implementors can use , but it also allows for modular policy implementations. Implementors also have the option of using a single implementation class if they wish, but they multiple configs would still have to be set all the relevant configs. Also, there would be one interface per supported request type. Once we move to Java 8, we can provide an interface that inherits from the various interfaces with an empty implementation via default methods.

Compatibility, Deprecation, and Migration Plan

There is no compatibility impact as there is no change in behaviour unless the new config is used.

Future Work

1. Return error messages to the client: in order to do this, we would have to modify the CreateTopicsResponse to include an optional error string for each partition. Once this is done, we can pass the message from the exception thrown by validate to the client.
2. As we add new policy interfaces, it could make sense to introduce an admin.policy.class.name config so that one could provide a single implementation for multiple policies without having to set multiple configs. Not clear if the cost is worth the benefit, however.
3. Classloader isolation: it may make sense to load user supplied implementations in a separate classloader to avoid dependency version clashes. This is currently being explored in the context of Connect and it may make sense to extend it to all user supplied code in the broker (including Authorizers).

Rejected Alternatives

1. A single config for an implementation of an interface with multiple validate methods: there would be no way to add methods without breaking binary compatibility until we move to Java 8.
2. A single config for an implementation of an abstract class with multiple validate methods, each with an empty implementation by default: this makes it easier to evolve when compared to rejected alternative 1, but provides less flexibility/modularity to implementors.
3. Restrict it via configs instead of pluggable interfaces: it would be clunky to provide configs for every requirement that users may have.
4. Extending Authorizer to perform validation: even though this could work, it's not particularly intuitive.

...