Aias00 opened a new pull request, #10606:
URL: https://github.com/apache/rocketmq/pull/10606
## 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 Caffeine
- 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
```java
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](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
