apache/seatunnel · warning · IOException

Error closing MQTT source client

Error message

Error closing MQTT source client

What it means

MqttSourceReader.close() tears down the Paho MQTT client: it disconnects (forcibly if needed) and calls mqttClient.close(). If any MqttException is thrown during teardown, the reader wraps it in an IOException with this message. It indicates the client session could not be cleanly released, often after an already-broken connection.

Source

Thrown at seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceReader.java:143

    @Override
    public void close() throws IOException {
        if (mqttClient == null) {
            return;
        }
        try {
            if (mqttClient.isConnected()) {
                if (sourceConfig.isCleanSession()) {
                    mqttClient.unsubscribe(sourceConfig.getTopic());
                }
                mqttClient.disconnect();
            } else {
                mqttClient.disconnectForcibly();
            }
            mqttClient.close();
            LOG.info("MQTT source reader [{}] closed", sourceConfig.getClientId());
        } catch (MqttException e) {
            throw new IOException("Error closing MQTT source client", e);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        disconnectedSinceMs = currentTimeMillis.getAsLong();
        disconnectCause = cause;
        LOG.warn(
                "MQTT source connection lost for client [{}], auto-reconnect will attempt recovery",
                sourceConfig.getClientId(),
                cause);
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if (!reconnect) {
            return;
        }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Inspect the wrapped MqttException cause: reason code 32110 (client disconnected) or 'already closed' is benign and can be ignored
2. Check broker availability/logs to see why the connection was already broken at close time
3. Avoid calling close() twice; guard with an idempotent close flag
4. Upgrade the Paho client version if close() spuriously throws on disconnected clients

Example fix

// before
try {
    mqttClient.disconnectForcibly();
    mqttClient.close();
} catch (MqttException e) {
    throw new IOException("Error closing MQTT source client", e);
}
// after
try {
    if (mqttClient.isConnected()) {
        mqttClient.disconnectForcibly();
    }
    mqttClient.close();
} catch (MqttException e) {
    LOG.warn("MQTT client close failed for [{}]", sourceConfig.getClientId(), e);
}

// before closing
if (client != null && client.isConnected()) { /* disconnect path */ }
else { /* skip disconnect, just close */ }

try {
    reader.close();
} catch (IOException e) {
    if (e.getCause() instanceof MqttException
            && (32110 == ((MqttException) e.getCause()).getReasonCode())) {
        // already disconnected — safe to ignore
    } else {
        throw e;
    }
}

When it happens

Trigger: Calling close() when mqttClient.disconnect() or disconnectForcibly() or mqttClient.close() throws MqttException — e.g. closing a reader whose broker connection is already dead, or closing twice concurrently.

Common situations: Broker restarted or network dropped before checkpoint-time close; job cancellation racing with connection loss; double-close of the source reader; Paho client internal state corrupted after connectionLost.

Related errors

• Error closing MQTT client
• Unable to delete directory " + localFileDir
• File check failed
• SftpException wrapped (pwd failed while deleting)
• SftpException wrapped (pwd failed while listing status)

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/937d9174673f9bdd. Report an issue: GitHub.