{"record":{"id":"4c10b74ece3a0754","repo":"apache/seatunnel","slug":"mqtt-source-connection-lost-for-client-auto","errorCode":null,"errorMessage":"MQTT source connection lost for client [{}], auto-reconnect will attempt recovery","messagePattern":"MQTT source connection lost for client \\[(.+?)\\], auto-reconnect will attempt recovery","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceReader.java","lineNumber":151,"sourceCode":"                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        }\n        try {\n            subscribeTopic();\n            disconnectedSinceMs = -1L;\n            disconnectCause = null;\n            LOG.info(\n                    \"MQTT source reader [{}] resubscribed to topic [{}] after reconnect to [{}]\",\n                    sourceConfig.getClientId(),\n                    sourceConfig.getTopic(),","sourceCodeStart":133,"sourceCodeEnd":169,"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#L133-L169","documentation":"The MQTT source reader's connectionLost callback fired after the broker dropped the connection. The reader records disconnectedSinceMs and disconnectCause, then logs this warning; Paho auto-reconnect is expected to restore the connection and resubscribe. If reconnect does not succeed within the configured timeout, the reader fails the task (see tests referencing reconnect timeout).","triggerScenarios":"MqttSourceReader.connectionLost(Throwable) is invoked by Paho when the source client's connection drops; it logs with the configured clientId.","commonSituations":"Broker restart or crash; keep-alive timeout on idle networks; duplicate clientId causing broker kick; TLS/network issues; NAT idle timeouts.","solutions":["Confirm auto-reconnect restores the session; if the task eventually fails, check the reconnect timeout setting and increase it.","Use a unique clientId per source reader instance to avoid broker kick loops.","Inspect the logged disconnectCause (often reason code 32109 / socket loss) to fix the network or broker issue.","Tune keepAlive interval to tolerate network latency; use ssl:// for unstable links.","Verify broker-side connection limits aren't evicting clients."],"exampleFix":"// before\nsource {\n  Mqtt {\n    client_id = \"seatunnel-reader\"\n  }\n}\n// after\nsource {\n  Mqtt {\n    client_id = \"seatunnel-reader-${subtask_index}\"\n  }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"@Override\npublic void connectionLost(Throwable cause) {\n    disconnectedSinceMs = System.currentTimeMillis();\n    disconnectCause = cause;\n    LOG.warn(\"MQTT connection lost, auto-reconnect in progress\", cause);\n}","preventionTips":["Give each source reader a unique clientId","Increase reconnect timeout if brokers take long to recover","Log and alert on disconnectCause patterns (32109 = socket loss)","Use keepAlive and TLS tuned to your network's idle-timeout behavior"],"tags":["mqtt","connection-lost","reconnect","source"],"backgroundTag":"connection-lost","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}