feat: implement AckMode RECORD_FILTERED

Signed-off-by: Chaedong Im <chaedong.im.dev@gmail.com>

Author: Chaedie

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -111,6 +111,7 @@
 import org.springframework.kafka.listener.ContainerProperties.AssignmentCommitOption;
 import org.springframework.kafka.listener.ContainerProperties.EOSMode;
 import org.springframework.kafka.listener.adapter.AsyncRepliesAware;
+import org.springframework.kafka.listener.adapter.FilteringAware;
 import org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
@@ -172,6 +173,7 @@
  * @author Christian Fredriksson
  * @author Timofey Barabanov
  * @author Janek Lasocki-Biczysko
+ * @author Chaedong Im
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -677,6 +679,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final boolean isRecordAck;
 
+		private final boolean isRecordFilteredAck;
+
 		private final BlockingQueue<ConsumerRecord<K, V>> acks = new LinkedBlockingQueue<>();
 
 		private final BlockingQueue<TopicPartitionOffset> seeks = new LinkedBlockingQueue<>();
@@ -871,6 +875,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 			this.isManualImmediateAck = AckMode.MANUAL_IMMEDIATE.equals(this.ackMode);
 			this.isAnyManualAck = this.isManualAck || this.isManualImmediateAck;
 			this.isRecordAck = this.ackMode.equals(AckMode.RECORD);
+			this.isRecordFilteredAck = this.ackMode.equals(AckMode.RECORD_FILTERED);
 			boolean isOutOfCommit = this.isAnyManualAck && this.asyncReplies;
 			this.offsetsInThisBatch = isOutOfCommit ? new ConcurrentHashMap<>() : null;
 			this.deferredOffsets = isOutOfCommit ? new ConcurrentHashMap<>() : null;
@@ -933,8 +938,8 @@ else if (listener instanceof MessageListener) {
 			this.isConsumerAwareListener = listenerType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE)
 					|| listenerType.equals(ListenerType.CONSUMER_AWARE);
 			this.commonErrorHandler = determineCommonErrorHandler();
-			Assert.state(!this.isBatchListener || !this.isRecordAck,
-					"Cannot use AckMode.RECORD with a batch listener");
+			Assert.state(!this.isBatchListener || (!this.isRecordAck && !this.isRecordFilteredAck),
+					"Cannot use AckMode.RECORD or AckMode.RECORD_FILTERED with a batch listener");
 			if (this.containerProperties.getScheduler() != null) {
 				this.taskScheduler = this.containerProperties.getScheduler();
 				this.taskSchedulerExplicitlySet = true;
@@ -1510,7 +1515,7 @@ protected void handleAsyncFailure() {
 		}
 
 		private void doProcessCommits() {
-			if (!this.autoCommit && !this.isRecordAck) {
+			if (!this.autoCommit && !this.isRecordAck && !this.isRecordFilteredAck) {
 				try {
 					processCommits();
 				}
@@ -2260,7 +2265,7 @@ private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V>
 					}
 					getAfterRollbackProcessor().clearThreadState();
 				}
-				if (!this.autoCommit && !this.isRecordAck) {
+				if (!this.autoCommit && !this.isRecordAck && !this.isRecordFilteredAck) {
 					processCommits();
 				}
 			}
@@ -2710,7 +2715,7 @@ private void listenerInfo(final ConsumerRecord<K, V> cRecord) {
 		}
 
 		private void handleNack(final ConsumerRecords<K, V> records, final ConsumerRecord<K, V> cRecord) {
-			if (!this.autoCommit && !this.isRecordAck) {
+			if (!this.autoCommit && !this.isRecordAck && !this.isRecordFilteredAck) {
 				processCommits();
 			}
 			List<ConsumerRecord<?, ?>> list = new ArrayList<>();
@@ -3060,12 +3065,26 @@ public void checkDeser(final ConsumerRecord<K, V> cRecord, String headerName) {
 			}
 		}
 
+		private boolean isRecordFiltered(ConsumerRecord<K, V> cRecord) {
+			Object listener = KafkaMessageListenerContainer.this.getContainerProperties().getMessageListener();
+			if (listener instanceof FilteringAware<?, ?>) {
+				@SuppressWarnings("unchecked")
+				FilteringAware<K, V> filteringAware = (FilteringAware<K, V>) listener;
+				return filteringAware.wasFiltered(cRecord);
+			}
+			return false;
+		}
+
 		public void ackCurrent(final ConsumerRecord<K, V> cRecord) {
 			ackCurrent(cRecord, false);
 		}
 
 		public void ackCurrent(final ConsumerRecord<K, V> cRecord, boolean commitRecovered) {
-			if (this.isRecordAck && this.producer == null) {
+			if (this.isRecordFilteredAck && isRecordFiltered(cRecord)) {
+				return;
+			}
+
+			if ((this.isRecordAck || this.isRecordFilteredAck) && this.producer == null) {
 				Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = buildSingleCommits(cRecord);
 				this.commitLogger.log(() -> COMMITTING + offsetsToCommit);
 				commitOffsets(offsetsToCommit);

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2024-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener.adapter;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+/**
+ * An interface to indicate that a message listener adapter can report
+ * whether a record was filtered during processing.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Chaedong Im
+ * @since 4.0
+ */
+public interface FilteringAware<K, V> {
+
+	/**
+	 * Check if the most recent record processed was filtered out.
+	 * This method should be called after a record has been processed
+	 * to determine if the record was filtered and should not trigger
+	 * an offset commit in RECORD_FILTERED acknowledge mode.
+	 * @param record the record to check
+	 * @return true if the record was filtered, false if it was processed
+	 */
+	boolean wasFiltered(ConsumerRecord<K, V> record);
+
+}

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

@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.listener.adapter;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.jspecify.annotations.Nullable;
@@ -32,14 +34,30 @@
  * @param <V> the value type.
  *
  * @author Gary Russell
+ * @author Chaedong Im
  *
  */
 public class FilteringMessageListenerAdapter<K, V>
 		extends AbstractFilteringMessageListener<K, V, MessageListener<K, V>>
-		implements AcknowledgingConsumerAwareMessageListener<K, V> {
+		implements AcknowledgingConsumerAwareMessageListener<K, V>, FilteringAware<K, V> {
+
+	private static class FilterResult<K, V> {
+
+		final ConsumerRecord<K, V> record;
+
+		final boolean wasFiltered;
+
+		FilterResult(ConsumerRecord<K, V> record, boolean wasFiltered) {
+			this.record = record;
+			this.wasFiltered = wasFiltered;
+		}
+
+	}
 
 	private final boolean ackDiscarded;
 
+	private final AtomicReference<@Nullable FilterResult<K, V>> lastResult = new AtomicReference<>();
+
 	/**
 	 * Create an instance with the supplied strategy and delegate listener.
 	 * @param delegate the delegate.
@@ -68,7 +86,12 @@ public FilteringMessageListenerAdapter(MessageListener<K, V> delegate,
 	public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgment acknowledgment,
 			@Nullable Consumer<?, ?> consumer) {
 
-		if (!filter(consumerRecord)) {
+		boolean filtered = filter(consumerRecord);
+
+		// Atomically update both the record and its filtered state together
+		this.lastResult.set(new FilterResult<>(consumerRecord, filtered));
+
+		if (!filtered) {
 			switch (this.delegateType) {
 				case ACKNOWLEDGING_CONSUMER_AWARE -> this.delegate.onMessage(consumerRecord, acknowledgment, consumer);
 				case ACKNOWLEDGING -> this.delegate.onMessage(consumerRecord, acknowledgment);
@@ -93,6 +116,12 @@ private void ackFilteredIfNecessary(@Nullable Acknowledgment acknowledgment) {
 		}
 	}
 
+	@Override
+	public boolean wasFiltered(ConsumerRecord<K, V> record) {
+		FilterResult<K, V> result = this.lastResult.get();
+		return result != null && result.record == record && result.wasFiltered;
+	}
+
 	/*
 	 * Since the container uses the delegate's type to determine which method to call, we
 	 * must implement them all.