{"record":{"id":"315ec1e0ec9a10dc","repo":"alibaba/canal","slug":"disconnect-pulsar-consumer-error","errorCode":null,"errorMessage":"Disconnect pulsar consumer error","messagePattern":"Disconnect pulsar consumer error","errorType":"exception","errorClass":"CanalClientException","httpStatus":null,"severity":"warning","filePath":"connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java","lineNumber":296,"sourceCode":"            if (isConsumerActive() && hasLastMessages()) {\n                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);\n            }\n        } finally {\n            this.lastGetBatchMessage = null;\n        }\n    }\n\n    @Override\n    public void disconnect() {\n        if (null == this.pulsarMQConsumer || !this.pulsarMQConsumer.isConnected()) {\n            return;\n        }\n        try {\n            // 会导致暂停期间数据丢失\n            // this.pulsarMQConsumer.unsubscribe();\n            this.pulsarClient.close();\n        } catch (PulsarClientException e) {\n            throw new CanalClientException(\"Disconnect pulsar consumer error\", e);\n        }\n    }\n\n    /**\n     * 是否消费可用\n     *\n     * @return true消费者可用\n     */\n    private boolean isConsumerActive() {\n        return null != this.pulsarMQConsumer && this.pulsarMQConsumer.isConnected();\n    }\n\n    /**\n     * 是否有未确认消息\n     *\n     * @return true有正在消费的待确认消息\n     */\n    private boolean hasLastMessages() {","sourceCodeStart":278,"sourceCodeEnd":314,"githubUrl":"https://github.com/alibaba/canal/blob/87be50e87686a3e8af08c368d0e1ffd1f59eb04a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java#L278-L314","documentation":"Thrown by CanalPulsarMQConsumer.disconnect when pulsarClient.close() raises a PulsarClientException. disconnect first short-circuits if the consumer is null or not connected, so reaching the close() means there was an active client; the failure is during shutdown teardown and usually indicates the client was already closing or the broker connection was already broken.","triggerScenarios":"pulsarClient.close() at line 294 throws — typically because the client is already closed, the underlying connection was lost, or a concurrent close is in progress. Note disconnect closes the whole PulsarClient, not just the consumer.","commonSituations":"Calling disconnect() twice; shutdown racing with a broker-initiated disconnect; closing during a network partition where the client cannot flush; the unsubscribe call above it is commented out so lingering state can complicate close.","solutions":["Make disconnect() idempotent — catch and log PulsarClientException during close rather than propagating, since shutdown failures are rarely actionable.","Avoid calling disconnect() more than once; track a closed flag.","If you need a clean subscription teardown, consider explicitly closing the consumer (pulsarMQConsumer.close()) before closing the client."],"exampleFix":"// before\ntry {\n    this.pulsarClient.close();\n} catch (PulsarClientException e) {\n    throw new CanalClientException(\"Disconnect pulsar consumer error\", e);\n}\n\n// after — shutdown failures are logged, not fatal\ntry {\n    this.pulsarClient.close();\n} catch (PulsarClientException e) {\n    logger.warn(\"Error closing pulsar client during disconnect\", e);\n}","handlingStrategy":"try-catch","validationCode":"if (pulsarMQConsumer == null || !pulsarMQConsumer.isConnected()) return;","typeGuard":null,"tryCatchPattern":"try {\n    this.pulsarClient.close();\n} catch (PulsarClientException e) {\n    logger.warn(\"Error closing pulsar client during disconnect\", e);\n    // do not rethrow — shutdown is best-effort\n}","preventionTips":["Make disconnect idempotent — guard with a closed flag.","Log, do not throw, on shutdown-time close failures.","Close the consumer before closing the client for cleaner teardown."],"tags":["pulsar","shutdown","canal-connector","lifecycle"],"backgroundTag":null,"analyzedSha":"87be50e87686a3e8af08c368d0e1ffd1f59eb04a","analyzedAt":"2026-08-14T04:30:11.918Z","schemaVersion":2},"datasetVersion":"2026-08-14T05:17:29.042Z"}