apache/seatunnel · warning · IOException

Error closing MQTT client

Error message

Error closing MQTT client

What it means

MqttSinkWriter.close() disconnects and closes the Paho MQTT client and wraps any MqttException raised during that cleanup in an IOException. This surfaces when disconnect() or close() fails — typically because the connection is in an unexpected state (already lost, mid-reconnect) or the broker drops the socket during disconnect.

Source

Thrown at seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java:155

    @Override
    public void abortPrepare() {
        // Stateless sink — nothing to roll back.
    }

    @Override
    public void close() throws IOException {
        try {
            flushBuffer();
        } finally {
            if (mqttClient != null) {
                try {
                    if (mqttClient.isConnected()) {
                        mqttClient.disconnect();
                    }
                    mqttClient.close();
                    log.info("MQTT sink writer closed");
                } catch (MqttException e) {
                    throw new IOException("Error closing MQTT client", e);
                }
            }
        }
    }

    // ---- MqttCallback implementation ----

    @Override
    public void connectionLost(Throwable cause) {
        // Auto-reconnect is enabled; log for observability but do not throw.
        log.warn("MQTT connection lost, auto-reconnect will attempt recovery", cause);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        // Sink-only client — inbound messages are not expected.
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Check the cause MqttException reason; a lost-connection error (32109) during shutdown is usually harmless.
2. Verify broker stability and keep-alive settings; reduce reconnect/timeout windows so close() completes cleanly.
3. If the exception occurs only at shutdown with all data flushed, it can be safely downgraded/tolerated — all writes already succeeded.
4. Upgrade the connector/Paho client if disconnect races are reproducible.

Example fix

// before (cleanup throws at shutdown)
} catch (MqttException e) {
    throw new IOException("Error closing MQTT client", e);
}
// after (user-side): ensure job stops gracefully so close() runs while connected
// keep-alive = 60 (broker-side tuning) and stop the job without SIGKILL

try {
    writer.close();
} catch (IOException e) {
    if (e.getMessage().contains("Error closing MQTT client")
        && e.getCause() instanceof MqttException
        && ((MqttException) e.getCause()).getReasonCode() == MqttException.REASON_CODE_CLIENT_DISCONNECTING) {
        // benign race during shutdown; log and continue
        return;
    }
    throw e;
}

When it happens

Trigger: Calling close() (job shutdown, checkpoint-anchored close, failover) where mqttClient.disconnect() or mqttClient.close() throws MqttException; the catch rethrows as IOException("Error closing MQTT client", e).

Common situations: Broker already closed the TCP connection before disconnect; task cancelled while client is reconnecting; long keep-alive timeouts making disconnect block and fail; network flap during job shutdown.

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• Error closing MQTT source client
• IotdbConnectorErrorCode.CLOSE_SESSION_FAILED
• Failed to close %s via JDBC.
• MqttConnectorErrorCode.CONNECTION_FAILED
• Failed to close Pulsar admin.

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