{"record":{"id":"a1692d0f6d7dbd81","repo":"apache/seatunnel","slug":"error-closing-mqtt-client","errorCode":null,"errorMessage":"Error closing MQTT client","messagePattern":"Error closing MQTT client","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java","lineNumber":155,"sourceCode":"    @Override\n    public void abortPrepare() {\n        // Stateless sink — nothing to roll back.\n    }\n\n    @Override\n    public void close() throws IOException {\n        try {\n            flushBuffer();\n        } finally {\n            if (mqttClient != null) {\n                try {\n                    if (mqttClient.isConnected()) {\n                        mqttClient.disconnect();\n                    }\n                    mqttClient.close();\n                    log.info(\"MQTT sink writer closed\");\n                } catch (MqttException e) {\n                    throw new IOException(\"Error closing MQTT client\", e);\n                }\n            }\n        }\n    }\n\n    // ---- MqttCallback implementation ----\n\n    @Override\n    public void connectionLost(Throwable cause) {\n        // Auto-reconnect is enabled; log for observability but do not throw.\n        log.warn(\"MQTT connection lost, auto-reconnect will attempt recovery\", cause);\n    }\n\n    @Override\n    public void messageArrived(String topic, MqttMessage message) {\n        // Sink-only client — inbound messages are not expected.\n    }\n","sourceCodeStart":137,"sourceCodeEnd":173,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java#L137-L173","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Check the cause MqttException reason; a lost-connection error (32109) during shutdown is usually harmless.","Verify broker stability and keep-alive settings; reduce reconnect/timeout windows so close() completes cleanly.","If the exception occurs only at shutdown with all data flushed, it can be safely downgraded/tolerated — all writes already succeeded.","Upgrade the connector/Paho client if disconnect races are reproducible."],"exampleFix":"// before (cleanup throws at shutdown)\n} catch (MqttException e) {\n    throw new IOException(\"Error closing MQTT client\", e);\n}\n// after (user-side): ensure job stops gracefully so close() runs while connected\n// keep-alive = 60 (broker-side tuning) and stop the job without SIGKILL","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    writer.close();\n} catch (IOException e) {\n    if (e.getMessage().contains(\"Error closing MQTT client\")\n        && e.getCause() instanceof MqttException\n        && ((MqttException) e.getCause()).getReasonCode() == MqttException.REASON_CODE_CLIENT_DISCONNECTING) {\n        // benign race during shutdown; log and continue\n        return;\n    }\n    throw e;\n}","preventionTips":["Let jobs shut down gracefully so close() runs on a live connection.","Tune keep-alive/reconnect timeouts to avoid disconnect races.","Treat lost-connection (32109) during shutdown as benign when data is flushed.","Keep Paho client and connector versions current."],"tags":["mqtt","connection","shutdown","cleanup"],"backgroundTag":"connection-refused","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}