fix: implement requester-side deduplication of reply messages by replyIndex

Author: helios57

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## 5.3.5 - 2026-02-05
+### Fixed
+- Requester now deduplicates reply messages by `replyIndex` to prevent corrupt results when replies are duplicated (e.g. after broker disconnect/reconnect during in-place Solace broker updates).
+- Dedup bookkeeping is cleaned up when a request completes, errors, or times out.
+
 ## 5.3.4 - 2026-01-23
 ### Changed
 - Add support to not respond to requests

README.md

@@ -599,6 +599,23 @@ More precisely:
 - If a service dies, any relations are forgotten and replies can no longer be related to request,
   potentially resulting in message loss.
 
+### Reply duplication & requester-side deduplication
+
+In some operational scenarios (e.g. Solace in-place broker updates / short disconnects / reconnects) the same request can be sent twice and the replier may therefore produce **duplicate replies**.
+
+To make request-reply robust against such duplicate delivery, the requester keeps per-request bookkeeping and **deduplicates incoming reply messages by `replyIndex`**:
+
+- If the requester receives multiple messages with the same `replyIndex`, only the first one is processed; later duplicates are ignored.
+- This also supports range indices such as `replyIndex="0-45"` (used when replies are grouped into an SDTStream). The entire grouped message will be consumed only once.
+- Terminal messages (finish/error) are always processed, even if they share a `replyIndex` with another message.
+
+#### Memory / cleanup
+
+The dedup bookkeeping is scoped to a single in-flight request (correlationId).
+It is cleared automatically when the request completes successfully, completes with a remote error, or times out.
+
+> Note: For best results, repliers should always set `totalReplies` and `replyIndex` consistently on every reply message (this is done automatically when using `RequestReplyMessageHeaderSupportService.wrap*`).
+
 ## External Links
 - [Spring Cloud Stream Solace Samples](https://solace.com/samples/solace-samples-spring/spring-cloud-stream/)
 

src/main/java/community/solace/spring/cloud/requestreply/service/RequestReplyServiceImpl.java

@@ -453,6 +453,13 @@ void onReplyReceived(final Message<?> message) {
         if (handler == null) {
             requestReplyLogger.log(LOG, Level.INFO, "Received unexpected message or maybe too late response: {}", message);
         } else {
+            String replyIndex = messageHeaderSupportService.getReplyIndex(message);
+            // Allow terminal messages (EMPTY_RESPONSE) to share replyIndex with a previous message,
+            // because they don't carry a unique index but are required to complete/error the request.
+            if (StringUtils.hasText(replyIndex) && (totalReplies == null || totalReplies != EMPTY_RESPONSE) && handler.checkDuplicate(replyIndex)) {
+                return;
+            }
+
             if (totalReplies != null) {
                 if (totalReplies == UNKNOWN_SIZE) {
                     handler.setUnknownReplies();
@@ -493,18 +500,28 @@ private static boolean isMultiResponse(Message<?> message) {
     private static List<Message<?>> parseMultiResponse(Message<SDTStream> message) {
         try {
             List<Message<?>> msgs = new ArrayList<>();
+
+            // When creating grouped (SDTStream) responses, we temporarily move the content-type to
+            // SpringHeaderParser.GROUPED_CONTENT_TYPE to prevent Spring from re-encoding the outer message.
+            // When unpacking, restore it so downstream conversion (e.g. JSON -> POJO) works as expected.
+            Map<String, Object> baseHeaders = new IntegrationMessageHeaderAccessor(message).toMap();
+            Object groupedContentType = baseHeaders.get(SpringHeaderParser.GROUPED_CONTENT_TYPE);
+            if (groupedContentType != null && baseHeaders.get(MessageHeaders.CONTENT_TYPE) == null) {
+                baseHeaders.put(MessageHeaders.CONTENT_TYPE, groupedContentType);
+            }
+
             while (message.getPayload().hasRemaining()) {
                 switch (message.getPayload().readString()) {
                     case "BytesMessage" -> msgs.add(
                             MessageBuilder
                                     .withPayload(message.getPayload().readBytes())
-                                    .copyHeaders(new IntegrationMessageHeaderAccessor(message).toMap())
+                                    .copyHeaders(baseHeaders)
                                     .build()
                     );
                     case "TextMessage", "XMLContentMessage" -> msgs.add(
                             MessageBuilder
                                     .withPayload(new String(message.getPayload().readBytes(), StandardCharsets.UTF_8))
-                                    .copyHeaders(new IntegrationMessageHeaderAccessor(message).toMap())
+                                    .copyHeaders(baseHeaders)
                                     .build()
                     );
                     case "StreamMessage", "MapMessage" -> throw new IllegalArgumentException(

src/main/java/community/solace/spring/cloud/requestreply/service/ResponseHandler.java

@@ -11,6 +11,8 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -31,6 +33,7 @@ public class ResponseHandler {
     private String errorMessage;
 
     private final RequestReplyLogger requestReplyLogger;
+    private final Set<String> receivedIndices = ConcurrentHashMap.newKeySet();
 
     public ResponseHandler(Consumer<Message<?>> responseMessageConsumer, boolean supportMultipleResponses, Timer timer, RequestReplyLogger requestReplyLogger) {
         this.countDownLatch = new CountDownLatch(1);
@@ -54,6 +57,18 @@ public void receive(Message<?> message) {
         requestReplyLogger.logReply(LOG, Level.DEBUG, "received response(remaining={}) {}", remainingReplies, message);
     }
 
+    public void receive(Message<?> message, String replyIndex) {
+        receive(message);
+    }
+
+    public boolean checkDuplicate(String replyIndex) {
+        if (replyIndex != null && !receivedIndices.add(replyIndex)) {
+            requestReplyLogger.log(LOG, Level.WARN, "received duplicate response(index={})", replyIndex);
+            return true;
+        }
+        return false;
+    }
+
     public void await() throws RemoteErrorException, InterruptedException {
         countDownLatch.await();
         if (StringUtils.hasText(errorMessage)) {
@@ -88,6 +103,10 @@ public void errorResponse(String errorMessage) {
     }
 
     private void finished() {
+        // Clear per-request dedup bookkeeping to avoid retaining replyIndex values
+        // longer than necessary (success, error, or timeout/abort paths all call finished()).
+        receivedIndices.clear();
+
         if (timer != null) {
             timer.record(Duration.between(requestTime, Instant.now()));
         }

src/main/java/community/solace/spring/cloud/requestreply/service/header/RequestReplyMessageHeaderSupportService.java

@@ -18,6 +18,7 @@
 import community.solace.spring.cloud.requestreply.service.header.parser.errormessage.MessageErrorMessageParser;
 import community.solace.spring.cloud.requestreply.service.header.parser.replyto.MessageReplyToParser;
 import community.solace.spring.cloud.requestreply.service.header.parser.totalreplies.MessageTotalRepliesParser;
+import community.solace.spring.cloud.requestreply.service.header.parser.replyindex.MessageReplyIndexParser;
 import community.solace.spring.cloud.requestreply.service.messageinterceptor.ReplyWrappingInterceptor;
 import community.solace.spring.cloud.requestreply.util.MessageChunker;
 import org.apache.commons.lang3.tuple.Pair;
@@ -57,6 +58,9 @@ public class RequestReplyMessageHeaderSupportService {
     @Autowired
     private List<MessageTotalRepliesParser> totalRepliesParsers;
 
+    @Autowired
+    private List<MessageReplyIndexParser> replyIndexParsers;
+
     @Autowired
     private List<MessageErrorMessageParser> errorMessageParsers;
 
@@ -113,6 +117,17 @@ Long getTotalReplies(Message<?> message) {
                        .orElse(null);
     }
 
+    public @Nullable
+    String getReplyIndex(Message<?> message) {
+        return message == null ? null
+                               : replyIndexParsers
+                       .stream()
+                       .map(p -> p.getReplyIndex(message))
+                       .filter(Objects::nonNull)
+                       .findFirst()
+                       .orElse(null);
+    }
+
     public @Nullable
     String getErrorMessage(Message<?> message) {
         return message == null ? null

src/main/java/community/solace/spring/cloud/requestreply/service/header/parser/SpringHeaderParser.java

@@ -1,6 +1,7 @@
 package community.solace.spring.cloud.requestreply.service.header.parser;
 
 import community.solace.spring.cloud.requestreply.service.header.parser.errormessage.MessageHeaderErrorMessageParser;
+import community.solace.spring.cloud.requestreply.service.header.parser.replyindex.MessageHeaderReplyIndexParser;
 import community.solace.spring.cloud.requestreply.service.header.parser.replyto.MessageHeaderReplyToParser;
 import community.solace.spring.cloud.requestreply.service.header.parser.totalreplies.MessageHeaderTotalRepliesParser;
 import org.springframework.core.annotation.Order;
@@ -9,7 +10,7 @@
 
 @Service
 @Order(40000)
-public class SpringHeaderParser implements MessageHeaderReplyToParser, MessageHeaderTotalRepliesParser, MessageHeaderErrorMessageParser {
+public class SpringHeaderParser implements MessageHeaderReplyToParser, MessageHeaderTotalRepliesParser, MessageHeaderErrorMessageParser, MessageHeaderReplyIndexParser {
     public final static String MULTI_TOTAL_REPLIES = "totalReplies";
     public final static String MULTI_REPLY_INDEX = "replyIndex";
     public final static String GROUPED_MESSAGES = "groupedMessages";
@@ -49,4 +50,13 @@ public String getErrorMessage(MessageHeaders headers) {
 
         return null;
     }
+
+    @Override
+    public String getReplyIndex(MessageHeaders headers) {
+        Object index = headers.get(MULTI_REPLY_INDEX);
+        if (index == null) {
+            return null;
+        }
+        return index.toString();
+    }
 }

src/main/java/community/solace/spring/cloud/requestreply/service/header/parser/replyindex/MessageHeaderReplyIndexParser.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright © Schweizerische Bundesbahnen SBB, 2026.
+ */
+
+package community.solace.spring.cloud.requestreply.service.header.parser.replyindex;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+
+@FunctionalInterface
+public interface MessageHeaderReplyIndexParser extends MessageReplyIndexParser {
+    @Override
+    default String getReplyIndex(Message<?> message) {
+        return message == null ?
+                null :
+                getReplyIndex(message.getHeaders());
+    }
+
+    @Nullable
+    String getReplyIndex(MessageHeaders headers);
+}
+

src/main/java/community/solace/spring/cloud/requestreply/service/header/parser/replyindex/MessageReplyIndexParser.java

@@ -0,0 +1,15 @@
+/*
+ * Copyright © Schweizerische Bundesbahnen SBB, 2026.
+ */
+
+package community.solace.spring.cloud.requestreply.service.header.parser.replyindex;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+
+@FunctionalInterface
+public interface MessageReplyIndexParser {
+    @Nullable
+    String getReplyIndex(Message<?> message);
+}
+

src/test/java/community/solace/spring/cloud/requestreply/AbstractRequestReplyLoggingIT.java

@@ -44,9 +44,9 @@ public abstract class AbstractRequestReplyLoggingIT {
     public static final String PROFILE_TEST_LOGGING = "testLogging";
     public static final String PROFILE_LOCAL_APP = "localApp";
 
-    public final Timestamp Ten_oClock = new Timestamp(1682928000); // 2023-05-01 10:00:00
-    public final Timestamp Eleven_oClock = new Timestamp(1682931600); // 2023-05-01 11:00:00
-    public final Timestamp Twelve_oClock = new Timestamp(1682935200); // 2023-05-01 12:00:00
+    public final Timestamp Ten_oClock = new Timestamp(1682928000000L); // 2023-05-01 10:00:00
+    public final Timestamp Eleven_oClock = new Timestamp(1682931600000L); // 2023-05-01 11:00:00
+    public final Timestamp Twelve_oClock = new Timestamp(1682935200000L); // 2023-05-01 12:00:00
 
     private static List<Object> mocks = new ArrayList<>();
 
@@ -65,7 +65,7 @@ protected void resetMocks() {
     }
 
     @AfterEach
-    private void validateTestEndpoint() {
+    public void validateTestEndpoint() {
         try {
             assertFalse(testEndpoint.hasExceptions());
         }

src/test/java/community/solace/spring/cloud/requestreply/AbstractRequestReplyMessageInterceptorIT.java

@@ -45,9 +45,9 @@ public abstract class AbstractRequestReplyMessageInterceptorIT {
     public static final String PROFILE_TEST_MESSAGE_INTERCEPTOR = "testMessageInterceptor";
     public static final String PROFILE_LOCAL_APP = "localApp";
 
-    public final Timestamp Ten_oClock = new Timestamp(1682928000); // 2023-05-01 10:00:00
-    public final Timestamp Eleven_oClock = new Timestamp(1682931600); // 2023-05-01 11:00:00
-    public final Timestamp Twelve_oClock = new Timestamp(1682935200); // 2023-05-01 12:00:00
+    public final Timestamp Ten_oClock = new Timestamp(1682928000000L); // 2023-05-01 10:00:00
+    public final Timestamp Eleven_oClock = new Timestamp(1682931600000L); // 2023-05-01 11:00:00
+    public final Timestamp Twelve_oClock = new Timestamp(1682935200000L); // 2023-05-01 12:00:00
 
     private static List<Object> mocks = new ArrayList<>();
 
@@ -66,7 +66,7 @@ protected void resetMocks() {
     }
 
     @AfterEach
-    private void validateTestEndpoint() {
+    public void validateTestEndpoint() {
         try {
             assertFalse(testEndpoint.hasExceptions());
         }