What causes "Disconnect pulsar consumer error" in alibaba/canal?

Make disconnect() idempotent — catch and log PulsarClientException during close rather than propagating, since shutdown failures are rarely actionable.

alibaba/canal · warning · CanalClientException

Disconnect pulsar consumer error

Error message

Disconnect pulsar consumer error

What it means

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.

Source

Thrown at connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java:296

            if (isConsumerActive() && hasLastMessages()) {
                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void disconnect() {
        if (null == this.pulsarMQConsumer || !this.pulsarMQConsumer.isConnected()) {
            return;
        }
        try {
            // 会导致暂停期间数据丢失
            // this.pulsarMQConsumer.unsubscribe();
            this.pulsarClient.close();
        } catch (PulsarClientException e) {
            throw new CanalClientException("Disconnect pulsar consumer error", e);
        }
    }

    /**
     * 是否消费可用
     *
     * @return true消费者可用
     */
    private boolean isConsumerActive() {
        return null != this.pulsarMQConsumer && this.pulsarMQConsumer.isConnected();
    }

    /**
     * 是否有未确认消息
     *
     * @return true有正在消费的待确认消息
     */
    private boolean hasLastMessages() {

View on GitHub (pinned to 87be50e876)

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.

Example fix

// before
try {
    this.pulsarClient.close();
} catch (PulsarClientException e) {
    throw new CanalClientException("Disconnect pulsar consumer error", e);
}

// after — shutdown failures are logged, not fatal
try {
    this.pulsarClient.close();
} catch (PulsarClientException e) {
    logger.warn("Error closing pulsar client during disconnect", e);
}

Defensive patterns

Strategy: try-catch

Validate before calling

if (pulsarMQConsumer == null || !pulsarMQConsumer.isConnected()) return;

Try / catch

try {
    this.pulsarClient.close();
} catch (PulsarClientException e) {
    logger.warn("Error closing pulsar client during disconnect", e);
    // do not rethrow — shutdown is best-effort
}

Prevention

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.

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of alibaba/canal@87be50e876 (2026-08-14). Data as JSON: /api/errors/315ec1e0ec9a10dc. Report an issue: GitHub.