Merge branch 'main' into codex/fix-issue-with-kafka-header-extraction

Signed-off-by: Igor Macedo Quintanilha <igor.quintanilha@gmail.com>

Author: igormq

build.gradle

@@ -17,7 +17,7 @@ buildscript {
 plugins {
 	id 'base'
 	id 'idea'
-	id 'org.ajoberstar.grgit' version '5.3.0'
+	id 'org.ajoberstar.grgit' version '5.3.2'
 	id 'io.spring.nohttp' version '0.0.11'
 	id 'io.spring.dependency-management' version '1.1.7' apply false
 	id 'io.freefair.aggregate-javadoc' version '8.11'
@@ -55,16 +55,16 @@ ext {
 	awaitilityVersion = '4.3.0'
 	hamcrestVersion = '3.0'
 	hibernateValidationVersion = '8.0.2.Final'
-	jacksonBomVersion = '2.19.0'
+	jacksonBomVersion = '2.19.1'
 	jaywayJsonPathVersion = '2.9.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.13.1'
 	kafkaVersion = '4.0.0'
 	kotlinCoroutinesVersion = '1.10.2'
 	log4jVersion = '2.24.3'
 	micrometerDocsVersion = '1.0.4'
-	micrometerVersion = '1.15.0'
-	micrometerTracingVersion = '1.5.0'
+	micrometerVersion = '1.15.1'
+	micrometerTracingVersion = '1.5.1'
 	mockitoVersion = '5.18.0'
 	reactorVersion = '2025.0.0-SNAPSHOT'
 	scalaVersion = '2.13'

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/message-listener-container.adoc

@@ -22,6 +22,35 @@ IMPORTANT: If the interceptor mutates the record (by creating a new one), the `t
 
 The `CompositeRecordInterceptor` and `CompositeBatchInterceptor` can be used to invoke multiple interceptors.
 
+Starting with version 4.0, `AbstractMessageListenerContainer` exposes `getRecordInterceptor()` as a public method.
+If the returned interceptor is an instance of `CompositeRecordInterceptor`, additional `RecordInterceptor` instances can be added to it even after the container instance extending `AbstractMessageListenerContainer` has been created and a `RecordInterceptor` has already been configured.
+The following example shows how to do so:
+
+[source, java]
+----
+public void configureRecordInterceptor(KafkaMessageListenerContainer<Integer, String> container) {
+    CompositeRecordInterceptor compositeInterceptor;
+
+    RecordInterceptor<Integer, String> previousInterceptor = container.getRecordInterceptor();
+    if (previousInterceptor instanceof CompositeRecordInterceptor interceptor) {
+        compositeInterceptor = interceptor;
+    } else {
+        compositeInterceptor = new CompositeRecordInterceptor<>();
+        container.setRecordInterceptor(compositeInterceptor);
+    }
+
+    if (previousInterceptor != null) {
+        compositeRecordInterceptor.addRecordInterceptor(previousInterceptor);
+    }
+
+    RecordInterceptor<Integer, String> recordInterceptor1 = new RecordInterceptor() {...};
+    RecordInterceptor<Integer, String> recordInterceptor2 = new RecordInterceptor() {...};
+
+    compositeInterceptor.addRecordInterceptor(recordInterceptor1);
+    compositeInterceptor.addRecordInterceptor(recordInterceptor2);
+}
+----
+
 By default, starting with version 2.8, when using transactions, the interceptor is invoked before the transaction has started.
 You can set the listener container's `interceptBeforeTx` property to `false` to invoke the interceptor after the transaction has started instead.
 Starting with version 2.9, this will apply to any transaction manager, not just `KafkaAwareTransactionManager`+++s+++.
@@ -265,4 +294,3 @@ The listener containers implement `SmartLifecycle`, and `autoStartup` is `true`
 The containers are started in a late phase (`Integer.MAX-VALUE - 100`).
 Other components that implement `SmartLifecycle`, to handle data from listeners, should be started in an earlier phase.
 The `- 100` leaves room for later phases to enable components to be auto-started after the containers.
-

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

@@ -77,8 +77,16 @@ For details, see xref:kafka/receiving-messages/rebalance-listeners.adoc#new-reba
 The `DefaultKafkaHeaderMapper` and `SimpleKafkaHeaderMapper` support multi-value header mapping for Kafka records.
 More details are available in xref:kafka/headers.adoc#multi-value-header[Support multi-value header mapping].
 
+<<<<<<< codex/fix-issue-with-kafka-header-extraction
 [[x40-batch-observability]]
 === Per-Record Observation in Batch Listeners
 
 It is now possible to get an observation for each record when using a batch listener.
 See xref:kafka/micrometer.adoc#batch-listener-obs[Observability for Batch Listeners] for more information.
+=======
+[[x40-add-record-interceptor]]
+=== Configure additional `RecordInterceptor`
+
+Listener containers now support interceptor customization via `getRecordInterceptor()`.
+See the xref:kafka/receiving-messages/message-listener-container.adoc#message-listener-container[Message Listener Containers] section for details.
+>>>>>>> main

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

@@ -460,7 +460,12 @@ public void setKafkaAdmin(KafkaAdmin kafkaAdmin) {
 		this.kafkaAdmin = kafkaAdmin;
 	}
 
-	protected @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
+	/**
+	 * Get the {@link RecordInterceptor} for modification, if configured.
+	 * @return the {@link RecordInterceptor}, or {@code null} if not configured
+	 * @since 4.0
+	 */
+	public @Nullable RecordInterceptor<K, V> getRecordInterceptor() {
 		return this.recordInterceptor;
 	}
 

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

@@ -35,6 +35,7 @@
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Sanghyeok An
  * @since 2.3
  *
  */
@@ -92,4 +93,13 @@ public void afterRecord(ConsumerRecord<K, V> record, Consumer<K, V> consumer) {
 		this.delegates.forEach(del -> del.afterRecord(record, consumer));
 	}
 
+	/**
+	 * Add an {@link RecordInterceptor} to delegates.
+	 * @param recordInterceptor the interceptor.
+	 * @since 4.0
+	 */
+	public void addRecordInterceptor(RecordInterceptor<K, V> recordInterceptor) {
+		this.delegates.add(recordInterceptor);
+	}
+
 }

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

@@ -1498,7 +1498,7 @@ protected void handleAsyncFailure() {
 			// We will give up on retrying with the remaining copied and failed Records.
 			for (FailedRecordTuple<K, V> copyFailedRecord : copyFailedRecords) {
 				try {
-					invokeErrorHandlerBySingleRecord(copyFailedRecord);
+					copyFailedRecord.observation.scoped(() -> invokeErrorHandlerBySingleRecord(copyFailedRecord));
 				}
 				catch (Exception e) {
 					this.logger.warn(() ->
@@ -3453,8 +3453,13 @@ private Collection<ConsumerRecord<K, V>> getHighestOffsetRecords(ConsumerRecords
 					.values();
 		}
 
+		private Observation getCurrentObservation() {
+			Observation currentObservation = this.observationRegistry.getCurrentObservation();
+			return currentObservation == null ? Observation.NOOP : currentObservation;
+		}
+
 		private void callbackForAsyncFailure(ConsumerRecord<K, V> cRecord, RuntimeException ex) {
-			this.failedRecords.addLast(new FailedRecordTuple<>(cRecord, ex));
+			this.failedRecords.addLast(new FailedRecordTuple<>(cRecord, ex, getCurrentObservation()));
 		}
 
 		@Override
@@ -4071,6 +4076,6 @@ private static class StopAfterFenceException extends KafkaException {
 
 	}
 
-	private record FailedRecordTuple<K, V>(ConsumerRecord<K, V> record, RuntimeException ex) { }
+	private record FailedRecordTuple<K, V>(ConsumerRecord<K, V> record, RuntimeException ex, Observation observation) { }
 
 }

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

@@ -549,7 +549,7 @@ else if (!(result instanceof CompletableFuture<?>)) {
 		}
 
 		completableFutureResult.whenComplete((r, t) -> {
-			try {
+			try (var scope = observation.openScope()) {
 				if (t == null) {
 					asyncSuccess(r, replyTopic, source, messageReturnType);
 					if (isAsyncReplies()) {
@@ -736,13 +736,15 @@ protected void asyncFailure(Object request, @Nullable Acknowledgment acknowledgm
 									"Async Fail", Objects.requireNonNull(source).getPayload()), cause));
 		}
 		catch (Throwable ex) {
-			this.logger.error(t, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
 			acknowledge(acknowledgment);
 			if (canAsyncRetry(request, ex) && this.asyncRetryCallback != null) {
 				@SuppressWarnings("unchecked")
 				ConsumerRecord<K, V> record = (ConsumerRecord<K, V>) request;
 				this.asyncRetryCallback.accept(record, (RuntimeException) ex);
 			}
+			else {
+				this.logger.error(ex, () -> "Future, Mono, or suspend function was completed with an exception for " + source);
+			}
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DelegatingByTypeSerializer.java

@@ -16,9 +16,10 @@
 
 package org.springframework.kafka.support.serializer;
 
-import java.util.LinkedHashMap;
+import java.util.Comparator;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.TreeMap;
 
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.header.Headers;
@@ -32,13 +33,29 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Wang Zhiyang
+ * @author Mahesh Aravind V
  *
  * @since 2.7.9
  *
  */
 public class DelegatingByTypeSerializer implements Serializer<Object> {
 
-	private final Map<Class<?>, Serializer<?>> delegates = new LinkedHashMap<>();
+	private static final Comparator<Class<?>> DELEGATES_ASSIGNABILITY_COMPARATOR =
+			(class1, class2) -> {
+
+				if (class1 == class2) {
+					return 0; // Classes are the same
+				}
+				if (class1.isAssignableFrom(class2)) {
+					return 1; // class2 is a superclass or superinterface of class1, so class2 should come first
+				}
+				if (class2.isAssignableFrom(class1)) {
+					return -1; // class1 is a superclass or superinterface of class2, so class1 should come first
+				}
+				return class1.getName().compareTo(class2.getName()); // If no inheritance relation, compare by name
+			};
+
+	private final Map<Class<?>, Serializer<?>> delegates = new TreeMap<>(DELEGATES_ASSIGNABILITY_COMPARATOR);
 
 	private final boolean assignable;
 
@@ -51,17 +68,21 @@ public DelegatingByTypeSerializer(Map<Class<?>, Serializer<?>> delegates) {
 	}
 
 	/**
-	 * Construct an instance with the map of delegates; keys matched exactly or if the
-	 * target object is assignable to the key, depending on the assignable argument.
-	 * If assignable, entries are checked in the natural entry order so an ordered map
-	 * such as a {@link LinkedHashMap} is recommended.
-	 * @param delegates the delegates.
-	 * @param assignable whether the target is assignable to the key.
+	 * Construct an instance with the map of delegates.
+	 * If {@code assignable} is {@code false}, only exact key matches are considered.
+	 * If {@code assignable} is {@code true}, a delegate is selected if its key class
+	 * is assignable from the target object's class. When multiple matches are possible,
+	 * the most specific matching class is selected — that is, the closest match in the
+	 * class hierarchy.
+	 * @param delegates the delegates
+	 * @param assignable true if {@link #findDelegate(Object, Map)} should consider assignability to
+	 * the key rather than an exact match.
 	 * @since 2.8.3
 	 */
 	public DelegatingByTypeSerializer(Map<Class<?>, Serializer<?>> delegates, boolean assignable) {
 		Assert.notNull(delegates, "'delegates' cannot be null");
 		Assert.noNullElements(delegates.values(), "Serializers in delegates map cannot be null");
+
 		this.delegates.putAll(delegates);
 		this.assignable = assignable;
 	}
@@ -101,7 +122,7 @@ public byte[] serialize(String topic, Headers headers, Object data) {
 		return delegate.serialize(topic, headers, data);
 	}
 
-	private  <T> Serializer<T> findDelegate(T data) {
+	protected  <T> Serializer<T> findDelegate(T data) {
 		return findDelegate(data, this.delegates);
 	}
 

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

@@ -3842,7 +3842,7 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		containerProps.setClientId("clientId");
 
 		CountDownLatch afterLatch = new CountDownLatch(1);
-		RecordInterceptor<Integer, String> recordInterceptor = spy(new RecordInterceptor<Integer, String>() {
+		RecordInterceptor<Integer, String> recordInterceptor1 = spy(new RecordInterceptor<Integer, String>() {
 
 			@Override
 			public @NonNull ConsumerRecord<Integer, String> intercept(ConsumerRecord<Integer, String> record,
@@ -3858,25 +3858,54 @@ public void clearThreadState(Consumer<?, ?> consumer) {
 
 		});
 
+		RecordInterceptor<Integer, String> recordInterceptor2 = spy(new RecordInterceptor<Integer, String>() {
+
+			@Override
+			public @NonNull ConsumerRecord<Integer, String> intercept(ConsumerRecord<Integer, String> record,
+																	Consumer<Integer, String> consumer) {
+
+				return record;
+			}
+
+			@Override
+			public void clearThreadState(Consumer<?, ?> consumer) {
+				afterLatch.countDown();
+			}
+
+		});
+
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
-		container.setRecordInterceptor(recordInterceptor);
+		container.setRecordInterceptor(new CompositeRecordInterceptor<>());
+		if (container.getRecordInterceptor() instanceof CompositeRecordInterceptor<Integer, String> composite) {
+			composite.addRecordInterceptor(recordInterceptor1);
+			composite.addRecordInterceptor(recordInterceptor2);
+		}
+
 		container.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(afterLatch.await(10, TimeUnit.SECONDS)).isTrue();
 
-		InOrder inOrder = inOrder(recordInterceptor, messageListener, consumer);
-		inOrder.verify(recordInterceptor).setupThreadState(eq(consumer));
+		InOrder inOrder = inOrder(recordInterceptor1, recordInterceptor2, messageListener, consumer);
+		inOrder.verify(recordInterceptor1).setupThreadState(eq(consumer));
+		inOrder.verify(recordInterceptor2).setupThreadState(eq(consumer));
 		inOrder.verify(consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
-		inOrder.verify(recordInterceptor).intercept(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).intercept(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).intercept(eq(firstRecord), eq(consumer));
 		inOrder.verify(messageListener).onMessage(eq(firstRecord));
-		inOrder.verify(recordInterceptor).success(eq(firstRecord), eq(consumer));
-		inOrder.verify(recordInterceptor).afterRecord(eq(firstRecord), eq(consumer));
-		inOrder.verify(recordInterceptor).intercept(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).success(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).success(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).afterRecord(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).afterRecord(eq(firstRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).intercept(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).intercept(eq(secondRecord), eq(consumer));
 		inOrder.verify(messageListener).onMessage(eq(secondRecord));
-		inOrder.verify(recordInterceptor).success(eq(secondRecord), eq(consumer));
-		inOrder.verify(recordInterceptor).afterRecord(eq(secondRecord), eq(consumer));
-		inOrder.verify(recordInterceptor).clearThreadState(eq(consumer));
+		inOrder.verify(recordInterceptor1).success(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).success(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).afterRecord(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor2).afterRecord(eq(secondRecord), eq(consumer));
+		inOrder.verify(recordInterceptor1).clearThreadState(eq(consumer));
+		inOrder.verify(recordInterceptor2).clearThreadState(eq(consumer));
 		container.stop();
 	}