apply review

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

@@ -36,6 +36,7 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Sanghyeok An
  *
  */
 public class FilteringBatchMessageListenerAdapter<K, V>
@@ -82,20 +83,18 @@ public void onMessage(List<ConsumerRecord<K, V>> records, @Nullable Acknowledgme
 		final List<ConsumerRecord<K, V>> consumerRecords = recordFilterStrategy.filterBatch(records);
 		Assert.state(consumerRecords != null, "filter returned null from filterBatch");
 
-		if (recordFilterStrategy.ignoreEmptyBatch()) {
-			if (acknowledgment != null) {
-				invokeDelegate(consumerRecords, acknowledgment, consumer);
-			}
+		if (recordFilterStrategy.ignoreEmptyBatch() &&
+			consumerRecords.isEmpty() &&
+			acknowledgment != null) {
+			acknowledgment.acknowledge();
+		}
+		else if (consumerRecords.size() > 0 || this.consumerAware
+			|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
+			invokeDelegate(consumerRecords, acknowledgment, consumer);
 		}
 		else {
-			if (this.consumerAware
-				|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
-				invokeDelegate(consumerRecords, acknowledgment, consumer);
-			}
-			else {
-				if (this.ackDiscarded && acknowledgment != null) {
-					acknowledgment.acknowledge();
-				}
+			if (this.ackDiscarded && acknowledgment != null) {
+				acknowledgment.acknowledge();
 			}
 		}
 	}

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.
@@ -21,6 +21,8 @@
 
 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 +32,7 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
- *
+ * @author Sanghyeok An
  */
 public interface RecordFilterStrategy<K, V> {
 
@@ -58,6 +60,12 @@ default List<ConsumerRecord<K, V>> filterBatch(List<ConsumerRecord<K, V>> record
 		return records;
 	}
 
+	/**
+	 * Returns a boolean value that determines whether {@link FilteringBatchMessageListenerAdapter} invoke the {@link BatchMessageListener} when all {@link ConsumerRecord}
+	 * have been filtered and return an EmptyList. The default is not to invoke the {@link BatchMessageListener} (false).
+	 * @return If true is returned, the {@link FilteringBatchMessageListenerAdapter} will not invoke the {@link BatchMessageListener}
+	 * @since 3.2.0
+	 */
 	default boolean ignoreEmptyBatch() {
 		return false;
 	}