[Enhancement]Add deduplication for message consumers

[ChaoXoX]

Before Creating the Enhancement Request

• 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