Trait ConsumerInterceptor 

pub trait ConsumerInterceptor<K, V>: Send + 'static {
    // Required methods
    fn on_consume(&self, records: &mut ConsumerRecords<K, V>);
    fn on_commit(&self, offsets: &HashMap<TopicPartition, OffsetAndMetadata>);

    // Provided methods
    fn configure(&mut self, configs: &HashMap<String, String>) { ... }
    fn close(&mut self) { ... }
}

A plugin interface that allows you to intercept (and possibly mutate) records received by the consumer. A primary use-case is for third-party components to hook into the consumer applications for custom monitoring, logging, etc.

This trait will get consumer config properties via configure, including the client id assigned by KafkaConsumer if not specified in the consumer config. The interceptor implementation needs to be aware that it will be sharing consumer config namespace with other interceptors and serializers, and ensure that there are no conflicts.

Errors raised by ConsumerInterceptor methods will be caught, logged, but not propagated further. As a result, if the user configures the interceptor with the wrong key and value type parameters, the consumer will not return an error, just log the errors.

ConsumerInterceptor callbacks are called from the same task that invokes Consumer::poll.

Translation notes

The methods return no error: a panic is what Java’s thrown exception becomes. KafkaConsumer takes no interceptors yet, and an interceptor implementing ClusterResourceListener or Monitorable gets no callback.

Required Methods

fn on_consume(&self, records: &mut ConsumerRecords<K, V>)

This is called just before the records are returned by Consumer::poll.

This method is allowed to modify consumer records, in which case the new records will be returned. There is no limitation on number of records that could be returned from this method. I.e., the interceptor can filter the records or generate new records.

Any panic raised by this method will be caught by the caller, logged, but not propagated to the client.

Since the consumer may run multiple interceptors, a particular interceptor’s on_consume() callback will be called in the order the interceptors are configured. The first interceptor in the list gets the consumed records, the following interceptor will be passed the records returned by the previous interceptor, and so on. Since interceptors are allowed to modify records, interceptors may potentially get the records already modified by other interceptors. However, building a pipeline of mutable interceptors that depend on the output of the previous interceptor is discouraged, because of potential side-effects caused by interceptors potentially failing to modify the record and panicking. If one of the interceptors in the list panics from on_consume(), the panic is caught, logged, and the next interceptor is called with the records returned by the last successful interceptor in the list, or otherwise the original consumed records.

Arguments

• records - records to be consumed by the client or records returned by the previous interceptors in the list.

Translation notes

Java returns the records; here the interceptor modifies records in place, or replaces the batch with *records = new_batch;. Since &mut lets a panicking interceptor leave a half-modified batch, build the new batch first and assign it last: then a panic leaves *records unchanged, as Java’s immutable ConsumerRecords is. Do not std::mem::take the batch before work that can panic, or the next interceptor gets an empty batch.

fn on_commit(&self, offsets: &HashMap<TopicPartition, OffsetAndMetadata>)

This is called when offsets get committed.

Any panic raised by this method will be ignored by the caller.

Arguments

• offsets - A map of offsets by partition with associated metadata

Provided Methods

fn configure(&mut self, configs: &HashMap<String, String>)

Configure this class with the given key-value pairs.

Arguments

• configs - The configuration

fn close(&mut self)

This is called when interceptor is closed.

Implementors