Page History

/**
 * A policy that is enforced on actions affecting topics.
 * An implementation of this policy can be configured on a broker via the
 * {@code topic.actions.policy.class.name} broker config. 
 * When this is configured the named class will be instantiated reflectively 
 * using its nullary constructor and will then pass the broker configs to 
 * its <code>configure()</code> method. During broker shutdown, the 
 * <code>close()</code> method will be invoked so that resources can be 
 * released (if necessary).
 */
interface TopicActionsPolicy extends Configurable, AutoCloseable {
    /** Enumerates possible actions on topics. */
    static enum Action {
        /** The creation of a topic. */
        CREATE,

        /** The modification of a topic. */
        MODIFY,

        /** The deletion of a topic. */
        DELETE
    }

    /**
     * Represents the state of a topic either before, or as a result of, an administrative request affecting the topic.
     */
    static interface TopicState {
        /**
         * The number of partitions of the topic.
         */
        public int numPartitions();

        /**
         * The replication factor of the topic.
         */
        public Short replicationFactor();

        /**
         * The replica assignments of the topic.
         */
        public Map<Integer, List<Integer>> replicasAssignments()

        /**
         * The topic config.
         */
        public Map<String,String> configs();
    }

    /**
     * Parameters for a request to perform an {@linkplain #action} on a {@linkplain #topic}
     * @see #validate(RequestMetadata)
     */
    static interface RequestMetadata {

        /**
         * The {@linkplain Action action} being performed on the topic.
         */
        public Action action();

        /**
         * The topic the {@linkplain #action() action} is being performed upon.
         */
        public String topic();

        /**
         * The authenticated principal making the request, or null if the session is not authenticated.
         */
        public PrincipalKafkaPrincipal principal();

         /**
          * The state the topic has before the request.
          * <ul>
          * <li>For {@link Action#CREATE} this will be null.</li>
          * <li>For {@link Action#MODIFY} this will be the state the topic currently has (before the modification).</li>
          * <li>For {@link Action#DELETE} this will be the state of the topic which is going to be deleted.</li>
          * </ul>
          */
        public TopicState preRequestState();

        /**
         * The state the topic will have after the request.
         * <ul>
         * <li>For {@link Action#CREATE} this will be the requested state of the topic to be created.</li>
         * <li>For {@link Action#MODIFY} this will be the state the topic will have after the modification.</li>
         * <li>For {@link Action#DELETE} this will be null.</li>
         * </ul>
         */
        public TopicState postRequestState();

    }

    /**
     * Validate the request parameters and throw a <code>PolicyViolationException</code> with a suitable error
     * message if the request parameters for the provided topic do not satisfy this policy.
     *
     * Clients will receive the POLICY_VIOLATION error code along with the exception's message. Note that validation
     * failure only affects the relevant topic, other topics in the request will still be processed.
     *
     * @param requestMetadata the request parameters for the provided topic.
     * @throws PolicyViolationException if the request parameters do not satisfy this policy.
     */
    void validate(RequestMetadata requestMetadata) throws PolicyViolationException;
}