Why Did My RocketMQ Consumer Accumulate 300M Messages? The Hidden ClientId Bug

Preface

Anyone who has used a message queue may have encountered message accumulation. In this case an alert indicated that the topic XXX had more than 300 million pending messages, prompting a deep investigation.

Main Content

After receiving the alert, I logged into the RocketMQ console (self‑hosted open‑source version) and discovered the massive backlog. The producer and consumer applications appeared healthy, with normal disk I/O and network, yet the queue kept growing.

Producer speed >> Consumer speed Producer traffic spikes suddenly. Consumer instances become slow due to I/O blockage or crashes.

Inspecting the consumer logs showed no errors, but the three consumer instances all reported the same ClientId . This raised the suspicion that identical clientIds might be confusing the broker during message distribution.

Problem Analysis

The identical clientIds were traced to the Docker host‑network mode. When containers run in host mode they share the host’s network stack, and the virtual bridge

docker0

provides the default IP

172.17.0.1

. The clientId generation logic in RocketMQ concatenates the client IP with an instance name.

<code>public String buildMQClientId() {
    StringBuilder sb = new StringBuilder();
    sb.append(this.getClientIP());
    sb.append("@");
    sb.append(this.getInstanceName());
    if (!UtilAll.isBlank(this.unitName)) {
        sb.append("@");
        sb.append(this.unitName);
    }
    return sb.toString();
}
</code>

The IP is obtained via

RemotingUtil.getLocalAddress()

, which scans network interfaces and prefers a non‑loopback, non‑private IPv4 address. In host mode the first address returned is the

docker0

IP, so every container receives the same IP component of the clientId.

<code>public static String getLocalAddress() {
    try {
        Enumeration<NetworkInterface> enumeration = NetworkInterface.getNetworkInterfaces();
        ArrayList<String> ipv4Result = new ArrayList<>();
        while (enumeration.hasMoreElements()) {
            NetworkInterface networkInterface = enumeration.nextElement();
            Enumeration<InetAddress> en = networkInterface.getInetAddresses();
            while (en.hasMoreElements()) {
                InetAddress address = en.nextElement();
                if (!address.isLoopbackAddress()) {
                    if (address instanceof Inet6Address) {
                        // add IPv6
                    } else {
                        ipv4Result.add(normalizeHostAddress(address));
                    }
                }
            }
        }
        // prefer IPv4, skip 127.0.* and 192.168.*
        for (String ip : ipv4Result) {
            if (ip.startsWith("127.0") || ip.startsWith("192.168")) continue;
            return ip;
        }
        return ipv4Result.get(ipv4Result.size() - 1);
    } catch (Exception e) {
        // fallback
    }
    return null;
}
</code>

The instance name defaults to the system property

rocketmq.client.name

or

DEFAULT

. If left unchanged, RocketMQ replaces

DEFAULT

with the process PID, which is also identical across containers because they share the same PID namespace.

Source Exploration

Key excerpts from

ClientConfig

illustrate the generation steps:

<code>private String clientIP = RemotingUtil.getLocalAddress();
private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");

public String getInstanceName() { return instanceName; }
public void setInstanceName(String instanceName) { this.instanceName = instanceName; }
public void changeInstanceNameToPID() {
    if (this.instanceName.equals("DEFAULT")) {
        this.instanceName = String.valueOf(UtilAll.getPid());
    }
}
</code>

Load‑Balancing Mechanism

RocketMQ performs consumer load‑balancing on the client side. The broker stores a

consumerTable

that maps each consumer’s clientId to its assigned message queues. The core algorithm resides in

rebalanceByTopic()

and ultimately calls the

AllocateMessageQueueStrategy

implementation.

<code>private void rebalanceByTopic(final String topic, final boolean isOrder) {
    Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
    List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
    if (mqSet != null && cidAll != null) {
        List<MessageQueue> mqAll = new ArrayList<>(mqSet);
        Collections.sort(mqAll);
        Collections.sort(cidAll);
        AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;
        List<MessageQueue> allocateResult = strategy.allocate(
            this.consumerGroup,
            this.mQClientFactory.getClientId(),
            mqAll,
            cidAll);
        // update processQueueTable based on allocateResult
    }
}
</code>

If multiple consumers share the same clientId, the

cidAll

list contains duplicate entries, causing every consumer to receive the same index (0) and thus be allocated the same set of queues. This explains why the consumption speed was extremely low despite three consumer instances.

Solution

Fix the clientId generation by setting a unique

rocketmq.client.name

environment variable, e.g. appending a timestamp to the PID:

<code>@PostConstruct
public void init() {
    System.setProperty("rocketmq.client.name",
        String.valueOf(UtilAll.getPid()) + "@" + System.currentTimeMillis());
}
</code>

After deploying the change, the backlog gradually decreased and the alert disappeared.

Conclusion

RocketMQ consumer clientId is composed of

clientIP + "@" + instanceName

; in Docker host‑network mode the IP resolves to

docker0

's default address, making clientIds identical.

Identical clientIds cause the client‑side load‑balancing algorithm to assign the same message queues to all consumers, leading to severe consumption slowdown.

Customizing

rocketmq.client.name

(or otherwise ensuring unique clientIds) resolves the load‑balancing error and eliminates message accumulation.

When handling large backlogs, always verify producer/consumer health, network configuration, and client identifier uniqueness.