{"record":{"id":"e78b69d5ec637e70","repo":"alibaba/canal","slug":"subscript-pulsar-consumer-error","errorCode":null,"errorMessage":"Subscript pulsar consumer error","messagePattern":"Subscript pulsar consumer error","errorType":"exception","errorClass":"CanalClientException","httpStatus":null,"severity":"error","filePath":"connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java","lineNumber":225,"sourceCode":"                dlqBuilder.deadLetterTopic(dlqTopic);\n            }\n\n            // 默认关闭，如果需要重试则开启\n            builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());\n        }\n\n        // ack超时\n        builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);\n\n        // pulsar批量获取消息设置\n        builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)\n            .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)\n            .build());\n\n        try {\n            this.pulsarMQConsumer = builder.subscribe();\n        } catch (PulsarClientException e) {\n            throw new CanalClientException(\"Subscript pulsar consumer error\", e);\n        }\n    }\n\n    @SuppressWarnings(\"unchecked\")\n    @Override\n    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {\n        List<CommonMessage> messageList = Lists.newArrayList();\n        try {\n            Messages<byte[]> messages = pulsarMQConsumer.batchReceive();\n            if (null == messages || messages.size() == 0) {\n                return messageList;\n            }\n            // 保存当前消费记录，用于ack和rollback\n            this.lastGetBatchMessage = messages;\n            for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) {\n                byte[] data = msg.getData();\n                if (!this.flatMessage) {\n                    Message message = CanalMessageSerializerUtil.deserializer(data);","sourceCodeStart":207,"sourceCodeEnd":243,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java#L207-L243","documentation":"Thrown by CanalPulsarMQConsumer.connect when ConsumerBuilder.subscribe() raises a PulsarClientException. By this point the PulsarClient is already built; the failure is specifically about creating the consumer on a topic/subscription — invalid service URL, authentication failure, non-existent topic (when not auto-creating), subscription config conflict, or broker unreachable.","triggerScenarios":"builder.subscribe() at line 223 fails. Causes: wrong/malformed serviceUrl; token auth rejected; topic does not exist and broker auto-creation is disabled; subscription name conflicts with an incompatible existing subscription type; broker/network down after client connect.","commonSituations":"Misconfigured pulsarmq.server.url; expired or wrong role token; topic name typo or tenant/namespace that does not exist; Failover subscription already held by another active consumer; broker temporarily unavailable during consumer startup.","solutions":["Verify pulsarmq.server.url is reachable and is a valid puls:// or http(s) Pulsar service URL.","Confirm the topic exists (or broker permits auto-creation) and the topic/tenant/namespace string is correct.","Check the role token is valid and not expired when PULSARMQ_ROLE_TOKEN is set.","For Failover subscriptions, ensure no other consumer currently holds the subscription, or switch to Shared if concurrent consumers are intended.","Retry with backoff — a transient broker outage during startup often clears on reconnect."],"exampleFix":"// before\ntry {\n    this.pulsarMQConsumer = builder.subscribe();\n} catch (PulsarClientException e) {\n    throw new CanalClientException(\"Subscript pulsar consumer error\", e);\n}\n\n// after — preserve the broker error detail for diagnosis\ntry {\n    this.pulsarMQConsumer = builder.subscribe();\n} catch (PulsarClientException e) {\n    throw new CanalClientException(\n        \"Subscript pulsar consumer error [url=\" + serviceUrl + \", topic=\" + topic\n        + \", sub=\" + subscriptName + \"]: \" + e.getMessage(), e);\n}","handlingStrategy":"retry","validationCode":"// Pre-flight checks before connect()\nif (StringUtils.isEmpty(serviceUrl)) throw new IllegalStateException(\"pulsar serviceUrl missing\");\nif (StringUtils.isEmpty(topic)) throw new IllegalStateException(\"pulsar topic missing\");\nif (StringUtils.isEmpty(subscriptName)) throw new IllegalStateException(\"pulsar subscriptName missing\");","typeGuard":null,"tryCatchPattern":"int attempts = 0;\nwhile (attempts++ < 3) {\n    try {\n        this.pulsarMQConsumer = builder.subscribe();\n        break;\n    } catch (PulsarClientException e) {\n        if (attempts >= 3)\n            throw new CanalClientException(\"Subscript pulsar consumer error\", e);\n    }\n}","preventionTips":["Verify the service URL and topic/namespace exist before subscribing.","Ensure the role token is valid and not expired.","For Failover subscriptions, avoid competing consumers holding the same subscription."],"tags":["pulsar","network","authentication","canal-connector","initialization"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}