GH-1938: Support Whole Batch Filtering

Resolves #1938

Previously, when adding a filter strategy to a batch listener, the batch
members were filtered one-at-a-time. When database lookups are involved
it is more efficient to act on the entire batch.

Add another method to the filter strategy to allow filtering the entire
batch in one call.

* Add unsaved doc changes.

* Fix what's new anchor.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -1624,6 +1624,7 @@ public void pollResults(ConsumerRecords<?, ?> records) {
 
 IMPORTANT: If the container factory has a `RecordFilterStrategy` configured, it is ignored for `ConsumerRecords<?, ?>` listeners, with a `WARN` log message emitted.
 Records can only be filtered with a batch listener if the `<List<?>>` form of listener is used.
+By default, records are filtered one-at-a-time; starting with version 2.8, you can override `filterBatch` to filter the entire batch in one call.
 
 [[annotation-properties]]
 ====== Annotation Properties

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

@@ -27,6 +27,9 @@ Batch listeners can now handle conversion exceptions.
 
 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.
+
 [[x28-template]]
 ==== `KafkaTemplate` Changes
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -42,6 +42,10 @@ protected AbstractFilteringMessageListener(T delegate, RecordFilterStrategy<K, V
 		this.recordFilterStrategy = recordFilterStrategy;
 	}
 
+	protected RecordFilterStrategy<K, V> getRecordFilterStrategy() {
+		return this.recordFilterStrategy;
+	}
+
 	protected boolean filter(ConsumerRecord<K, V> consumerRecord) {
 		return this.recordFilterStrategy.filter(consumerRecord);
 	}

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

@@ -16,7 +16,6 @@
 
 package org.springframework.kafka.listener.adapter;
 
-import java.util.Iterator;
 import java.util.List;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -27,6 +26,7 @@
 import org.springframework.kafka.listener.ListenerType;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 
 /**
  * A {@link BatchMessageListener} adapter that implements filter logic
@@ -51,6 +51,7 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 	 */
 	public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 			RecordFilterStrategy<K, V> recordFilterStrategy) {
+
 		super(delegate, recordFilterStrategy);
 		this.ackDiscarded = false;
 	}
@@ -67,19 +68,17 @@ public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 	 */
 	public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 			RecordFilterStrategy<K, V> recordFilterStrategy, boolean ackDiscarded) {
+
 		super(delegate, recordFilterStrategy);
 		this.ackDiscarded = ackDiscarded;
 	}
 
 	@Override
-	public void onMessage(List<ConsumerRecord<K, V>> consumerRecords, @Nullable Acknowledgment acknowledgment,
+	public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer) {
-		Iterator<ConsumerRecord<K, V>> iterator = consumerRecords.iterator();
-		while (iterator.hasNext()) {
-			if (filter(iterator.next())) {
-				iterator.remove();
-			}
-		}
+
+		List<ConsumerRecord<K, V>> consumerRecords = getRecordFilterStrategy().filterBatch(records);
+		Assert.state(consumerRecords != null, "filter returned null from filterBatch");
 		boolean consumerAware = this.delegateType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE)
 						|| this.delegateType.equals(ListenerType.CONSUMER_AWARE);
 		/*

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -16,6 +16,9 @@
 
 package org.springframework.kafka.listener.adapter;
 
+import java.util.Iterator;
+import java.util.List;
+
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 /**
@@ -38,4 +41,21 @@ public interface RecordFilterStrategy<K, V> {
 	 */
 	boolean filter(ConsumerRecord<K, V> consumerRecord);
 
+	/**
+	 * Filter an entire batch of records; to filter all records, return an empty list, not
+	 * null.
+	 * @param records the records.
+	 * @return the filtered records.
+	 * @since 2.8
+	 */
+	default List<ConsumerRecord<K, V>> filterBatch(List<ConsumerRecord<K, V>> records) {
+		Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
+		while (iterator.hasNext()) {
+			if (filter(iterator.next())) {
+				iterator.remove();
+			}
+		}
+		return records;
+	}
+
 }

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

@@ -552,7 +552,7 @@ public void testBatch() throws Exception {
 		List<?> list = (List<?>) this.listener.payload;
 		assertThat(list.size()).isGreaterThan(0);
 		assertThat(list.get(0)).isInstanceOf(String.class);
-		assertThat(this.recordFilter.called).isTrue();
+		assertThat(this.recordFilter.batchCalled).isTrue();
 		assertThat(this.config.listen10Exception).isNotNull();
 		assertThat(this.listener.receivedGroupId).isEqualTo("list1");
 
@@ -2474,14 +2474,25 @@ public void setBar(String bar) {
 
 	public static class RecordPassAllFilter implements RecordFilterStrategy<Integer, CharSequence> {
 
-		private boolean called;
+		boolean called;
+
+		boolean batchCalled;
 
 		@Override
 		public boolean filter(ConsumerRecord<Integer, CharSequence> consumerRecord) {
-			called = true;
+			this.called = true;
 			return false;
 		}
 
+		@Override
+		public List<ConsumerRecord<Integer, CharSequence>> filterBatch(
+				List<ConsumerRecord<Integer, CharSequence>> records) {
+
+			this.batchCalled = true;
+			return records;
+		}
+
+
 	}
 
 	public static class FooConverter implements Converter<String, Foo> {