{"record":{"id":"3b7c775b0ae428d2","repo":"alibaba/canal","slug":"pulsar-consumer-subscriptname-required-3b7c77","errorCode":null,"errorMessage":"Pulsar Consumer subscriptName required","messagePattern":"Pulsar Consumer subscriptName required","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java","lineNumber":127,"sourceCode":"\n    @Override\n    public void init(Properties properties, String topic, String groupId) {\n        this.topic = topic;\n        String flatMessageStr = properties.getProperty(CanalConstants.CANAL_MQ_FLAT_MESSAGE);\n        if (StringUtils.isNotEmpty(flatMessageStr)) {\n            this.flatMessage = Boolean.parseBoolean(flatMessageStr);\n        }\n        this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);\n        this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN);\n        this.listenerName = properties.getProperty(PulsarMQConstants.PULSARMQ_LISTENER_NAME);\n        this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);\n        // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称\n        if (StringUtils.isEmpty(this.subscriptName)) {\n            this.subscriptName = groupId;\n        }\n\n        if (StringUtils.isEmpty(this.subscriptName)) {\n            throw new RuntimeException(\"Pulsar Consumer subscriptName required\");\n        }\n        String batchSizeStr = properties.getProperty(CanalConstants.CANAL_MQ_CANAL_BATCH_SIZE);\n        if (StringUtils.isNotEmpty(batchSizeStr)) {\n            this.batchSize = Integer.parseInt(batchSizeStr);\n        }\n        String getBatchTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_GET_BATCH_TIMEOUT_SECONDS);\n        if (StringUtils.isNotEmpty(getBatchTimeoutSecondsStr)) {\n            this.getBatchTimeoutSeconds = Integer.parseInt(getBatchTimeoutSecondsStr);\n        }\n        String batchProcessTimeoutStr = properties.getProperty(PulsarMQConstants.PULSARMQ_BATCH_PROCESS_TIMEOUT);\n        if (StringUtils.isNotEmpty(batchProcessTimeoutStr)) {\n            this.batchProcessTimeout = Integer.parseInt(batchProcessTimeoutStr);\n        }\n        String redeliveryDelaySecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_REDELIVERY_DELAY_SECONDS);\n        if (StringUtils.isNotEmpty(redeliveryDelaySecondsStr)) {\n            this.redeliveryDelaySeconds = Integer.parseInt(redeliveryDelaySecondsStr);\n        }\n        String ackTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_ACK_TIMEOUT_SECONDS);","sourceCodeStart":109,"sourceCodeEnd":145,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java#L109-L145","documentation":"Thrown by CanalPulsarMQConsumer.init when both the PULSARMQ_SUBSCRIPT_NAME property and the groupId are empty/null. The consumer needs a non-empty subscription name to call builder.subscriptionName(...); without it Pulsar would refuse the subscribe, so the connector fails fast during initialization.","triggerScenarios":"init(properties, topic, groupId) is called with groupId null/empty AND properties does not contain PULSARMQ_SUBSCRIPT_NAME (pulsarmq.subscriptName). The fallback `subscriptName = groupId` is also empty, so the final isEmpty check at line 126 throws.","commonSituations":"Pulsar consumer started without setting canal.mq.groupId or pulsarmq.subscriptName; groupId computed dynamically and came back blank; copy-paste config that dropped the subscription property when migrating from another MQ connector.","solutions":["Set the Pulsar subscription name explicitly: properties.setProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME, \"my-subscription\").","Or ensure a non-empty groupId is passed to init() — it is used as the subscription name when the property is absent.","Validate the property at startup and fail with a clearer message before constructing the consumer."],"exampleFix":"// before\nString subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);\nif (StringUtils.isEmpty(subscriptName)) subscriptName = groupId; // groupId also empty\n\n// after — require an explicit subscription name with a clear error\nif (StringUtils.isEmpty(this.subscriptName)) {\n    throw new IllegalArgumentException(\n        \"Pulsar subscription name required: set \" + PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME\n        + \" or pass a non-empty groupId\");\n}","handlingStrategy":"validation","validationCode":"String sub = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);\nif (StringUtils.isEmpty(sub)) sub = groupId;\nif (StringUtils.isEmpty(sub)) {\n    throw new IllegalStateException(\n        \"Set \" + PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME + \" or pass a non-empty groupId\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set pulsarmq.subscriptName or canal.mq.groupId for Pulsar consumers.","Validate subscription config before constructing the consumer.","Give each logical consumer a distinct, stable subscription name."],"tags":["pulsar","configuration","canal-connector","initialization"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}