fix: implement requester-side deduplication of reply messages by repl…

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

@@ -8,6 +8,8 @@ Consult the table below to determine which version you need to use:
 
 | Spring Cloud | spring-cloud-stream-starter-request-reply | Spring Boot | sol-jcsmp |
 |--------------|-------------------------------------------|-------------|-----------|
+| 2025.0.0     | 5.3.5                                     | 3.5.8       | 10.29.0   |
+| 2025.0.0     | 5.3.4                                     | 3.5.8       | 10.29.0   |
 | 2025.0.0     | 5.3.3                                     | 3.5.8       | 10.29.0   |
 | 2025.0.0     | 5.3.2                                     | 3.5.8       | 10.29.0   |
 | 2025.0.0     | 5.3.1                                     | 3.5.6       | 10.28.1   |
@@ -27,7 +29,7 @@ To enable the request/reply functionality, please add the following section to y
 <dependency>
     <groupId>community.solace.spring.cloud</groupId>
     <artifactId>spring-cloud-stream-starter-request-reply</artifactId>
-    <version>5.3.3</version>
+    <version>5.3.5</version>
 </dependency>
 ```
 
@@ -599,23 +601,32 @@ More precisely:
 - If a service dies, any relations are forgotten and replies can no longer be related to request,
   potentially resulting in message loss.
 
-## External Links
-- [Spring Cloud Stream Solace Samples](https://solace.com/samples/solace-samples-spring/spring-cloud-stream/)
+### 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**.
 
-## Compatibility
+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.
+
+#### Dedup bitmap size limit (unknown / streaming totalReplies)
+
+When `totalReplies` is not known yet (e.g. streaming / unknown-size reply patterns), the requester still deduplicates numeric `replyIndex` values, but it must place an upper bound on how large the internal bitmap can grow.
 
-Tested with:
+You can configure this limit via:
 
-| SpringBoot 	 | SpringCloudStream   |
-|:-------------|:--------------------|
-| 2.6.6      	 | 2021.0.1            |
-| 2.6.6      	 | 2021.0.3            |
-| 3.2.5        | 2023.0.1          	 |
-| 3.5.4        | 2025.0.0            |
+```properties
+spring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown=100000
+```
 
-<!-- reused links -->
+- Default: **100000** bits
+- Effect: any `replyIndex` (or range end) above this limit will not be deduplicated.
 
-[@Order in Spring @Baeldung]: https://www.baeldung.com/spring-order
-[Spring Cloud Stream Binders]: https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#spring-cloud-stream-overview-binders
+#### Example log message
 
+```
+2023-10-04 10:00:00.000  INFO 12345 --- [nio-8080-exec-1] c.s.s.requestreply.examples.sending     : <<< MyRequest(location=livingroom) [correlationId=12345, replyTo=requestReply/response/solace/*/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d, ...]
+2023-10-04 10:00:00.000  INFO 12345 --- [nio-8080-exec-1] c.s.s.requestreply.examples.sending     : >>> SensorReading(foo=1337) [correlationId=12345, replyTo=requestReply/response/solace/*/pub_sub_sending_K353456_315fd96b-b981-417b-be99-3be065c6611d, remainingReplies=0, ...]
+```

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,7 @@
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.BitSet;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -32,6 +33,25 @@ public class ResponseHandler {
 
     private final RequestReplyLogger requestReplyLogger;
 
+    /**
+     * Fast-path dedup store for replyIndex values when totalReplies is known.
+     *
+     * <p>We store numeric indices (e.g. "12") and numeric ranges (e.g. "0-15") inside a BitSet.
+     * This is designed for very large reply counts without excessive allocation pressure.</p>
+     *
+     * <p>Protected by synchronizing on {@code numericReplyIndexBitSetLock} because BitSet is not thread-safe.</p>
+     */
+    private volatile BitSet numericReplyIndexBitSet;
+    private volatile int numericReplyIndexBitSetSize = -1;
+    private final Object numericReplyIndexBitSetLock = new Object();
+
+    // Grow limit for unknown-size / streaming cases to avoid unbounded memory use on malformed indices.
+    // For known totalReplies, numericReplyIndexBitSetSize will cap growth.
+    private static final int MAX_DEDUP_BITS_WHEN_UNKNOWN = Integer.getInteger(
+            "spring.cloud.stream.requestreply.dedup.maxBitsWhenUnknown",
+            100_000
+    );
+
     public ResponseHandler(Consumer<Message<?>> responseMessageConsumer, boolean supportMultipleResponses, Timer timer, RequestReplyLogger requestReplyLogger) {
         this.countDownLatch = new CountDownLatch(1);
         this.responseMessageConsumer = responseMessageConsumer;
@@ -54,6 +74,49 @@ public void receive(Message<?> message) {
         requestReplyLogger.logReply(LOG, Level.DEBUG, "received response(remaining={}) {}", remainingReplies, message);
     }
 
+
+    public boolean checkDuplicate(String replyIndex) {
+        if (replyIndex == null) {
+            return false;
+        }
+
+        // Support both "n" and "start-end" (range dedup is by start only).
+        Integer start = tryParseNonNegativeReplyIndexStart(replyIndex);
+        if (start == null) {
+            // Keep runtime fast: ignore non-numeric indices rather than allocating fallback structures.
+            requestReplyLogger.log(LOG, Level.DEBUG, "replyIndex '{}' is not numeric; skipping dedup", replyIndex);
+            return false;
+        }
+
+        synchronized (numericReplyIndexBitSetLock) {
+            // Lazily allocate & grow BitSet.
+            if (numericReplyIndexBitSet == null) {
+                numericReplyIndexBitSet = new BitSet(Math.min(1024, Math.max(start + 1, 0)));
+            }
+
+            // If totalReplies is known, clamp to [0, total-1].
+            if (numericReplyIndexBitSetSize > 0) {
+                if (start >= numericReplyIndexBitSetSize) {
+                    return false;
+                }
+            } else {
+                // Unknown totalReplies (streaming): cap growth to avoid unbounded memory.
+                if (start >= MAX_DEDUP_BITS_WHEN_UNKNOWN) {
+                    return false;
+                }
+            }
+
+            // Dedup by start index only.
+            if (numericReplyIndexBitSet.get(start)) {
+                requestReplyLogger.log(LOG, Level.WARN, "received duplicate response(index={})", replyIndex);
+                return true;
+            }
+
+            numericReplyIndexBitSet.set(start);
+            return false;
+        }
+    }
+
     public void await() throws RemoteErrorException, InterruptedException {
         countDownLatch.await();
         if (StringUtils.hasText(errorMessage)) {
@@ -66,6 +129,12 @@ public void setTotalReplies(Long totalReplies) {
             // Set total messages to expect when a multi message on a first message.
             expectedReplies.set(totalReplies);
             isFirstMessage = false;
+
+            // If totalReplies is known and within Integer range, enable bounded numeric dedup.
+            if (totalReplies <= Integer.MAX_VALUE) {
+                numericReplyIndexBitSetSize = totalReplies.intValue();
+                // Don't eagerly allocate; it might never be needed if replyIndex isn't present.
+            }
         }
     }
 
@@ -87,7 +156,46 @@ public void errorResponse(String errorMessage) {
         finished();
     }
 
+
+    private static Integer tryParseNonNegativeReplyIndexStart(String text) {
+        // Accept:
+        // - "123" -> 123
+        // - "12-34" -> 12 (start)
+        // Reject:
+        // - "-1", "12-", "-12", "12-34-56", "", non-digits
+        int len = text.length();
+        if (len == 0) {
+            return null;
+        }
+
+        int value = 0;
+        for (int i = 0; i < len; i++) {
+            char c = text.charAt(i);
+
+            if (c == '-') {
+                // must not be leading or trailing
+                return (i == 0 || i == len - 1) ? null : value;
+            }
+
+            if (c < '0' || c > '9') {
+                return null;
+            }
+
+            value = value * 10 + (c - '0');
+            if (value < 0) {
+                return null;
+            }
+        }
+
+        return value;
+    }
+
     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()).
+        numericReplyIndexBitSet = null;
+        numericReplyIndexBitSetSize = -1;
+
         if (timer != null) {
             timer.record(Duration.between(requestTime, Instant.now()));
         }