How to End Duplicate Consumption in RocketMQ with Idempotence and High‑Concurrency Architecture

Why Duplicate Consumption Happens

In distributed messaging systems such as RocketMQ, Kafka or Pulsar, the "at‑least‑once" delivery guarantee means that a message may be redelivered when the broker does not receive a consumer acknowledgement, when a producer retries after a network timeout, or when a consumer crashes before committing its offset.

Three stages can generate duplicates:

Producer send stage : network jitter may cause the producer to think a send failed while the broker has already persisted the message, leading to a retry that creates two identical messages.

Broker storage and failover stage : master‑slave switch, network partition, disk flush delay or replica lag cause the broker to replay messages to avoid loss.

Consumer stage : if the consumer crashes after processing but before sending CONSUME_SUCCESS, the broker will redeliver the message; long processing time can also trigger a timeout and a retry.

Idempotence Is the Real Defense

Because duplicate delivery cannot be eliminated, the only reliable protection is to make the business operation idempotent. The key principles are:

Never rely on msgId for deduplication; it changes on each retry.

Derive a business key from the domain data (e.g., orderId, paymentNo, userId+activityId+couponId).

The business key must be unique per logical operation and include the consumer group when the same topic is consumed by multiple groups.

Production‑Grade Idempotent Architecture

The recommended flow is:

Receive message → Parse bizKey → Check Redis success cache → Acquire Redisson lock → Insert PROCESSING record into MySQL → If record exists, handle according to its status (SUCCESS, PROCESSING, FAILED) → Execute business logic → On success, update record to SUCCESS and write Redis cache → Release lock

Key components:

Redis success cache : stores a short‑lived flag (e.g., 3 days) so that subsequent duplicates return immediately.

Redisson distributed lock : prevents concurrent processing of the same business key.

MySQL idempotent table ( mq_idempotent_record) with a unique index on (msg_key, topic, consumer_group) and a status column (PROCESSING, SUCCESS, FAILED).

AOP interceptor ( @MqIdempotent) that wraps the consumer method, performs the above steps, and throws MqRetryableException for retryable failures or MqAlreadyConsumedException when the record is already SUCCESS.

Database Schema (simplified)

CREATE TABLE mq_idempotent_record (
  id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
  msg_key VARCHAR(128) NOT NULL,
  topic VARCHAR(128) NOT NULL,
  consumer_group VARCHAR(128) NOT NULL,
  status TINYINT NOT NULL COMMENT '0=PROCESSING,1=SUCCESS,2=FAILED',
  retry_count INT NOT NULL DEFAULT 0,
  request_body TEXT,
  error_msg VARCHAR(1024),
  processing_expire_time DATETIME NOT NULL,
  create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  UNIQUE KEY uk_msg_key_group (msg_key, topic, consumer_group)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

Sample AOP Implementation

@Around("@annotation(idempotent)")
public Object around(ProceedingJoinPoint jp, MqIdempotent idempotent) throws Throwable {
    String bizKey = bizKeyResolver.resolve(jp, idempotent.bizKey());
    String successKey = buildSuccessCacheKey(idempotent.topic(), idempotent.consumerGroup(), bizKey);
    if ("1".equals(redisTemplate.opsForValue().get(successKey))) {
        // cache hit → idempotent success
        return null;
    }
    RLock lock = redissonClient.getLock(buildLockKey(...));
    if (!lock.tryLock(idempotent.lockWaitMillis(), idempotent.processingTimeoutSeconds(), TimeUnit.MILLISECONDS)) {
        throw new MqRetryableException("lock busy");
    }
    try {
        // insert PROCESSING record, handle existing record, execute business, update status, cache success
        Object result = jp.proceed();
        repository.markSuccess(record.getId());
        redisTemplate.opsForValue().set(successKey, "1", Duration.ofSeconds(idempotent.successCacheSeconds()));
        return result;
    } catch (Throwable ex) {
        repository.markFailed(record.getId(), ex.getMessage());
        throw ex;
    } finally {
        if (lock.isHeldByCurrentThread()) lock.unlock();
    }
}

High‑Concurrency Optimizations

Key sharding: include a hash slot in Redis keys to spread load across cluster nodes.

Local Caffeine cache for recent SUCCESS entries (only SUCCESS, never PROCESSING).

MySQL partitioning by create_time to keep the idempotent table manageable; hot data stays in a non‑partitioned table, older data is archived.

Kubernetes Graceful Shutdown

Configure terminationGracePeriodSeconds and a pre‑stop hook that blocks new traffic, allowing the consumer thread to finish processing and commit offsets before the pod is killed. Spring Boot’s graceful shutdown and a readiness probe ensure no new messages are assigned during termination.

Observability & Monitoring

Instrument the flow with Micrometer counters such as mq_idempotent_success_cache_hit, mq_idempotent_lock_busy, mq_idempotent_processing_hit, and latency metrics. Log a structured line for every consume attempt containing traceId, topic, consumerGroup, bizKey, reconsumeTimes, cost, and idempotentStatus. Export metrics to Prometheus and trace bizKey via OpenTelemetry.

Dead‑Letter Queue Handling

When a message repeatedly fails (format error, missing data, permanent downstream outage), it lands in a dead‑letter queue. Provide an admin UI to view the payload, failure reason, and bizKey, allow data correction, and replay the message while preserving the original bizKey.

Real‑World Example: Coupon Grant Idempotence

A coupon‑grant service uses userId+activityId+couponId as the business key. The handler is annotated with @MqIdempotent; the service inserts a row into user_coupon with a unique index on the same three columns, and decrements stock with a conditional UPDATE … WHERE stock>0. This guarantees that even if the grant message is delivered many times, each user receives the coupon only once and stock never goes negative.

Common Pitfalls and Fixes

Writing a Redis lock before business success leads to lost messages on failure – write SUCCESS cache only after the transaction commits.

Returning silently on PROCESSING causes lost messages – throw a retryable exception instead.

Using only msgId for deduplication – it changes on each retry.

Designing the idempotent key too coarse (e.g., only userId) blocks unrelated operations; too fine (e.g., requestId) fails to deduplicate genuine retries.

Evolution Roadmap

Start with a simple DB unique index for low QPS workloads, then add Redis caching for medium traffic, later introduce local caches for bursty duplicate spikes, and finally adopt a unified governance platform for multi‑team, multi‑topic environments.

Final Recommendation

Combine a business‑level unique key, RocketMQ message key, Redis success cache, Redisson lock, MySQL idempotent table, a robust state machine, downstream idempotent services, Prometheus monitoring, and a dead‑letter compensation workflow. This layered approach guarantees exactly‑once business results while preserving the high availability and scalability of RocketMQ.