GH-3989 Consder the custom name of the reply topic name in sendAndReceive

Signed-off-by: mipo256 <[email protected]>

Author: mipo256

build.gradle

@@ -1,3 +1,6 @@
+import org.jetbrains.kotlin.gradle.dsl.JvmTarget
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompilationTask
+
 buildscript {
 	ext.kotlinVersion = '2.1.21'
 	ext.isCI = System.getenv('GITHUB_ACTION')
@@ -147,9 +150,9 @@ configure(javaProjects) { subproject ->
 		options.encoding = 'UTF-8'
 	}
 
-	tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompilationTask).configureEach {
+	tasks.withType(KotlinCompilationTask).configureEach {
 		compilerOptions {
-			jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17
+			jvmTarget = JvmTarget.JVM_17
 			javaParameters = true
 			allWarningsAsErrors = true
 		}

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

@@ -23,7 +23,6 @@
  *
  * @author Gary Russell
  * @since 2.5
- *
  */
 @FunctionalInterface
 public interface DeliveryAttemptAware {

spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

@@ -16,19 +16,6 @@
 
 package org.springframework.kafka.requestreply;
 
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.Collection;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
-
 import io.micrometer.observation.Observation;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -37,7 +24,6 @@
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.jspecify.annotations.Nullable;
-
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.SmartLifecycle;
@@ -62,6 +48,20 @@
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.Assert;
 
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
 /**
  * A KafkaTemplate that implements request/reply semantics.
  *
@@ -73,6 +73,7 @@
  * @author Artem Bilan
  * @author Borahm Lee
  * @author Francois Rosiere
+ * @author Mikhail Polivakha
  *
  * @since 2.1.3
  *
@@ -415,14 +416,11 @@ public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record) {
 	@Override
 	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, @Nullable Duration replyTimeout) {
 		Assert.state(this.running, "Template has not been started"); // NOSONAR (sync)
-		Duration timeout = replyTimeout;
-		if (timeout == null) {
-			timeout = this.defaultReplyTimeout;
-		}
+		Duration timeout = Optional.ofNullable(replyTimeout).orElse(defaultReplyTimeout);
 		CorrelationKey correlationId = this.correlationStrategy.apply(record);
 		Assert.notNull(correlationId, "the created 'correlationId' cannot be null");
 		Headers headers = record.headers();
-		boolean hasReplyTopic = headers.lastHeader(KafkaHeaders.REPLY_TOPIC) != null;
+		boolean hasReplyTopic = headers.lastHeader(this.replyTopicHeaderName) != null;
 		if (!hasReplyTopic && this.replyTopic != null) {
 			headers.add(new RecordHeader(this.replyTopicHeaderName, this.replyTopic));
 			if (this.replyPartition != null) {

spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

@@ -291,12 +291,8 @@ protected boolean doesMatchMultiValueHeader(String headerName) {
 	}
 
 	private boolean doesMatchMultiValueHeaderInternal(String headerName) {
-		for (HeaderMatcher headerMatcher : this.multiValueHeaderMatchers) {
-			if (headerMatcher.matchHeader(headerName)) {
-				return true;
-			}
-		}
-		return false;
+		return multiValueHeaderMatchers.stream()
+				.anyMatch(headerMatcher -> headerMatcher.matchHeader(headerName));
 	}
 
 	/**