{"record":{"id":"e92ac2ea9696c60a","repo":"alibaba/canal","slug":"receive-pulsar-batch-message-error","errorCode":null,"errorMessage":"Receive pulsar batch message error","messagePattern":"Receive pulsar batch message 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":252,"sourceCode":"            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);\n                    List<CommonMessage> list = MessageUtil.convert(message);\n                    messageList.addAll(list);\n                } else {\n                    CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);\n                    messageList.add(commonMessage);\n                }\n            }\n        } catch (PulsarClientException e) {\n            throw new CanalClientException(\"Receive pulsar batch message error\", e);\n        }\n\n        return messageList;\n    }\n\n    @Override\n    public void rollback() {\n        try {\n            if (isConsumerActive() && hasLastMessages()) {\n                // 回滚所有消息\n                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);\n            }\n        } finally {\n            this.lastGetBatchMessage = null;\n        }\n    }\n\n    @Override","sourceCodeStart":234,"sourceCodeEnd":270,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java#L234-L270","documentation":"Thrown by CanalPulsarMQConsumer.getMessage when pulsarMQConsumer.batchReceive() raises a PulsarClientException during steady-state consumption. Unlike subscribe-time errors, this happens after the consumer is active, so the usual cause is a runtime broker/network problem or the consumer having been closed/Disconnected underneath the caller.","triggerScenarios":"pulsarMQConsumer.batchReceive() at line 234 throws. Causes: consumer was closed (e.g. disconnect() ran concurrently); broker connection lost and not yet recovered; topic/subscription forcibly terminated by the broker; authorization revoked mid-stream.","commonSituations":"Long-running consumer where the broker restarted or dropped the connection; concurrent disconnect() during getMessage; network partition; broker-side topic unload causing a transient failure before the client auto-reconnects.","solutions":["Treat this as retriable: catch CanalClientException, wait briefly, and re-call getMessage (the Pulsar client attempts auto-reconnect).","Ensure getMessage and disconnect are not called concurrently; guard with the isConsumerActive() check.","If it recurs, verify broker health and that pulsarmq.server.url still resolves/reaches.","Re-init the consumer (connect()) if isConsumerActive() returns false."],"exampleFix":"// before\npublic List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {\n    try {\n        Messages<byte[]> messages = pulsarMQConsumer.batchReceive();\n        ...\n    } catch (PulsarClientException e) {\n        throw new CanalClientException(\"Receive pulsar batch message error\", e);\n    }\n}\n\n// after — retry once on transient failure, re-init if inactive\nint attempts = 0;\nwhile (attempts++ < 2) {\n    try {\n        if (!isConsumerActive()) { connect(); }\n        Messages<byte[]> messages = pulsarMQConsumer.batchReceive();\n        ...\n        return messageList;\n    } catch (PulsarClientException e) {\n        if (attempts >= 2) throw new CanalClientException(\"Receive pulsar batch message error\", e);\n    }\n}","handlingStrategy":"retry","validationCode":"if (!isConsumerActive()) {\n    connect(); // re-init if the consumer was closed\n}","typeGuard":"boolean consumerReady() {\n    return pulsarMQConsumer != null && pulsarMQConsumer.isConnected();\n}","tryCatchPattern":"try {\n    Messages<byte[]> messages = pulsarMQConsumer.batchReceive();\n    ...\n} catch (PulsarClientException e) {\n    if (!isConsumerActive()) connect();\n    throw new CanalClientException(\"Receive pulsar batch message error\", e);\n}","preventionTips":["Do not call getMessage concurrently with disconnect().","Re-init the consumer if isConsumerActive() returns false.","Treat batchReceive failures as retriable for transient broker issues."],"tags":["pulsar","network","consumer","canal-connector","runtime"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}