GH-2134: Add filter to @KafkaListener

Resolves #2134

Also clean up parsing of `errorHandler` and `contentTypeConverter`.

**cherry-pick to 2.8.x**

* Polishing per review comments.

* Add info property to KafkaListener.

* Rebase; add since.

Author: garyrussell

Diff for: spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2315,6 +2315,18 @@ In addition, a `FilteringBatchMessageListenerAdapter` is provided, for when you
 
 IMPORTANT: The `FilteringBatchMessageListenerAdapter` is ignored if your `@KafkaListener` receives a `ConsumerRecords<?, ?>` instead of `List<ConsumerRecord<?, ?>>`, because `ConsumerRecords` is immutable.
 
+Starting with version 2.8.4, you can override the listener container factory's default `RecordFilterStrategy` by using the `filter` property on the listener annotations.
+
+====
+[source, java]
+----
+@KafkaListener(id = "filtered", topics = "topic", filter = "differentFilter")
+public void listen(Thing thing) {
+    ...
+}
+----
+====
+
 [[retrying-deliveries]]
 ===== Retrying Deliveries
 
@@ -2594,6 +2606,7 @@ Mutually exclusive; at least one must be provided; enforced by `ContainerPropert
 |See <<transactions>>.
 |===
 
+[[alc-props]]
 .`AbstractListenerContainer` Properties
 [cols="9,10,16", options="header"]
 |===
@@ -2651,6 +2664,12 @@ See <<error-handlers>>.
 |See desc.
 |The bean name for user-configured containers or the `id` attribute of `@KafkaListener` s.
 
+|[[listenerInfo]]<<listenerInfo,`listenerInfo`>>
+|null
+|A value to populate in the `KafkaHeaders.LISTENER_INFO` header.
+With `@KafkaListener`, this value is obtained from the `info` attribute.
+This header can be used in various places, such as a `RecordInterceptor`, `RecordFilterStrategy` and in the listener code itself.
+
 |[[pauseRequested]]<<pauseRequested,`pauseRequested`>>
 |(read only)
 |True if a consumer pause has been requested.
@@ -5544,6 +5563,30 @@ It is disabled by default to avoid the (small) overhead of looking up the state
 
 The `DefaultErrorHandler` and `DefaultAfterRollbackProcessor` support this feature.
 
+[[li-header]]
+===== Listener Info Header
+
+In some cases, it is useful to be able to know which container a listener is running in.
+
+Starting with version 2.8.4, you can now set the `listenerInfo` property on the listener container, or set the `info` attribute on the `@KafkaListener` annotation.
+Then, the container will add this in the `KafkaListener.LISTENER_INFO` header to all incoming messages; it can then be used in record interceptors, filters, etc., or in the listener itself.
+
+====
+[source, java]
+----
+@KafkaListener(id = "something", topic = "topic", filter = "someFilter",
+        info = "this is the something listener")
+public void listen2(@Payload Thing thing,
+        @Header(KafkaHeaders.LISTENER_INFO) String listenerInfo) {
+...
+}
+----
+====
+
+When used in a `RecordInterceptor` or `RecordFilterStrategy` implementation, the header is in the consumer record as a byte array, converted using the `KafkaListenerAnnotationBeanPostProcessor` 's `charSet` property.
+
+The header mappers also convert to `String` when creating `MessageHeaders` from the consumer record and never map this header on an outbound record.
+
 [[dead-letters]]
 ===== Publishing Dead-letter Records
 

Diff for: spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -44,6 +44,12 @@ See <<batch-listener-conv-errors>> for more information.
 `RecordFilterStrategy`, when used with batch listeners, can now filter the entire batch in one call.
 See the note at the end of <<batch-listeners>> for more information.
 
+The `@KafkaListener` annotation now has the `filter` attribute, to override the container factory's `RecordFilterStrategy` for just this listener.
+
+The `@KafkaListener` annotation now the "info" attribute; this is used to populate the new listener container property `listenerInfo`.
+This is then used to populate a `KafkaHeaders.LISTENER_INFO` header in each record which can be used in `RecordInterceptor`, `RecordFilterStrategy`, or the listener itself.
+See <<li-header>> and <<alc-props,Abstract Listener Container Properties>> for more information.
+
 [[x28-template]]
 ==== `KafkaTemplate` Changes
 

Diff for: spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -297,4 +297,31 @@
 	 */
 	String batch() default "";
 
+	/**
+	 * Set an {@link org.springframework.kafka.listener.adapter.RecordFilterStrategy} bean
+	 * name to override the strategy configured on the container factory. If a SpEL
+	 * expression is provided ({@code #{...}}), the expression can either evaluate to a
+	 * {@link org.springframework.kafka.listener.adapter.RecordFilterStrategy} instance or
+	 * a bean name.
+	 * @return the error handler.
+	 * @since 2.8.4
+	 */
+	String filter() default "";
+
+	/**
+	 * Static information that will be added as a header with key
+	 * {@link org.springframework.kafka.support.KafkaHeaders#LISTENER_INFO}. This can be used, for example, in a
+	 * {@link org.springframework.kafka.listener.RecordInterceptor}, {@link RecordFiorg.springframework.kafka.listener.adapter.RecordFilterStrategylterStrategy} or the listener itself, for
+	 * any purposes.
+	 * <p>
+	 * SpEL {@code #{...}} and property place holders {@code ${...}} are supported, but it
+	 * must resolve to a String or `byte[]`.
+	 * <p>
+	 * This header will be stripped out if an outbound record is created with the headers
+	 * from an input record.
+	 * @return the info.
+	 * @since 2.8.4
+	 */
+	String info() default "";
+
 }

Diff for: spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -85,6 +85,7 @@
 import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
 import org.springframework.kafka.listener.ContainerGroupSequencer;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
+import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.retrytopic.RetryTopicBootstrapper;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 import org.springframework.kafka.retrytopic.RetryTopicConfigurer;
@@ -143,6 +144,12 @@
 public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
 
+	private static final String THE_LEFT = "The [";
+
+	private static final String RESOLVED_TO_LEFT = "Resolved to [";
+
+	private static final String RIGHT_FOR_LEFT = "] for [";
+
 	private static final String GENERATED_ID_PREFIX = "org.springframework.kafka.KafkaListenerEndpointContainer#";
 
 	/**
@@ -251,8 +258,8 @@ public void setBeanFactory(BeanFactory beanFactory) {
 	}
 
 	/**
-	 * Set a charset to use when converting byte[] to String in method arguments.
-	 * Default UTF-8.
+	 * Set a charset to use when converting byte[] to String in method arguments and other
+	 * String/byte[] conversions. Default UTF-8.
 	 * @param charset the charset.
 	 * @since 2.2
 	 */
@@ -489,7 +496,7 @@ private boolean processMainAndRetryListeners(KafkaListener kafkaListener, Object
 		}
 
 		RetryTopicConfigurer.EndpointProcessor endpointProcessor = endpointToProcess ->
-				this.processKafkaListenerAnnotationForRetryTopic(endpointToProcess, kafkaListener, bean, topics, tps);
+				this.processKafkaListenerAnnotation(endpointToProcess, kafkaListener, bean, topics, tps);
 
 		KafkaListenerContainerFactory<?> factory =
 				resolveContainerFactory(kafkaListener, resolve(kafkaListener.containerFactory()), beanName);
@@ -560,24 +567,16 @@ private Method checkProxy(Method methodArg, Object bean) {
 	protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener,
 								Object bean, String beanName, String[] topics, TopicPartitionOffset[] tps) {
 
-		processKafkaListenerAnnotationBeforeRegistration(endpoint, kafkaListener, bean, topics, tps);
+		processKafkaListenerAnnotation(endpoint, kafkaListener, bean, topics, tps);
 
 		String containerFactory = resolve(kafkaListener.containerFactory());
-		KafkaListenerContainerFactory<?> listenerContainerFactory = resolveContainerFactory(kafkaListener, containerFactory, beanName);
+		KafkaListenerContainerFactory<?> listenerContainerFactory = resolveContainerFactory(kafkaListener,
+				containerFactory, beanName);
 
 		this.registrar.registerEndpoint(endpoint, listenerContainerFactory);
-
-		processKafkaListenerEndpointAfterRegistration(endpoint, kafkaListener);
-	}
-
-	private void processKafkaListenerAnnotationForRetryTopic(MethodKafkaListenerEndpoint<?, ?> endpoint,
-			KafkaListener kafkaListener, Object bean, String[] topics, TopicPartitionOffset[] tps) {
-
-		processKafkaListenerAnnotationBeforeRegistration(endpoint, kafkaListener, bean, topics, tps);
-		processKafkaListenerEndpointAfterRegistration(endpoint, kafkaListener);
 	}
 
-	private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint,
+	private void processKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?> endpoint,
 			KafkaListener kafkaListener, Object bean, String[] topics, TopicPartitionOffset[] tps) {
 
 		endpoint.setBean(bean);
@@ -588,6 +587,7 @@ private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListene
 		endpoint.setTopics(topics);
 		endpoint.setTopicPattern(resolvePattern(kafkaListener));
 		endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));
+		endpoint.setListenerInfo(resolveExpressionAsBytes(kafkaListener.info(), "info"));
 		String group = kafkaListener.containerGroup();
 		if (StringUtils.hasText(group)) {
 			Object resolvedGroup = resolveExpression(group);
@@ -608,20 +608,10 @@ private void processKafkaListenerAnnotationBeforeRegistration(MethodKafkaListene
 		if (StringUtils.hasText(kafkaListener.batch())) {
 			endpoint.setBatchListener(Boolean.parseBoolean(kafkaListener.batch()));
 		}
-	}
-
-	private void processKafkaListenerEndpointAfterRegistration(MethodKafkaListenerEndpoint<?, ?> endpoint,
-			KafkaListener kafkaListener) {
-
 		endpoint.setBeanFactory(this.beanFactory);
-		String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");
-		if (StringUtils.hasText(errorHandlerBeanName)) {
-			resolveErrorHandler(endpoint, kafkaListener);
-		}
-		String converterBeanName = resolveExpressionAsString(kafkaListener.contentTypeConverter(), "contentTypeConverter");
-		if (StringUtils.hasText(converterBeanName)) {
-			resolveContentTypeConverter(endpoint, kafkaListener);
-		}
+		resolveErrorHandler(endpoint, kafkaListener);
+		resolveContentTypeConverter(endpoint, kafkaListener);
+		resolveFilter(endpoint, kafkaListener);
 	}
 
 	private void resolveErrorHandler(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
@@ -653,9 +643,24 @@ private void resolveContentTypeConverter(MethodKafkaListenerEndpoint<?, ?> endpo
 		}
 	}
 
+	private void resolveFilter(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
+		Object filter = resolveExpression(kafkaListener.filter());
+		if (filter instanceof RecordFilterStrategy) {
+			endpoint.setRecordFilterStrategy((RecordFilterStrategy) filter);
+		}
+		else {
+			String filterBeanName = resolveExpressionAsString(kafkaListener.filter(), "filter");
+			if (StringUtils.hasText(filterBeanName)) {
+				endpoint.setRecordFilterStrategy(
+						this.beanFactory.getBean(filterBeanName, RecordFilterStrategy.class));
+			}
+		}
+	}
+
 	@Nullable
 	private KafkaListenerContainerFactory<?> resolveContainerFactory(KafkaListener kafkaListener,
 			Object factoryTarget, String beanName) {
+
 		String containerFactory = kafkaListener.containerFactory();
 		if (!StringUtils.hasText(containerFactory)) {
 			return null;
@@ -934,8 +939,26 @@ private String resolveExpressionAsString(String value, String attribute) {
 			return (String) resolved;
 		}
 		else if (resolved != null) {
-			throw new IllegalStateException("The [" + attribute + "] must resolve to a String. "
-					+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
+			throw new IllegalStateException(THE_LEFT + attribute + "] must resolve to a String. "
+					+ RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]");
+		}
+		return null;
+	}
+
+	@Nullable
+	private byte[] resolveExpressionAsBytes(String value, String attribute) {
+		Object resolved = resolveExpression(value);
+		if (resolved instanceof String) {
+			if (StringUtils.hasText((CharSequence) resolved)) {
+				return ((String) resolved).getBytes(this.charset);
+			}
+		}
+		else if (resolved instanceof byte[]) {
+			return (byte[]) resolved;
+		}
+		else if (resolved != null) {
+			throw new IllegalStateException(THE_LEFT + attribute + "] must resolve to a String or byte[]. "
+					+ RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]");
 		}
 		return null;
 	}
@@ -951,8 +974,8 @@ else if (resolved instanceof Number) {
 		}
 		else if (resolved != null) {
 			throw new IllegalStateException(
-					"The [" + attribute + "] must resolve to an Number or a String that can be parsed as an Integer. "
-							+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
+					THE_LEFT + attribute + "] must resolve to an Number or a String that can be parsed as an Integer. "
+							+ RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]");
 		}
 		return result;
 	}
@@ -968,8 +991,8 @@ else if (resolved instanceof String) {
 		}
 		else if (resolved != null) {
 			throw new IllegalStateException(
-					"The [" + attribute + "] must resolve to a Boolean or a String that can be parsed as a Boolean. "
-							+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
+					THE_LEFT + attribute + "] must resolve to a Boolean or a String that can be parsed as a Boolean. "
+							+ RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]");
 		}
 		return result;
 	}

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

@@ -369,10 +369,12 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		return instance;
 	}
 
-	@SuppressWarnings("deprecation")
 	private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint) {
+		if (aklEndpoint.getRecordFilterStrategy() == null) {
+			JavaUtils.INSTANCE
+					.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy);
+		}
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy)
 				.acceptIfNotNull(this.ackDiscarded, aklEndpoint::setAckDiscarded)
 				.acceptIfNotNull(this.statefulRetry, aklEndpoint::setStatefulRetry)
 				.acceptIfNotNull(this.replyTemplate, aklEndpoint::setReplyTemplate)
@@ -381,7 +383,6 @@ private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint)
 		if (aklEndpoint.getBatchListener() == null) {
 			JavaUtils.INSTANCE
 					.acceptIfNotNull(this.batchListener, aklEndpoint::setBatchListener);
-
 		}
 	}
 
@@ -431,7 +432,8 @@ else if (this.autoStartup != null) {
 				.acceptIfHasText(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
 				.acceptIfHasText(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
 				.acceptIfNotNull(endpoint.getConsumerProperties(),
-						instance.getContainerProperties()::setKafkaConsumerProperties);
+						instance.getContainerProperties()::setKafkaConsumerProperties)
+				.acceptIfNotNull(endpoint.getListenerInfo(), instance::setListenerInfo);
 	}
 
 	private void customizeContainer(C instance) {

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

@@ -116,6 +116,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private BatchToRecordAdapter<K, V> batchToRecordAdapter;
 
+	private byte[] listenerInfo;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -432,6 +434,21 @@ public void setSplitIterables(boolean splitIterables) {
 		this.splitIterables = splitIterables;
 	}
 
+	@Override
+	@Nullable
+	public byte[] getListenerInfo() {
+		return this.listenerInfo; // NOSONAR
+	}
+
+	/**
+	 * Set the listener info to insert in the record header.
+	 * @param listenerInfo the info.
+	 * @since 2.8.4
+	 */
+	public void setListenerInfo(@Nullable byte[] listenerInfo) {  // NOSONAR
+		this.listenerInfo = listenerInfo; // NOSONAR
+	}
+
 	@Nullable
 	protected BatchToRecordAdapter<K, V> getBatchToRecordAdapter() {
 		return this.batchToRecordAdapter;
@@ -446,6 +463,7 @@ public void setBatchToRecordAdapter(BatchToRecordAdapter<K, V> batchToRecordAdap
 		this.batchToRecordAdapter = batchToRecordAdapter;
 	}
 
+
 	@Override
 	public void afterPropertiesSet() {
 		boolean topicsEmpty = getTopics().isEmpty();