{"record":{"id":"937d9174673f9bdd","repo":"apache/seatunnel","slug":"error-closing-mqtt-source-client","errorCode":null,"errorMessage":"Error closing MQTT source client","messagePattern":"Error closing MQTT source client","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceReader.java","lineNumber":143,"sourceCode":"\n    @Override\n    public void close() throws IOException {\n        if (mqttClient == null) {\n            return;\n        }\n        try {\n            if (mqttClient.isConnected()) {\n                if (sourceConfig.isCleanSession()) {\n                    mqttClient.unsubscribe(sourceConfig.getTopic());\n                }\n                mqttClient.disconnect();\n            } else {\n                mqttClient.disconnectForcibly();\n            }\n            mqttClient.close();\n            LOG.info(\"MQTT source reader [{}] closed\", sourceConfig.getClientId());\n        } catch (MqttException e) {\n            throw new IOException(\"Error closing MQTT source client\", e);\n        }\n    }\n\n    @Override\n    public void connectionLost(Throwable cause) {\n        disconnectedSinceMs = currentTimeMillis.getAsLong();\n        disconnectCause = cause;\n        LOG.warn(\n                \"MQTT source connection lost for client [{}], auto-reconnect will attempt recovery\",\n                sourceConfig.getClientId(),\n                cause);\n    }\n\n    @Override\n    public void connectComplete(boolean reconnect, String serverURI) {\n        if (!reconnect) {\n            return;\n        }","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceReader.java#L125-L161","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Inspect the wrapped MqttException cause: reason code 32110 (client disconnected) or 'already closed' is benign and can be ignored","Check broker availability/logs to see why the connection was already broken at close time","Avoid calling close() twice; guard with an idempotent close flag","Upgrade the Paho client version if close() spuriously throws on disconnected clients"],"exampleFix":"// before\ntry {\n    mqttClient.disconnectForcibly();\n    mqttClient.close();\n} catch (MqttException e) {\n    throw new IOException(\"Error closing MQTT source client\", e);\n}\n// after\ntry {\n    if (mqttClient.isConnected()) {\n        mqttClient.disconnectForcibly();\n    }\n    mqttClient.close();\n} catch (MqttException e) {\n    LOG.warn(\"MQTT client close failed for [{}]\", sourceConfig.getClientId(), e);\n}","handlingStrategy":"try-catch","validationCode":"// before closing\nif (client != null && client.isConnected()) { /* disconnect path */ }\nelse { /* skip disconnect, just close */ }","typeGuard":null,"tryCatchPattern":"try {\n    reader.close();\n} catch (IOException e) {\n    if (e.getCause() instanceof MqttException\n            && (32110 == ((MqttException) e.getCause()).getReasonCode())) {\n        // already disconnected — safe to ignore\n    } else {\n        throw e;\n    }\n}","preventionTips":["Make close() idempotent with a closed flag","Check isConnected() before disconnectForcibly","Treat reason code 32110 / already-closed as benign during teardown","Monitor broker uptime to correlate close failures with outages"],"tags":["mqtt","io","cleanup","connection"],"backgroundTag":"connection-closed","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"}