fix: use ThreadLocal to isolate filter result per thread in FilteringMessageListenerAdapter

Signed-off-by: Chaedong Im <[email protected]>

Author: Chaedie

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

@@ -16,8 +16,6 @@
 
 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;
@@ -43,7 +41,8 @@ public class FilteringMessageListenerAdapter<K, V>
 
 	private final boolean ackDiscarded;
 
-	private final AtomicReference<@Nullable FilterResult<K, V>> lastResult = new AtomicReference<>();
+	@SuppressWarnings("rawtypes")
+	private static final ThreadLocal<FilterResult> LAST = new ThreadLocal<>();
 
 	/**
 	 * Create an instance with the supplied strategy and delegate listener.
@@ -74,9 +73,8 @@ public void onMessage(ConsumerRecord<K, V> consumerRecord, @Nullable Acknowledgm
 			@Nullable Consumer<?, ?> consumer) {
 
 		boolean filtered = filter(consumerRecord);
+		LAST.set(new FilterResult<>(consumerRecord, filtered));
 
-		// Atomically update both the record and its filtered state together
-		this.lastResult.set(new FilterResult<>(consumerRecord, filtered));
 
 		if (!filtered) {
 			switch (this.delegateType) {
@@ -105,8 +103,13 @@ 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;
+		@SuppressWarnings("unchecked")
+		FilterResult<K, V> result = (FilterResult<K, V>) LAST.get();
+		return result != null
+				&& result.record.topic().equals(record.topic())
+				&& result.record.partition() == record.partition()
+				&& result.record.offset() == record.offset()
+				&& result.wasFiltered;
 	}
 
 	/*

spring-kafka/src/test/java/org/springframework/kafka/listener/AckModeRecordFilteredTest.java

@@ -33,6 +33,7 @@
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
+import org.springframework.kafka.support.Acknowledgment;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -326,4 +327,77 @@ void testRecordFilteredModeDoesNotBreakNormalProcessing() throws InterruptedExce
 		assertThat(processedValues).containsExactly("value0", "value1", "value2");
 		verify(consumer, times(3)).commitSync(any(), any(Duration.class));
 	}
+
+	@SuppressWarnings({"unchecked", "deprecation"})
+	@Test
+	void recordFilteredModeShouldBeThreadIsolated() throws Exception {
+		ConsumerFactory<String, String> cf = mock(ConsumerFactory.class);
+		Consumer<String, String> c0 = mock(Consumer.class);
+		Consumer<String, String> c1 = mock(Consumer.class);
+		given(cf.createConsumer(any(), any(), any(), any())).willReturn(c0, c1);
+
+		ContainerProperties props = new ContainerProperties("iso-topic");
+		props.setGroupId("iso-group");
+		props.setAckMode(ContainerProperties.AckMode.RECORD_FILTERED);
+
+		CountDownLatch aHasSetState = new CountDownLatch(1);
+		CountDownLatch bHasProcessed = new CountDownLatch(1);
+		RecordFilterStrategy<String, String> filter = rec -> rec.offset() == 0;
+
+		FilteringMessageListenerAdapter<String, String> adapter =
+				new FilteringMessageListenerAdapter<>(
+						(MessageListener<String, String>) r -> {
+						},
+						filter
+				) {
+					@Override
+					public void onMessage(ConsumerRecord<String, String> rec,
+					                      Acknowledgment ack,
+					                      Consumer<?, ?> consumer) {
+						super.onMessage(rec, ack, consumer);
+						if (rec.offset() == 0) {
+							aHasSetState.countDown();
+							try {
+								bHasProcessed.await(500, TimeUnit.MILLISECONDS);
+							} catch (InterruptedException e) {
+								Thread.currentThread().interrupt();
+							}
+						} else if (rec.offset() == 1) {
+							try {
+								aHasSetState.await(200, TimeUnit.MILLISECONDS);
+							} catch (InterruptedException e) {
+								Thread.currentThread().interrupt();
+							}
+							bHasProcessed.countDown();
+						}
+					}
+				};
+
+		ConcurrentMessageListenerContainer<String, String> container =
+				new ConcurrentMessageListenerContainer<>(cf, props);
+		container.setConcurrency(2);
+		container.setupMessageListener(adapter);
+
+		TopicPartition tp0 = new TopicPartition("iso-topic", 0);
+		TopicPartition tp1 = new TopicPartition("iso-topic", 1);
+
+		ConsumerRecords<String, String> poll0 = new ConsumerRecords<>(Map.of(
+				tp0, List.of(new ConsumerRecord<>("iso-topic", 0, 0, "k0", "v0"))
+		));
+		ConsumerRecords<String, String> poll1 = new ConsumerRecords<>(Map.of(
+				tp1, List.of(new ConsumerRecord<>("iso-topic", 1, 1, "k1", "v1"))
+		));
+
+		given(c0.poll(any(Duration.class))).willReturn(poll0).willReturn(ConsumerRecords.empty());
+		given(c1.poll(any(Duration.class))).willReturn(poll1).willReturn(ConsumerRecords.empty());
+
+		// when: containers process records concurrently (thread-local isolation should apply)
+		container.start();
+		Thread.sleep(400);
+		container.stop();
+
+		// then: consumer c1 commits only its record, while c0 (filtered) does not
+		verify(c1, times(1)).commitSync(any(), any(Duration.class));
+		verify(c0, never()).commitSync(any(), any(Duration.class));
+	}
 }