Page History

...

package org.apache.kafka.server;

/**
 * An interface for enforcing a policy on the records that are accepted by this
 * broker.
 * 
 * A common use case is for validating that records contain the correct schema
 * information.
 * 
 * If <code>record.validation.policy.class.name</code> is defined, Kafka will
 * create an instance of the specified class using the default 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).
 */
public interface RecordValidationPolicy extends Configurable, Closeable {

 /** 
 * TopicMetadata describes the topic-partition a record is being produced to.
 */
 public interface TopicMetadata {
 TopicPartition getTopicPartitiontopicPartition();

 /**
 * @return the value of the topic config <code>"validation.policy"</code>
 */
 String validationPolicy();
 }

 /**
 * RecordProxy contains the parts of a record that can be inspected by a validation policy.
 * 
 * For efficiency, only the data required by the policy {@linkplain}RecordIntrospectionHints
 * are guaranteed to be available to the policy.
 */
 public interface RecordProxy {
 Header[] getHeadersheaders();
 ByteBuffer getKeykey();
 ByteBuffer getValuevalue();
 }

 /**
 * @throws InvalidRecordException when this policy rejects a record 
 */
 void validate(TopicMetadata topic, RecordProxy record) throws InvalidRecordException;

 /**
 * The parts of the record that a policy needs to inspect. This is used for
 * iteration when using compression, so as to only decompress the minimum data
 * necessary to perform the validation
 */
 public interface RecordIntrospectionHints {
 /**
 * @return whether the policy will need to access a record's headers
 */
 boolean needHeadersaccessHeaders();

 /**
 * @return minimum number of bytes from the beginning of the record's key byte[]
 * @return 0 key is not needed for policy
 * @return -1 or Long.MAX_LONG for all
 */
 long needKeyBytesaccessKeyBytes();

 /**
 * @return minimum number of bytes from the beginning of the record's value byte[]
 * @return 0 key is not needed for policy
 * @return -1 or Long.MAX_LONG for all
 */
 long needValueBytesaccessValueBytes();
 }

 /**
 * @return hints describing the parts of the record that this policy
 * needs to inspect.
 */
 RecordIntrospectionHints getHints();
}

...