[Enhancement]Add deduplication for message consumers

### Before Creating the Enhancement Request

- [x] I have confirmed that this should be classified as an enhancement rather than a bug/feature.

### Summary

Add deduplication for message consumers.

### Motivation

At-least-once delivery may lead to duplicate processing.

### Describe the Solution You'd Like

Add message_id
Maintain processed ID set
Skip duplicates

### Describe Alternatives You've Considered

Accept duplicates (risk inconsistent state)

### Additional Context

_No response_

The issue names no files, tests, or entry points. Start by locating the Java message-consumer implementation and its message identity handling; define how message_id values are supplied and retained, then verify that repeated deliveries are skipped while new messages are processed.

Written by the indexing model from the issue text.