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

mipo256 · mipo256 · commit 95c9166d8204 · 2025-07-14T18:00:02.000+03:00
Signed-off-by: mipo256 &lt;mikhailpolivakha@gmail.com&gt;
diff --git a/build.gradle b/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
 		}
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeliveryAttemptAware.java b/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 {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/CorrelationKey.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/CorrelationKey.java
@@ -35,8 +35,16 @@ public final class CorrelationKey {
 
 	private final byte[] correlationId;
 
+	/**
+	 * Cached hex representation of the {@link #correlationId}.
+	 * TODO: Migrate to stable values JEP 502
+	 */
 	private @Nullable String asString;
 
+	/**
+	 * Cached hash code.
+	 * TODO: Migrate to stable values JEP 502
+	 */
 	private volatile @Nullable Integer hashCode;
 
 	public CorrelationKey(byte[] correlationId) { // NOSONAR array reference
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java
@@ -22,6 +22,7 @@
 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;
@@ -73,6 +74,7 @@
  * @author Artem Bilan
  * @author Borahm Lee
  * @author Francois Rosiere
+ * @author Mikhail Polivakha
  *
  * @since 2.1.3
  *
@@ -415,19 +417,17 @@ 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(this.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;
-		if (!hasReplyTopic && this.replyTopic != null) {
+		boolean replyTopicIsUnset = headers.lastHeader(this.replyTopicHeaderName) == null;
+		if (replyTopicIsUnset && this.replyTopic != null) {
 			headers.add(new RecordHeader(this.replyTopicHeaderName, this.replyTopic));
-			if (this.replyPartition != null) {
-				headers.add(new RecordHeader(this.replyPartitionHeaderName, this.replyPartition));
-			}
+		}
+		boolean replyPartitionIsUnset = headers.lastHeader(this.replyPartitionHeaderName) != null;
+		if (replyPartitionIsUnset && this.replyPartition != null) {
+			headers.add(new RecordHeader(this.replyPartitionHeaderName, this.replyPartition));
 		}
 		Object correlation = this.binaryCorrelation ? correlationId : correlationId.toString();
 		byte[] correlationValue = this.binaryCorrelation
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/requestreply/RequestReplyFuture.java b/spring-kafka/src/main/java/org/springframework/kafka/requestreply/RequestReplyFuture.java
@@ -26,7 +26,7 @@
 /**
  * A {@link CompletableFuture} for requests/replies.
  *
- * @param <K> the key type.
+ * @param <K> the key type of inbound kafka message.
  * @param <V> the outbound data type.
  * @param <R> the reply data type.
  *
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java
@@ -26,6 +26,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -47,9 +48,9 @@
  * @author Artem Bilan
  * @author Sanghyeok An
  * @author Soby Chacko
+ * @author Mikhail Polivakha
  *
  * @since 2.1.3
- *
  */
 public abstract class AbstractKafkaHeaderMapper implements KafkaHeaderMapper {
 
@@ -269,11 +270,7 @@ private boolean doesMatchInternal(String header) {
 	 * @since 2.2.5
 	 */
 	protected Object headerValueToAddOut(String key, Object value) {
-		Object valueToAdd = mapRawOut(key, value);
-		if (valueToAdd == null) {
-			valueToAdd = value;
-		}
-		return valueToAdd;
+		return Optional.ofNullable((Object) mapRawOut(key, value)).orElse(value);
 	}
 
 	/**
@@ -291,12 +288,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 this.multiValueHeaderMatchers.stream()
+				.anyMatch(headerMatcher -> headerMatcher.matchHeader(headerName));
 	}
 
 	/**
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java
@@ -33,6 +33,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import com.google.common.collect.Iterables;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -102,12 +103,14 @@
  * @author Gary Russell
  * @author Nathan Xu
  * @author Soby Chacko
+ * @author Mikhail Polivakha
  * @since 2.1.3
  *
  */
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(partitions = 5, topics = { ReplyingKafkaTemplateTests.A_REPLY, ReplyingKafkaTemplateTests.A_REQUEST,
+@EmbeddedKafka(partitions = 5, topics = {
+		ReplyingKafkaTemplateTests.A_REPLY, ReplyingKafkaTemplateTests.A_REQUEST,
 		ReplyingKafkaTemplateTests.B_REPLY, ReplyingKafkaTemplateTests.B_REQUEST,
 		ReplyingKafkaTemplateTests.C_REPLY, ReplyingKafkaTemplateTests.C_REQUEST,
 		ReplyingKafkaTemplateTests.D_REPLY, ReplyingKafkaTemplateTests.D_REQUEST,
@@ -119,7 +122,10 @@
 		ReplyingKafkaTemplateTests.J_REPLY, ReplyingKafkaTemplateTests.J_REQUEST,
 		ReplyingKafkaTemplateTests.K_REPLY, ReplyingKafkaTemplateTests.K_REQUEST,
 		ReplyingKafkaTemplateTests.L_REPLY, ReplyingKafkaTemplateTests.L_REQUEST,
-		ReplyingKafkaTemplateTests.M_REPLY, ReplyingKafkaTemplateTests.M_REQUEST })
+		ReplyingKafkaTemplateTests.M_REPLY, ReplyingKafkaTemplateTests.M_REQUEST,
+		ReplyingKafkaTemplateTests.CUSTOM_REPLY_HEADER_REPLY, ReplyingKafkaTemplateTests.CUSTOM_REPLY_HEADER_REQUEST,
+		ReplyingKafkaTemplateTests.CUSTOM_REPLY_HEADER_WITH_PARTITION_REPLY, ReplyingKafkaTemplateTests.CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST,
+})
 public class ReplyingKafkaTemplateTests {
 
 	public static final String A_REPLY = "aReply";
@@ -174,6 +180,18 @@ public class ReplyingKafkaTemplateTests {
 
 	public static final String M_REQUEST = "mRequest";
 
+	// GH-3989
+	public static final String CUSTOM_REPLY_HEADER_REPLY = "CUSTOM_REPLY_HEADER_REPLY";
+
+	// GH-3989
+	public static final String CUSTOM_REPLY_HEADER_REQUEST = "CUSTOM_REPLY_HEADER_REQUEST";
+
+	// GH-3989
+	public static final String CUSTOM_REPLY_HEADER_WITH_PARTITION_REPLY = "CUSTOM_REPLY_HEADER_WITH_PARTITION_REPLY";
+
+	// GH-3989
+	public static final String CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST = "CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST";
+
 	@Autowired
 	private EmbeddedKafkaBroker embeddedKafka;
 
@@ -365,6 +383,54 @@ public void testHandlerReturn() throws Exception {
 		}
 	}
 
+	@Test
+	public void testCustomReplyTopicHeaderIsNotDuplicated() throws Exception {
+		String customReplyHeaderName = "X-Custom-Reply-Header";
+		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(CUSTOM_REPLY_HEADER_REPLY);
+		template.setReplyTopicHeaderName(customReplyHeaderName);
+		try {
+			Message<String> message = MessageBuilder.withPayload("foo")
+					.setHeader(customReplyHeaderName, CUSTOM_REPLY_HEADER_REPLY)
+					.setHeader(KafkaHeaders.TOPIC, CUSTOM_REPLY_HEADER_REQUEST)
+					.build();
+
+			RequestReplyMessageFuture<Integer, String> future = template.sendAndReceive(message, Duration.ofSeconds(30));
+			future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
+			Message<?> resultingMessage = future.get(30, TimeUnit.SECONDS);
+			assertThat(resultingMessage.getPayload()).isEqualTo("OK");
+		}
+		finally {
+			template.stop();
+			template.destroy();
+		}
+	}
+
+	@Test
+	public void testCustomReplyHeadersAreNotDuplicated() throws Exception {
+		String customReplyTopicHeaderName = "X-Custom-Reply-Header";
+		String customReplyPartitionHeaderName = "X-Custom-Reply-Partition";
+		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(CUSTOM_REPLY_HEADER_WITH_PARTITION_REPLY);
+		template.setReplyTopicHeaderName(customReplyTopicHeaderName);
+		template.setReplyPartitionHeaderName(customReplyPartitionHeaderName);
+
+		try {
+			Message<String> message = MessageBuilder.withPayload("foo")
+					.setHeader(customReplyTopicHeaderName, CUSTOM_REPLY_HEADER_REPLY)
+					.setHeader(customReplyPartitionHeaderName, "test-partition")
+					.setHeader(KafkaHeaders.TOPIC, CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST)
+					.build();
+
+			RequestReplyMessageFuture<Integer, String> future = template.sendAndReceive(message, Duration.ofSeconds(30));
+			future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
+			Message<?> resultingMessage = future.get(30, TimeUnit.SECONDS);
+			assertThat(resultingMessage.getPayload()).isEqualTo("OK");
+		}
+		finally {
+			template.stop();
+			template.destroy();
+		}
+	}
+
 	@Test
 	public void testMessageReturnNoHeadersProvidedByListener() throws Exception {
 		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(H_REPLY);
@@ -958,6 +1024,48 @@ public String handleA(String in) throws InterruptedException {
 			return in.toUpperCase();
 		}
 
+		// GH-3989
+		@KafkaListener(id = CUSTOM_REPLY_HEADER_REQUEST, topics = CUSTOM_REPLY_HEADER_REQUEST)
+		@SendTo(value = CUSTOM_REPLY_HEADER_REPLY)  // send to custom topic back
+		public String handleCustomReplyHeaderNoReplyPartition(ConsumerRecord<?, String> inputMessage) {
+			Headers headers = inputMessage.headers();
+
+			if (Iterables.size(headers.headers("X-Custom-Reply-Header")) != 1) {
+				return "The X-Custom-Reply-Header header that signify the custom reply topic header name is duplicated. It is supposed to present only once";
+			}
+
+			if (Iterables.size(headers.headers(KafkaHeaders.REPLY_PARTITION)) != 0) {
+				return "It is expected that the user does NOT specify the reply partition in this test case";
+			}
+
+			if (!"foo".equals(inputMessage.value())) {
+				return "Expected message is 'foo', but got %s".formatted(inputMessage.value());
+			}
+
+			return "OK";
+		}
+
+		// GH-3989
+		@KafkaListener(id = CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST, topics = CUSTOM_REPLY_HEADER_WITH_PARTITION_REQUEST)
+		@SendTo(value = CUSTOM_REPLY_HEADER_WITH_PARTITION_REPLY)  // send to custom topic back
+		public String handleCustomReplyHeaderDefaultPartitionHeader(ConsumerRecord<?, String> inputMessage) {
+			Headers headers = inputMessage.headers();
+
+			if (Iterables.size(headers.headers("X-Custom-Reply-Header")) != 1) {
+				return "The X-Custom-Reply-Header header that signify the custom reply topic header name is duplicated. It is supposed to present only once";
+			}
+
+			if (Iterables.size(headers.headers("X-Custom-Reply-Partition")) != 1) {
+				return "Executed a single reply partition header '%s' in the incoming message".formatted(KafkaHeaders.REPLY_PARTITION);
+			}
+
+			if (!"foo".equals(inputMessage.value())) {
+				return "Expected message is 'foo', but got %s".formatted(inputMessage.value());
+			}
+
+			return "OK";
+		}
+
 		@KafkaListener(topics = B_REQUEST, containerFactory = "simpleMapperFactory")
 		@SendTo  // default REPLY_TOPIC header
 		public String handleB(String in) {

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,6 @@`
`23`	`23`	`*`
`24`	`24`	`* @author Gary Russell`
`25`	`25`	`* @since 2.5`
`26`		`- *`
`27`	`26`	`*/`
`28`	`27`	`@FunctionalInterface`
`29`	`28`	`public interface DeliveryAttemptAware {`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@`
`26`	`26`	`/**`
`27`	`27`	`* A {@link CompletableFuture} for requests/replies.`
`28`	`28`	`*`
`29`		`- * @param <K> the key type.`
	`29`	`+ * @param <K> the key type of inbound kafka message.`
`30`	`30`	`* @param <V> the outbound data type.`
`31`	`31`	`* @param <R> the reply data type.`
`32`	`32`	`*`