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* toReturns performwhether anthe {@linkplaintopic #action}is onmarked a {@linkplain #topic}
   for deletion
  * @see #validate(RequestMetadata)
     */
    static interface RequestMetadata {

   public boolean markedForDeletion();

    }

    /**
     * Parameters for a request to *perform Thean {@linkplain Action action} being performed on the topic.
    #action} on a {@linkplain #topic}
     * @see #validate(RequestMetadata, ClusterState)
      */
    static interface   public Action action();RequestMetadata {

        /**
         * The topic the {@linkplain #action()Action action} is being performed on the upontopic.
         */
        public StringAction topicaction();

         /**
         * The authenticated principal makingtopic the request, or null if the session is not authenticated{@linkplain #action() action} is being performed upon.
         */
        public KafkaPrincipalString principaltopic();

          /**
          * The authenticated principal statemaking the request, topicor hasnull beforeif the request session is not authenticated.
          * <ul>/
        public  * <li>For {@link Action#CREATE} this will be null.</li>
 KafkaPrincipal principal();

        /**
         * <li>ForThe {@linkstate Action#MODIFY}the thistopic will have beafter the request.
   state the topic currently has (before the modification).</li>* <ul>
          * <li>For {@link Action#DELETEAction#CREATE} this will be the requested state of the topic whichto is going to be deletedbe created.</li>
         * * </ul>
          */
   <li>For {@link Action#MODIFY} this will be the state the topic will have after the modification.</li>
     public TopicState preRequestState();

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

    }

    /** *The <li>Forcurrent {@linkstate Action#CREATE}of thisthe willtopics bein the requested state ofcluster, before the topicrequest totakes be created.</li>effect. */
    interface ClusterState {
   * <li>For {@link Action#MODIFY} this will be/**
 the state the topic will have after the modification.</li>
 * Returns the current state of the given topic, or null if the topic does not exist.
          * <li>For {@link Action#DELETE} this will be null.</li>
         * </ul>/
        public TopicState topicState(String topicName);

        /**
         * Returns a Map with all topics and their corresponding number of partitions.
         */
        public Map<String, TopicStateInteger> postRequestStatetopicsPartitionCount();

    }

     /**
     * 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.
     * @param clusterState the current state of the cluster
     * @throws PolicyViolationException if the request parameters do not satisfy this policy.
     */
    void validate(RequestMetadata requestMetadata, ClusterState clusterState) throws PolicyViolationException;
}