[ISSUE #10263] Add consumer message deduplication with correct ackIndex semantics

[Aias00]

What problem does this PR solve?

Issue: #10263

This PR adds message deduplication capability for consumers to prevent duplicate message processing, with correct handling of ackIndex semantics when duplicates are filtered.

Problem Summary

When messages are redelivered due to network issues, rebalancing, or broker restarts, consumers may process the same message multiple times, causing:

• Duplicate database writes
• Repeated business logic execution
• Data inconsistency

Key Changes

1. MessageDeduplicator: Lightweight deduplication cache using ConcurrentHashMap with scheduled cleanup

   • Keys: user-defined message keys (preferred) or msgId
   • Configurable cache size and expiration time
   • Thread-safe concurrent access

2. ConsumeMessageConcurrentlyService: Filter duplicates before listener invocation

   • Skip duplicate messages, advancing offset correctly
   • Correctly handle ackIndex semantics for partial success:
     • Only mark messages up to ackIndex in filteredMsgs
     • Map filtered-list ackIndex to original msgs position

3. Configuration:

   • deduplicationEnabled: Enable/disable feature (default: false)
   • deduplicationCacheSize: Max cache entries (default: 10000)
   • deduplicationExpireTime: Entry expiration in ms (default: 60000)

Critical Fix for ackIndex Semantics

When duplicates exist and listener returns partial success:

• Before: All filteredMsgs marked as processed, ackIndex set to msgs.size()-1 → Message loss
• After: Only filteredMsgs[0..ackIndex] marked, ackIndex mapped to original position → Correct retry behavior

Example:

msgs = [dup, new1, new2]
filteredMsgs = [new1, new2]
listener returns: ackIndex=0 (only new1 success)

Old behavior: new1, new2 both marked → new2 lost
New behavior: only new1 marked → new2 correctly retried

Tests

• MessageDeduplicationTest: 13 tests covering:

  • Basic duplicate detection
  • Key extraction (user keys vs msgId)
  • Cache expiration and size limits
  • Concurrent access safety
  • ackIndex mapping for partial/full success
  • Filtering order preservation

• ConsumeMessageConcurrentlyServiceTest: Existing tests pass

How to use

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
consumer.setDeduplicationEnabled(true);
consumer.setDeduplicationCacheSize(10000);
consumer.setDeduplicationExpireTime(60000);
consumer.start();

Documentation

• Updated DefaultMQPushConsumer.md with deduplication configuration

Verification

• All unit tests pass
• Diff format check passes

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR introduces an optional consumer-side message deduplication capability for concurrent consumption, aiming to reduce duplicate processing while preserving correct ackIndex semantics when duplicates are filtered out before listener invocation.

Changes:

• Add a MessageDeduplicator cache and lifecycle wiring in DefaultMQPushConsumerImpl.
• Filter duplicate messages in ConsumeMessageConcurrentlyService and map ackIndex from the filtered list back to the original batch.
• Expose consumer configuration knobs for enabling deduplication and tuning cache size/expiry, plus unit tests.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
client/src/main/java/org/apache/rocketmq/client/impl/consumer/MessageDeduplicator.java	Adds a time-expiring deduplication key cache with a scheduled cleanup task.
client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java	Filters duplicates before listener invocation and attempts to map/apply ackIndex correctly.
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java	Initializes/shuts down the deduplicator and validates dedup-related config.
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumer.java	Adds enablement + cache tuning configuration fields and accessors.
client/src/test/java/org/apache/rocketmq/client/impl/consumer/MessageDeduplicationTest.java	Adds unit tests for dedup key extraction, expiration, and ackIndex-related scenarios.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 64.48087% with 65 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.20%. Comparing base (0e4ccf1) to head (d9f9667).
⚠️ Report is 1 commits behind head on develop.

Files with missing lines	Patch %	Lines
...etmq/client/impl/consumer/MessageDeduplicator.java	70.58%	17 Missing and 8 partials ⚠️
...lient/impl/consumer/DefaultMQPushConsumerImpl.java	4.34%	16 Missing and 6 partials ⚠️
...pl/consumer/ConsumeMessageConcurrentlyService.java	80.95%	5 Missing and 7 partials ⚠️
...ocketmq/client/consumer/DefaultMQPushConsumer.java	50.00%	6 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10606      +/-   ##
=============================================
- Coverage      48.27%   48.20%   -0.08%     
- Complexity     13436    13455      +19     
=============================================
  Files           1378     1379       +1     
  Lines         100820   100987     +167     
  Branches       13041    13071      +30     
=============================================
+ Hits           48670    48678       +8     
- Misses         46208    46334     +126     
- Partials        5942     5975      +33     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Adds consumer-side message deduplication for concurrent consumption mode, using a time-based cache to detect and skip duplicate messages before listener invocation. Includes correct ackIndex handling for partial success scenarios.

Findings

• [Critical] MessageDeduplicator.java — Missing lifecycle management for cleanupExecutor: The ScheduledExecutorService is created in the constructor but there is no shutdown() method or integration with consumer shutdown lifecycle. This will cause a thread leak when the consumer is stopped — the cleanup thread will prevent JVM shutdown and waste resources. Consider adding a shutdown() method and calling it from ConsumeMessageConcurrentlyService.shutdown() or DefaultMQPushConsumerImpl.shutdown().

• [Warning] MessageDeduplicator.java — PR description says "using Caffeine" but implementation uses ConcurrentHashMap: The PR body states "Lightweight deduplication cache using Caffeine" but the actual implementation uses a plain ConcurrentHashMap with manual scheduled cleanup. Either update the description to match the implementation, or consider using Caffeine which provides built-in expiration (expireAfterWrite) and size-based eviction (maximumSize), eliminating the need for the manual cleanup executor entirely.

• [Warning] ConsumeMessageConcurrentlyService.java:455 — Inconsistent message list in hook context: When all messages are duplicates, the code passes the original msgs to consumeMessageContext.setMsgList() (via filteredMsgs.isEmpty() ? msgs : filteredMsgs) but uses an empty list for actual consumption. Hook implementations that inspect msgList will see the original list even though no messages were actually consumed. Consider passing the filtered list consistently, or documenting this behavior.

• [Warning] MessageDeduplicator.java:removeOldestEntries() — O(n log n) cleanup cost: When the cache exceeds maxCacheSize, the cleanup sorts all entries by timestamp to remove the oldest 25%. For a cache of 100,000 entries, this creates and sorts a large stream on every eviction. Consider using a more efficient eviction strategy (e.g., a ConcurrentLinkedDeque for access ordering, or Caffeine's built-in eviction).

• [Info] MessageDeduplicator.java — Thread factory naming: The cleanup thread uses a fixed prefix "DedupCleanup_" without the consumer group identifier. In environments with multiple consumers, this makes it hard to identify which consumer's cleanup thread is running in thread dumps. Consider passing the consumer group to the constructor and including it in the thread name.

• [Info] DefaultMQPushConsumer.java — No setter validation: setDeduplicationCacheSize() and setDeduplicationCacheExpireTime() accept any value without range checks. Validation only occurs in checkConfig() at consumer start. If values are changed after start (e.g., via JMX or dynamic config), invalid values could cause unexpected behavior. Consider adding range validation in the setters or documenting that changes after start are not supported.

• [Info] ConsumeMessageConcurrentlyService.java:filterDuplicateMessages() — isDuplicate race condition: The isDuplicate() method checks if a key exists and then removes it, but this check-then-remove is not atomic. Between the check and removal, another thread could have already removed the entry. While this is benign (worst case: the same message is processed twice, which deduplication is meant to prevent but cannot guarantee in all race conditions), it's worth documenting this limitation.

Suggestions

1. Add a shutdown() method to MessageDeduplicator and wire it into the consumer shutdown path — this is the most important fix.
2. Consider replacing the manual ConcurrentHashMap + cleanup executor with Caffeine's Cache<String, Boolean> with expireAfterWrite() and maximumSize() for cleaner, more efficient eviction.
3. Add an integration test that verifies deduplication works correctly across consumer restarts (cache should be empty after restart, so messages should not be deduplicated).
4. The feature is documented as concurrent-only. Consider adding a runtime warning in ConsumeMessageOrderlyService and ConsumeMessagePopConcurrentlyService when deduplication is enabled but not supported.

Cross-repo Note

This is a client-side feature. No changes needed in apache/rocketmq broker/proxy. However, consider documenting that this is a best-effort, in-memory deduplication — it does not provide exactly-once semantics across consumer restarts or in multi-instance deployments.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Review by github-manager-bot (Re-review)

Summary

Author addressed previous review feedback in commit 14d31189. Key improvements: lifecycle management added, race condition fixed, initialization guard for non-concurrent modes, and ackIndex mapping refactored into a testable static method with comprehensive tests.

Changes Since Last Review

✅ [Critical] Lifecycle management — FIXED: MessageDeduplicator.shutdown() properly shuts down the cleanup executor with timeout and interrupt handling. Called from ConsumeMessageConcurrentlyService.shutdown() in the consumer lifecycle. Thread leak concern resolved.

✅ [Warning] Race condition in recordProcessed() — FIXED: Changed from unconditional remove() + put() to conditional processedMessages.remove(messageKey, timestamp). This correctly prevents the ABA problem where a concurrent cleanup could remove an entry that was just re-added by another thread.

✅ [Warning] Deduplicator created for all consumption modes — FIXED: DefaultMQPushConsumerImpl.start() now only initializes the deduplicator when enableMessageDeduplication && consumeMode == CONCURRENTLY. Orderly and POP modes no longer create unnecessary resources.

✅ ackIndex mapping — IMPROVED: Extracted mapAckIndex() as a package-private static method, making it independently testable. Added proper handling for the "all filtered messages succeed" case (returns originalMsgs.size() - 1 to cover trailing duplicates). New tests cover trailing duplicates, middle duplicates, no duplicates, and out-of-range ackIndex scenarios.

✅ Code readability — IMPROVED: Extracted markMessagesAsProcessed() method, separating the marking logic from the consumption callback. Much cleaner than the previous inline implementation.

Remaining Items (Non-blocking)

• [Warning] removeOldestEntries() still uses O(n log n) sort-based eviction. With maxCacheSize=100000, this creates a large sorted stream on every eviction. Consider a LinkedHashMap with access-order for O(1) eviction, or simply removing a fixed number of entries by iterating the entrySet iterator (which for ConcurrentHashMap is weakly consistent but sufficient for cache eviction).

• [Info] Thread name "DedupCleanup_" still lacks consumer group context. Minor, but helpful for thread dump analysis in multi-consumer deployments.

• [Info] PR description mentions "Caffeine" but implementation uses ConcurrentHashMap. Consider updating the description for accuracy.

Verdict

The critical issues from the previous review have been properly addressed. The code is well-structured with good test coverage. LGTM — ready for committer review.

---

Automated re-review by github-manager-bot

[RockteMQ-AI]

Review by github-manager-bot (Re-review #2)

Summary

New commit 46dd21cf addresses two real bugs found after the previous LGTM:

1. Batch-internal dedup leak — identical messages within the same batch were not caught because the global cache is only populated after successful consumption.
2. Cross-topic false duplicates — the same business key on different topics would collide in the dedup cache.

Changes Since Last Review

✅ Batch-internal dedup — FIXED: Added a seenInBatch HashSet scoped to each filterDuplicateMessages() call. Combined with the global cache check via short-circuit ||, this correctly collapses intra-batch duplicates. The method was refactored to a static package-private overload for testability — clean approach.

✅ Cross-topic key scoping — FIXED: getDeduplicationKey() now returns topic + "#" + rawKey. The # separator is safe per TopicValidator rules (topic names only allow ^[%|a-zA-Z0-9_-]+$). Group-level isolation is correctly provided structurally by the per-consumer-instance deduplicator.

✅ Tests — COMPREHENSIVE: New tests cover intra-batch dedup, mixed global+batch dedup, topic-scoped keys, null deduplicator passthrough, and the no-topic defensive fallback. Well-structured.

Remaining Items (Non-blocking)

• [Info] removeOldestEntries() still uses O(n log n) sort-based eviction. Not a regression from this commit, but worth tracking for future optimization (e.g., LinkedHashMap access-order or fixed-count eviction).

• [Info] Thread name "DedupCleanup_" still lacks consumer group context. Minor.

Verdict

Both bugs are properly fixed with clean implementation and good test coverage. LGTM — ready for committer review.

---

Automated re-review by github-manager-bot

[RockteMQ-AI]

Remaining Items

• [Warning] filteredMsgs.isEmpty() fallback in ConsumeRequest.run() may cause duplicate retries:

  this.processedMsgs = filteredMsgs.isEmpty() ? msgs : filteredMsgs;

  When all messages in a batch are duplicates, filteredMsgs is empty and the code falls back to msgs (the original batch). The listener is then invoked with messages that were all identified as duplicates. If the listener returns RECONSUME_LATER, all those "duplicate" messages are sent back for retry — defeating the purpose of deduplication.

  Suggested fix: When filteredMsgs.isEmpty(), skip the listener invocation and treat the batch as CONSUME_SUCCESS (all were duplicates, nothing to do):

  List<MessageExt> filteredMsgs = filterDuplicateMessages(msgs);
  if (filteredMsgs.isEmpty()) {
      // All messages are duplicates — nothing to consume.
      // Mark offset and return without invoking the listener.
      long offset = consumeRequest.getProcessQueue().removeMessage(msgs);
      // ... update offset ...
      return;
  }
  this.processedMsgs = filteredMsgs;

  This also avoids the unnecessary re-marking of duplicate entries in the dedup cache (which extends their TTL without benefit).

• [Info] Minor: the processedMsgs field defaults to msgs in the constructor and is reassigned in run(). Consider initializing it to null and asserting non-null in processConsumeResult() to catch any future code path that forgets to set it. Non-blocking.

Verdict

The processedMsgs approach is a clear improvement in clarity and correctness. The filteredMsgs.isEmpty() fallback is the only remaining edge case worth addressing. LGTM with the above suggestion — ready for committer review once the empty-filter case is handled.

---

Automated re-review by github-manager-bot

[RockteMQ-AI]

Review by github-manager-bot (Re-review #3)

Summary

New commit d9f96676 ("Handle all-duplicate batch without re-marking dedup cache") correctly addresses the remaining warning items from the previous review. The all-duplicate batch edge case is now handled cleanly.

Changes Since Last Review

✅ [Warning] filteredMsgs.isEmpty() fallback — FIXED: When all messages in a batch are duplicates, processedMsgs is now an empty list. The listener is not invoked (returns CONSUME_SUCCESS directly), no messages are sent back for retry, and dedup cache entries are not re-stamped (TTL preserved). This eliminates the previous issue where the listener would be called with duplicate messages and cache entries would have their TTL silently extended.

✅ [Info] processedMsgs field initialization — IMPROVED: Changed from defaulting to msgs in the constructor to intentionally null until run() sets it. processConsumeResult now has an explicit null check with a warning log. This makes programming errors visible rather than silently using wrong data — good defensive design.

✅ Hook context correctness — FIXED: consumeMessageContext.setMsgList(processedMsgs) now passes the filtered list, so hook implementations see exactly what the listener consumes.

✅ Test coverage — COMPREHENSIVE: New test testAllDuplicatesSkipsListenerAndDoesNotRefreshTtl verifies:

• Listener not invoked when all messages are duplicates
• No messages sent back for retry
• ProcessQueue properly drained (offset advances)
• Cache timestamps not refreshed (TTL not extended)

Minor Observations (Non-blocking)

• [Info] msgsToConsume variable: processedMsgs.isEmpty() ? Collections.emptyList() : processedMsgs is functionally equivalent to just processedMsgs (since an empty list is already empty). The explicit intent comment makes the purpose clear, so this is fine as a readability choice.

• [Info] getProcessedTimestamp(): Package-private test-only accessor is acceptable. Consider adding a @VisibleForTesting annotation if the project uses Guava, for clarity.

Verdict

All previously identified issues have been addressed. The deduplication feature is now well-structured with correct edge case handling, proper lifecycle management, and comprehensive test coverage. LGTM.

---

Automated review by github-manager-bot