{"record":{"id":"bb7db3c96b4e89d2","repo":"alibaba/canal","slug":"start-rocketmq-consumer-error","errorCode":null,"errorMessage":"Start RocketMQ consumer error","messagePattern":"Start RocketMQ consumer error","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"critical","filePath":"client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java","lineNumber":160,"sourceCode":"            }\n            rocketMQConsumer.subscribe(this.topic, \"*\");\n            rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {\n\n                @Override\n                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeOrderlyContext context) {\n                    context.setAutoCommit(true);\n                    boolean isSuccess = process(messageExts);\n                    if (isSuccess) {\n                        return ConsumeOrderlyStatus.SUCCESS;\n                    } else {\n                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;\n                    }\n                }\n            });\n            rocketMQConsumer.start();\n            connected = true;\n        } catch (MQClientException ex) {\n            throw new RuntimeException(\"Start RocketMQ consumer error\", ex);\n        }\n    }\n\n    private boolean process(List<MessageExt> messageExts) {\n        if (logger.isDebugEnabled()) {\n            logger.debug(\"Get Message: {}\", messageExts);\n        }\n        List messageList = new ArrayList<>();\n        for (MessageExt messageExt : messageExts) {\n            byte[] data = messageExt.getBody();\n            if (data != null) {\n                try {\n                    if (!flatMessage) {\n                        Message message = CanalMessageDeserializer.deserializer(data);\n                        messageList.add(message);\n                    } else {\n                        FlatMessage flatMessage = JSON.parseObject(data, FlatMessage.class);\n                        messageList.add(flatMessage);","sourceCodeStart":142,"sourceCodeEnd":178,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java#L142-L178","documentation":"Thrown by RocketMQCanalConnector.subscribe when the DefaultMQPushConsumer fails to subscribe to the topic, register the message listener, or call start(). The original MQClientException is wrapped in a RuntimeException. RocketMQ throws MQClientException for broker connectivity, name-server, ACL, or client-configuration problems.","triggerScenarios":"Calling subscribe() with a nameServer address that is unreachable or misconfigured; providing a topic that does not exist or the group lacks subscription permission; ACL (accessKey/secretKey) mismatch against an ACL-enabled broker; the consumer group name colliding or exceeding RocketMQ client limits.","commonSituations":"Wrong or unreachable namesrvAddr in a new environment; RocketMQ broker not started or firewalled; accessChannel set to \"cloud\" against a non-Aliyun broker or vice versa; the consumer group already in use with incompatible settings; namespace misconfiguration.","solutions":["Verify nameServer (rocketMQConsumer.setNamesrvAddr) is reachable: telnet/curl the nameserver port from the client host.","Inspect the wrapped MQClientException cause — its error code/message identifies broker-vs-config-vs-ACL failure.","Confirm the topic exists and the consumer group has SUB permission on the broker.","If using Aliyun ACL, validate accessKey/secretKey and set accessChannel=\"cloud\"; for self-hosted, leave it unset.","Ensure only one consumer instance uses the group, or that concurrent instances are expected (cluster mode)."],"exampleFix":"// before\nconnector.subscribe(); // throws RuntimeException on broker/auth failure\n// after\ntry {\n    connector.subscribe();\n} catch (RuntimeException e) {\n    Throwable cause = (e.getCause() != null) ? e.getCause() : e;\n    log.error(\"RocketMQ subscribe failed: {}\", cause.getMessage(), e);\n    throw e;\n}","handlingStrategy":"try-catch","validationCode":"// Pre-flight: verify nameserver reachability before subscribing.\ntry (java.net.Socket s = new java.net.Socket()) {\n    s.connect(new java.net.InetSocketAddress(nameServerHost, nameServerPort), 3000);\n} catch (IOException e) {\n    throw new IllegalStateException(\"RocketMQ nameserver unreachable: \" + nameServer, e);\n}","typeGuard":null,"tryCatchPattern":"try {\n    connector.subscribe();\n} catch (RuntimeException e) {\n    Throwable cause = e.getCause() instanceof MQClientException ? e.getCause() : e;\n    logger.error(\"RocketMQ subscribe failed: {}\", cause.getMessage(), cause);\n    // implement startup retry/backoff or fail fast per your topology\n    throw e;\n}","preventionTips":["Validate nameserver connectivity and ACL credentials before calling subscribe().","Inspect the wrapped MQClientException cause for the broker error code.","For Aliyun, set accessChannel=\"cloud\" and provide valid accessKey/secretKey.","Confirm the topic and consumer group are provisioned and authorized."],"tags":["rocketmq","canal-client","broker-connectivity","acl","startup"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}