From 5abad099349692bca68b56fb8d2605f1df1e654d Mon Sep 17 00:00:00 2001
From: chickenchickenlove <ojt90902@naver.com>
Date: Thu, 25 Apr 2024 08:06:10 +0900
Subject: [PATCH] apply review

---
 .../adapter/FilteringBatchMessageListenerAdapter.java  | 10 ++++------
 .../kafka/listener/adapter/RecordFilterStrategy.java   |  4 ++++
 2 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java
index 1095fb59cc..434da0531c 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java
@@ -53,10 +53,7 @@ public class FilteringBatchMessageListenerAdapter<K, V>
 	 */
 	public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 			RecordFilterStrategy<K, V> recordFilterStrategy) {
-
-		super(delegate, recordFilterStrategy);
-		this.ackDiscarded = false;
-		this.consumerAware = this.delegateType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE) || this.delegateType.equals(ListenerType.CONSUMER_AWARE);
+		this(delegate, recordFilterStrategy, false);
 	}
 
 	/**
@@ -81,10 +78,11 @@ public FilteringBatchMessageListenerAdapter(BatchMessageListener<K, V> delegate,
 	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");
 
-		if (consumerRecords.isEmpty()) {
+		if (recordFilterStrategy.ignoreEmptyBatch()) {
 			if (acknowledgment != null) {
 				invokeDelegate(consumerRecords, acknowledgment, consumer);
 			}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java
index f2de862e64..5f13a693ec 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java
@@ -58,4 +58,8 @@ default List<ConsumerRecord<K, V>> filterBatch(List<ConsumerRecord<K, V>> record
 		return records;
 	}
 
+	default boolean ignoreEmptyBatch() {
+		return false;
+	}
+
 }