Polishing - ackDiscarded on @KafkaListener

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -53,6 +53,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private RecordFilterStrategy<K, V> recordFilterStrategy;
 
+	private Boolean ackDiscarded;
+
 	private ApplicationEventPublisher applicationEventPublisher;
 
 	/**
@@ -94,13 +96,21 @@ public void setMessageConverter(MessageConverter messageConverter) {
 	}
 
 	/**
-	 * Set the de-duplication strategy.
+	 * Set the record filter strategy.
 	 * @param recordFilterStrategy the strategy.
 	 */
 	public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
+	/**
+	 * Set to true to ack discards when a filter strategy is in use.
+	 * @param ackDiscarded the ackDiscarded.
+	 */
+	public void setAckDiscarded(Boolean ackDiscarded) {
+		this.ackDiscarded = ackDiscarded;
+	}
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -133,8 +143,13 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 			instance.setBeanName(endpoint.getId());
 		}
 
-		if (this.recordFilterStrategy != null && endpoint instanceof AbstractKafkaListenerEndpoint) {
-			((AbstractKafkaListenerEndpoint<K, V>) endpoint).setRecordFilterStrategy(this.recordFilterStrategy);
+		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
+			if (this.recordFilterStrategy != null) {
+				((AbstractKafkaListenerEndpoint<K, V>) endpoint).setRecordFilterStrategy(this.recordFilterStrategy);
+			}
+			if (this.ackDiscarded != null) {
+				((AbstractKafkaListenerEndpoint<K, V>) endpoint).setAckDiscarded(this.ackDiscarded);
+			}
 		}
 
 		endpoint.setupListenerContainer(instance, this.messageConverter);

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -31,8 +31,11 @@
 import org.springframework.beans.factory.config.BeanExpressionContext;
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.kafka.listener.AcknowledgingMessageListener;
 import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.util.Assert;
@@ -70,6 +73,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private RecordFilterStrategy<K, V> recordFilterStrategy;
 
+	private boolean ackDiscarded;
+
 
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
@@ -208,6 +213,19 @@ public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrat
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
+	protected boolean isAckDiscarded() {
+		return this.ackDiscarded;
+	}
+
+	/**
+	 * Set to true if the {@link #setRecordFilterStrategy(RecordFilterStrategy) recordFilterStrategy}
+	 * is in use.
+	 * @param ackDiscarded the ackDiscarded.
+	 */
+	public void setAckDiscarded(boolean ackDiscarded) {
+		this.ackDiscarded = ackDiscarded;
+	}
+
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer, MessageConverter messageConverter) {
 		setupMessageListener(listenerContainer, messageConverter);
@@ -226,7 +244,21 @@ protected abstract MessageListener<K, V> createMessageListener(MessageListenerCo
 	private void setupMessageListener(MessageListenerContainer container, MessageConverter messageConverter) {
 		MessageListener<K, V> messageListener = createMessageListener(container, messageConverter);
 		Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener");
-		container.setupMessageListener(messageListener);
+		if (this.recordFilterStrategy != null) {
+			if (messageListener instanceof AcknowledgingMessageListener) {
+				@SuppressWarnings("unchecked")
+				AcknowledgingMessageListener<K, V> aml = (AcknowledgingMessageListener<K, V>) messageListener;
+				aml = new FilteringAcknowledgingMessageListenerAdapter<>(this.recordFilterStrategy, aml, this.ackDiscarded);
+				container.setupMessageListener(aml);
+			}
+			else {
+				messageListener = new FilteringMessageListenerAdapter<>(this.recordFilterStrategy, messageListener);
+				container.setupMessageListener(messageListener);
+			}
+		}
+		else {
+			container.setupMessageListener(messageListener);
+		}
 	}
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

@@ -94,9 +94,6 @@ protected MessagingMessageListenerAdapter<K, V> createMessageListener(MessageLis
 		Assert.state(this.messageHandlerMethodFactory != null,
 				"Could not create message listener - MessageHandlerMethodFactory not set");
 		MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance();
-		if (getRecordFilterStrategy() != null) {
-			messageListener.setRecordFilterStrategy(getRecordFilterStrategy());
-		}
 		messageListener.setHandlerMethod(configureListenerAdapter(messageListener));
 		if (messageConverter != null) {
 			messageListener.setMessageConverter(messageConverter);

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

@@ -59,8 +59,6 @@ public class MessagingMessageListenerAdapter<K, V> extends AbstractAdaptableMess
 
 	private MessageConverter messageConverter = new MessagingMessageConverter();
 
-	private RecordFilterStrategy<K, V> recordFilterStrategy;
-
 
 	public MessagingMessageListenerAdapter(Method method) {
 		this.inferredType = determineInferredType(method);
@@ -103,18 +101,6 @@ public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment
 		invokeHandler(record, acknowledgment, message);
 	}
 
-	protected RecordFilterStrategy<K, V> getRecordFilterStrategy() {
-		return this.recordFilterStrategy;
-	}
-
-	/**
-	 * Set a {@link RecordFilterStrategy} implementation.
-	 * @param recordFilterStrategy the strategy implementation.
-	 */
-	public void setRecordFilterStrategy(RecordFilterStrategy<K, V> recordFilterStrategy) {
-		this.recordFilterStrategy = recordFilterStrategy;
-	}
-
 	protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment) {
 		return getMessageConverter().toMessage(record, acknowledgment, this.inferredType);
 	}
@@ -128,9 +114,6 @@ protected Message<?> toMessagingMessage(ConsumerRecord<K, V> record, Acknowledgm
 	 * @return the result of invocation.
 	 */
 	private Object invokeHandler(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Message<?> message) {
-		if (this.recordFilterStrategy != null && this.recordFilterStrategy.filter(record)) {
-			return null;
-		}
 		try {
 			return this.handlerMethod.invoke(message, record, acknowledgment);
 		}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -50,6 +50,7 @@
 import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.adapter.FilteringAcknowledgingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.Acknowledgment;
@@ -128,6 +129,11 @@ public void testSimple() throws Exception {
 		assertThat(this.listener.ack).isNotNull();
 		assertThat(this.listener.eventLatch.await(20, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.listener.event.getListenerId().startsWith("qux-"));
+		MessageListenerContainer manualContainer = this.registry.getListenerContainer("qux");
+		assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener"))
+				.isInstanceOf(FilteringAcknowledgingMessageListenerAdapter.class);
+		assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener.ackDiscarded",
+				Boolean.class)).isTrue();
 
 		template.send("annotated5", 0, 0, "foo");
 		template.send("annotated5", 1, 0, "bar");
@@ -214,12 +220,17 @@ public PlatformTransactionManager transactionManager() {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setRecordFilterStrategy(filter());
+			factory.setRecordFilterStrategy(recordFilter());
 			return factory;
 		}
 
 		@Bean
-		public RecordFilterImpl filter() {
+		public RecordFilterImpl recordFilter() {
+			return new RecordFilterImpl();
+		}
+
+		@Bean
+		public RecordFilterImpl manualFilter() {
 			return new RecordFilterImpl();
 		}
 
@@ -241,6 +252,8 @@ public KafkaListenerContainerFactory<?> kafkaJsonListenerContainerFactory() {
 			ContainerProperties props = factory.getContainerProperties();
 			props.setAckMode(AckMode.MANUAL_IMMEDIATE);
 			props.setIdleEventInterval(100L);
+			factory.setRecordFilterStrategy(manualFilter());
+			factory.setAckDiscarded(true);
 			return factory;
 		}
 

src/reference/asciidoc/kafka.adoc

@@ -336,19 +336,19 @@ The framework cannot know whether such a message has been processed or not, that
 function.
 This is known as the http://www.enterpriseintegrationpatterns.com/patterns/messaging/IdempotentReceiver.html[Idempotent
 Receiver] pattern and Spring Integration provides an
-http://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#idempotent-receiver
-[implementation thereof].
+http://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#idempotent-receiver[implementation thereof].
 
 The Spring for Apache Kafka project also provides some assistance by means of the `FilteringMessageListenerAdapter`
 classe, which can wrap your `MessageListener`.
 This class takes an implementation of `RecordFilterStrategy` where you implement the `filter` method to signal
 that a message is a duplicate and should be discarded.
 
-Similarly, when using `@KafkaListener`, the `RecordFilterStrategy` can be injected into the container factory.
-
 A `FilteringAcknowledgingMessageListenerAdapter` is also provided for wrapping an `AcknowledgingMessageListener`.
 This has an additional property `ackDiscarded` which indicates whether the adapter should acknowledge the discarded record.
 
+When using `@KafkaListener`, set the `RecordFilterStrategy` (and optionally `ackDiscarded`) on the container factory and the listener will be wrapped in the appropriate filtering adapter.
+
+
 ==== Serialization/Deserialization and Message Conversion
 
 Apache Kafka provides a high-level API for serializing/deserializing record values as well as their keys.