Package org.springframework.kafka.listener.adapter

Class FilteringBatchMessageListenerAdapter<K,V>

java.lang.Object

org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter<T>

org.springframework.kafka.listener.adapter.AbstractFilteringMessageListener<K,V,BatchMessageListener<K,V>>

org.springframework.kafka.listener.adapter.FilteringBatchMessageListenerAdapter<K,V>

Type Parameters:

K - the key type.

V - the value type.

All Implemented Interfaces:

BatchAcknowledgingConsumerAwareMessageListener<K,V>, BatchMessageListener<K,V>, ConsumerSeekAware, DelegatingMessageListener<BatchMessageListener<K,V>>, GenericMessageListener<List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>>>

public class FilteringBatchMessageListenerAdapter<K,V>
extends AbstractFilteringMessageListener<K,V,BatchMessageListener<K,V>>
implements BatchAcknowledgingConsumerAwareMessageListener<K,V>

A BatchMessageListener adapter that implements filter logic via a RecordFilterStrategy.

Author:

Gary Russell, Sanghyeok An

Nested Class Summary

Nested classes/interfaces inherited from interface org.springframework.kafka.listener.ConsumerSeekAware

ConsumerSeekAware.ConsumerSeekCallback

Field Summary

Fields inherited from class org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter

delegate, delegateType, logger

Constructor Summary

Constructors

Constructor	Description
FilteringBatchMessageListenerAdapter(BatchMessageListener<K,V> delegate, RecordFilterStrategy<K,V> recordFilterStrategy)	Create an instance with the supplied strategy and delegate listener.
FilteringBatchMessageListenerAdapter(BatchMessageListener<K,V> delegate, RecordFilterStrategy<K,V> recordFilterStrategy, boolean ackDiscarded)	Create an instance with the supplied strategy and delegate listener.

Method Summary

Modifier and Type	Method	Description
void	onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data)	Invoked with data from kafka.
void	onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data, org.apache.kafka.clients.consumer.Consumer<?,?> consumer)	Invoked with data from kafka and provides access to the Consumer.
void	onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data, Acknowledgment acknowledgment)	Invoked with data from kafka.
void	onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> records, Acknowledgment acknowledgment, org.apache.kafka.clients.consumer.Consumer<?,?> consumer)	Invoked with data from kafka and provides access to the Consumer.

Methods inherited from class org.springframework.kafka.listener.adapter.AbstractFilteringMessageListener

filter, getRecordFilterStrategy

Methods inherited from class org.springframework.kafka.listener.adapter.AbstractDelegatingMessageListenerAdapter

getDelegate, onIdleContainer, onPartitionsAssigned, onPartitionsRevoked, registerSeekCallback

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.springframework.kafka.listener.BatchMessageListener

onMessage, wantsPollResult

Methods inherited from interface org.springframework.kafka.listener.ConsumerSeekAware

onFirstPoll, unregisterSeekCallback

Constructor Details

FilteringBatchMessageListenerAdapter

public FilteringBatchMessageListenerAdapter(BatchMessageListener<K,V> delegate,
 RecordFilterStrategy<K,V> recordFilterStrategy)

Create an instance with the supplied strategy and delegate listener.

Parameters:

delegate - the delegate.

recordFilterStrategy - the filter.

FilteringBatchMessageListenerAdapter

public FilteringBatchMessageListenerAdapter(BatchMessageListener<K,V> delegate,
 RecordFilterStrategy<K,V> recordFilterStrategy,
 boolean ackDiscarded)

Create an instance with the supplied strategy and delegate listener. When 'ackDiscarded' is false, and all messages are filtered, an empty list is passed to the delegate (so it can decide whether or not to ack); when true, a completely filtered batch is ack'd by this class, and no call is made to the delegate.

Parameters:

delegate - the delegate.

recordFilterStrategy - the filter.

ackDiscarded - true to ack (commit offset for) discarded messages when the listener is configured for manual acks.

Method Details

onMessage

public void onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> records,
 @Nullable
 Acknowledgment acknowledgment,
 org.apache.kafka.clients.consumer.Consumer<?,?> consumer)

Description copied from interface: GenericMessageListener

Invoked with data from kafka and provides access to the Consumer. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface BatchAcknowledgingConsumerAwareMessageListener<K,V>

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

records - the data to be processed.

acknowledgment - the acknowledgment.

consumer - the consumer.

onMessage

public void onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data)

Description copied from interface: BatchAcknowledgingConsumerAwareMessageListener

Invoked with data from kafka. Containers should never call this since they will detect that we are a consumer aware acknowledging listener.

Specified by:

onMessage in interface BatchAcknowledgingConsumerAwareMessageListener<K,V>

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

onMessage

public void onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data,
 Acknowledgment acknowledgment)

Description copied from interface: GenericMessageListener

Invoked with data from kafka. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

acknowledgment - the acknowledgment.

onMessage

public void onMessage(List<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>> data,
 org.apache.kafka.clients.consumer.Consumer<?,?> consumer)

Description copied from interface: GenericMessageListener

Invoked with data from kafka and provides access to the Consumer. The default implementation throws UnsupportedOperationException.

Specified by:

onMessage in interface GenericMessageListener<K>

Parameters:

data - the data to be processed.

consumer - the consumer.