GH-2806: Receiving an empty list with RecordFilterStrategy

Fixes: #2806

Motivation:

    Receiving an empty list when using `RecordFilterStrategy` on batch messages
    In the current batch mode, even if the `RecordFilterStrategy` filters all records resulting in an Empty List being returned, the KafkaListener is still invoked.
    In contrast, in single record mode, if record are filtered, the `KafkaListener` is not called.
    This difference in behavior between the two modes can cause confusion for users.

Modifications:

    Add public method `isAnyManualAck()` to `Acknowledgment` to verify that `manualAck` is needed on `FilteringBatchMessageListenerAdapter`.
    Modify `FilteringBatchMessageListenerAdapter`.
        add field `consumerAware` as final (IMHO, we don't need to calculate it every single call `onMessage()`).
        add logic (if empty list and manual Ack == true, KafkaListener will be invoked.
        If empty list and manual Ack == false, `KafkaListener` will not be invoked even if listener is kind of ConsumerAware.
        In detail, See Discussion section below.

Result:

    Receiving an empty list when using RecordFilterStrategy on batch messages #2806
    When the RecordFilterStrategy filters all records and returns an Empty List, the KafkaListener is invoked only if it is in manual ACK mode.

Discussion:

    When using a ConsumerAware Listener, commits can be made using `Consumer.commitSync()` and `Consumer.commitAsync()`.
    However, when using a `ConsumerAwareAckListener`, it seems possible that commits using the Consumer and commits using Ack could be processed simultaneously.
    That situation seems quite ambiguous.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc

@@ -27,3 +27,63 @@ public void listen(Thing thing) {
 }
 ----
 
+Starting with version 3.3, Ignoring empty batches that result from filtering by `RecordFilterStrategy` is supported.
+When implementing `RecordFilterStrategy`, it can be configured through `ignoreEmptyBatch()`.
+The default setting is `false`, indicating `KafkaListener` will be invoked even if all `ConsumerRecord` s are filtered out.
+
+If `true` is returned, the `KafkaListener` [underline]#will not be invoked# when all `ConsumerRecord` are filtered out.
+However, commit to broker, will still be executed.
+
+If `false` is returned, the `KafkaListener` [underline]#will be invoked# when all `ConsumerRecord` are filtered out.
+
+Here are some examples.
+
+[source,java]
+----
+public class IgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
+    ...
+    @Override
+    public List<ConsumerRecord<String, String>> filterBatch(
+            List<ConsumerRecord<String, String>> consumerRecords) {
+        return List.of();
+    }
+
+    @Override
+    public boolean ignoreEmptyBatch() {
+        return true;
+    }
+};
+
+// NOTE: ignoreEmptyBatchRecordFilterStrategy is bean name of IgnoreEmptyBatchRecordFilterStrategy instance.
+@KafkaListener(id = "filtered", topics = "topic", filter = "ignoreEmptyBatchRecordFilterStrategy")
+public void listen(List<Thing> things) {
+    ...
+}
+----
+In this case, `IgnoreEmptyBatchRecordFilterStrategy` always returns empty list and return `true` as result of `ignoreEmptyBatch()`.
+Thus `KafkaListener#listen(...)` never will be invoked at all.
+
+[source,java]
+----
+public class NotIgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy {
+    ...
+    @Override
+    public List<ConsumerRecord<String, String>> filterBatch(
+            List<ConsumerRecord<String, String>> consumerRecords) {
+        return List.of();
+    }
+
+    @Override
+    public boolean ignoreEmptyBatch() {
+        return false;
+    }
+};
+
+// NOTE: notIgnoreEmptyBatchRecordFilterStrategy is bean name of NotIgnoreEmptyBatchRecordFilterStrategy instance.
+@KafkaListener(id = "filtered", topics = "topic", filter = "notIgnoreEmptyBatchRecordFilterStrategy")
+public void listen(List<Thing> things) {
+    ...
+}
+----
+However, in this case, `IgnoreEmptyBatchRecordFilterStrategy` always returns empty list and return `false` as result of `ignoreEmptyBatch()`.
+Thus `KafkaListener#listen(...)` always will be invoked.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -19,4 +19,9 @@ A new method, `getGroupId()`, has been added to the `ConsumerSeekCallback` inter
 This method allows for more selective seek operations by targeting only the desired consumer group.
 For more details, see xref:kafka/seek.adoc#seek[Seek API Docs].
 
+[[x33-new-option-ignore-empty-batch]]
+=== Configurable Handling of Empty Batches in Kafka Listener with RecordFilterStrategy
 
+`RecordFilterStrategy` now supports ignoring empty batches that result from filtering.
+This can be configured through overriding default method `ignoreEmptyBatch()`, which defaults to false, ensuring `KafkaListener` is invoked even if all `ConsumerRecords` are filtered out.
+For more details, see xref:kafka/receiving-messages/filtering.adoc[Message receive filtering Docs].

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2024 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.
@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Sanghyeok An
  *
  */
 public class FilteringBatchMessageListenerAdapter<K, V>
@@ -44,6 +45,8 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 
 	private final boolean ackDiscarded;
 
+	private final boolean consumerAware;
+
 	/**
 	 * Create an instance with the supplied strategy and delegate listener.
 	 * @param delegate the delegate.
@@ -51,9 +54,7 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 	 */
 	public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 			RecordFilterStrategy<K, V> recordFilterStrategy) {
-
-		super(delegate, recordFilterStrategy);
-		this.ackDiscarded = false;
+		this(delegate, recordFilterStrategy, false);
 	}
 
 	/**
@@ -71,22 +72,25 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 
 		super(delegate, recordFilterStrategy);
 		this.ackDiscarded = ackDiscarded;
+		this.consumerAware = this.delegateType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE) ||
+							this.delegateType.equals(ListenerType.CONSUMER_AWARE);
 	}
 
 	@Override
 	public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer) {
 
-		List<ConsumerRecord<K, V>> consumerRecords = getRecordFilterStrategy().filterBatch(records);
+		final RecordFilterStrategy<K, V> recordFilterStrategy = getRecordFilterStrategy();
+		final List<ConsumerRecord<K, V>> consumerRecords = recordFilterStrategy.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);
-		/*
-		 *  An empty list goes to the listener if ackDiscarded is false and the listener can ack
-		 *  either through the acknowledgment
-		 */
-		if (consumerRecords.size() > 0 || consumerAware
-				|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
+
+		if (recordFilterStrategy.ignoreEmptyBatch() &&
+			consumerRecords.isEmpty() &&
+			acknowledgment != null) {
+			acknowledgment.acknowledge();
+		}
+		else if (!consumerRecords.isEmpty() || this.consumerAware
+			|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
 			invokeDelegate(consumerRecords, acknowledgment, consumer);
 		}
 		else {
@@ -98,6 +102,7 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 
 	private void invokeDelegate(List<ConsumerRecord<K, V>> consumerRecords, Acknowledgment acknowledgment,
 			Consumer<?, ?> consumer) {
+
 		switch (this.delegateType) {
 			case ACKNOWLEDGING_CONSUMER_AWARE:
 				this.delegate.onMessage(consumerRecords, acknowledgment, consumer);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2024 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,11 +16,12 @@
 
 package org.springframework.kafka.listener.adapter;
 
-import java.util.Iterator;
 import java.util.List;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
+import org.springframework.kafka.listener.BatchMessageListener;
+
 /**
  * Implementations of this interface can signal that a record about
  * to be delivered to a message listener should be discarded instead
@@ -30,7 +31,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
  * @param <V> the value type.
  *
  * @author Gary Russell
- *
+ * @author Sanghyeok An
  */
 public interface RecordFilterStrategy<K, V> {
 
@@ -49,13 +50,20 @@ public interface RecordFilterStrategy<K, V> {
 	 * @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();
-			}
-		}
+		records.removeIf(this::filter);
 		return records;
 	}
 
+	/**
+	 * Determine whether {@link FilteringBatchMessageListenerAdapter} should invoke
+	 * the {@link BatchMessageListener} when all {@link ConsumerRecord}s in a batch have been filtered out
+	 * resulting in empty list. By default, do invoke the {@link BatchMessageListener} (return false).
+	 * @return true for {@link FilteringBatchMessageListenerAdapter} to {@link BatchMessageListener}
+	 * when all {@link ConsumerRecord} in a batch filtered out
+	 * @since 3.3
+	 */
+	default boolean ignoreEmptyBatch() {
+		return false;
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/FilteringAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -21,6 +21,7 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.only;
 import static org.mockito.Mockito.verify;
 
 import java.util.ArrayList;
@@ -29,6 +30,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
@@ -36,6 +38,7 @@ import org.springframework.kafka.support.Acknowledgment;
 
 /**
  * @author Gary Russell
+ * @author Sanghyeok An
  * @since 2.0
  *
  */
@@ -46,7 +49,7 @@ public class FilteringAdapterTests {
 	public void testBatchFilter() throws Exception {
 		BatchAcknowledgingMessageListener<String, String> listener = mock(BatchAcknowledgingMessageListener.class);
 		FilteringBatchMessageListenerAdapter<String, String> adapter =
-				new FilteringBatchMessageListenerAdapter<String, String>(listener, r -> false);
+				new FilteringBatchMessageListenerAdapter<>(listener, r -> false);
 		List<ConsumerRecord<String, String>> consumerRecords = new ArrayList<>();
 		final CountDownLatch latch = new CountDownLatch(1);
 		willAnswer(i -> {
@@ -64,12 +67,174 @@ public class FilteringAdapterTests {
 	public void testBatchFilterAckDiscard() throws Exception {
 		BatchAcknowledgingMessageListener<String, String> listener = mock(BatchAcknowledgingMessageListener.class);
 		FilteringBatchMessageListenerAdapter<String, String> adapter =
-				new FilteringBatchMessageListenerAdapter<String, String>(listener, r -> false, true);
+				new FilteringBatchMessageListenerAdapter<>(listener, r -> false, true);
 		List<ConsumerRecord<String, String>> consumerRecords = new ArrayList<>();
 		final CountDownLatch latch = new CountDownLatch(1);
-		adapter.onMessage(consumerRecords, () -> latch.countDown(), null);
+		adapter.onMessage(consumerRecords, latch::countDown, null);
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class));
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void listener_should_not_be_invoked_on_emptyList_and_ignoreEmptyBatch_true() {
+		// Given :
+		RecordFilterStrategy<String, String> filter = new RecordFilterStrategy<>() {
+
+			@Override
+			public boolean filter(ConsumerRecord<String, String> consumerRecord) {
+				return true;
+			}
+
+			@Override
+			public List<ConsumerRecord<String, String>> filterBatch(
+					List<ConsumerRecord<String, String>> consumerRecords) {
+				return List.of();
+			}
+
+			@Override
+			public boolean ignoreEmptyBatch() {
+				return true;
+			}
+		};
+
+		BatchAcknowledgingMessageListener<String, String> listener = mock();
+		FilteringBatchMessageListenerAdapter<String, String> adapter =
+				new FilteringBatchMessageListenerAdapter<>(listener, filter);
+		List<ConsumerRecord<String, String>> consumerRecords = new ArrayList<>();
+		Acknowledgment ack = mock();
+
+		// When :
+		adapter.onMessage(consumerRecords, ack, null);
+
+		// Then
+		verify(ack, only()).acknowledge();
+		verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class), any(KafkaConsumer.class));
+		verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class));
+		verify(listener, never()).onMessage(any(List.class), any(KafkaConsumer.class));
+		verify(listener, never()).onMessage(any(List.class));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void listener_should_be_invoked_on_notEmptyList_and_ignoreEmptyBatch_true() throws Exception {
+		// Given :
+		RecordFilterStrategy<String, String> filter = new RecordFilterStrategy<>() {
+
+			@Override
+			public boolean filter(ConsumerRecord<String, String> consumerRecord) {
+				return true;
+			}
+
+			@Override
+			public List<ConsumerRecord<String, String>> filterBatch(
+					List<ConsumerRecord<String, String>> consumerRecords) {
+				return consumerRecords;
+			}
+
+			@Override
+			public boolean ignoreEmptyBatch() {
+				return true;
+			}
+		};
+
+		BatchAcknowledgingMessageListener<String, String> listener = mock();
+		FilteringBatchMessageListenerAdapter<String, String> adapter =
+				new FilteringBatchMessageListenerAdapter<>(listener, filter);
+		List<ConsumerRecord<String, String>> consumerRecords =
+				List.of(new ConsumerRecord<>("hello-topic", 1, 1, "hello-key", "hello-value"));
+		Acknowledgment ack = mock();
+
+		CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(i -> {
+			latch.countDown();
+			return null;
+		}).given(listener).onMessage(any(List.class), any(Acknowledgment.class));
+
+		// When :
+		adapter.onMessage(consumerRecords, ack, null);
+
+		// Then
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(ack, never()).acknowledge();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void listener_should_be_invoked_on_emptyList_and_ignoreEmptyBatch_false() throws Exception {
+		// Given :
+		RecordFilterStrategy<String, String> filter = new RecordFilterStrategy<>() {
+
+			@Override
+			public boolean filter(ConsumerRecord<String, String> consumerRecord) {
+				return true;
+			}
+
+			@Override
+			public List<ConsumerRecord<String, String>> filterBatch(
+					List<ConsumerRecord<String, String>> consumerRecords) {
+				return List.of();
+			}
+		};
+
+		BatchAcknowledgingMessageListener<String, String> listener = mock();
+		FilteringBatchMessageListenerAdapter<String, String> adapter =
+				new FilteringBatchMessageListenerAdapter<>(listener, filter);
+		List<ConsumerRecord<String, String>> consumerRecords = new ArrayList<>();
+		Acknowledgment ack = mock();
+
+		CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(i -> {
+			latch.countDown();
+			return null;
+		}).given(listener).onMessage(any(List.class), any(Acknowledgment.class));
+
+		// When :
+		adapter.onMessage(consumerRecords, ack, null);
+
+		// Then
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(ack, never()).acknowledge();
+	}
+
+	@SuppressWarnings("unchecked")
+	@Test
+	public void listener_should_be_invoked_on_notEmptyList_and_ignoreEmptyBatch_false() throws Exception {
+		// Given :
+		RecordFilterStrategy<String, String> filter = new RecordFilterStrategy<>() {
+
+			@Override
+			public boolean filter(ConsumerRecord<String, String> consumerRecord) {
+				return true;
+			}
+
+			@Override
+			public List<ConsumerRecord<String, String>> filterBatch(
+					// System Under Test
+					List<ConsumerRecord<String, String>> consumerRecords) {
+				return consumerRecords;
+			}
+		};
+
+		BatchAcknowledgingMessageListener<String, String> listener = mock();
+		FilteringBatchMessageListenerAdapter<String, String> adapter =
+				new FilteringBatchMessageListenerAdapter<>(listener, filter);
+		List<ConsumerRecord<String, String>> consumerRecords =
+				List.of(new ConsumerRecord<>("hello-topic", 1, 1, "hello-key", "hello-value"));
+		Acknowledgment ack = mock();
+
+		final CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(i -> {
+			latch.countDown();
+			return null;
+		}).given(listener).onMessage(any(List.class), any(Acknowledgment.class));
+
+		// When :
+		adapter.onMessage(consumerRecords, ack, null);
+
+		// Then
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(ack, never()).acknowledge();
+	}
+
 }