{"record":{"id":"80def56697bc532d","repo":"apache/seatunnel","slug":"mqttconnectorerrorcode-connection-failed","errorCode":"MqttConnectorErrorCode.CONNECTION_FAILED","errorMessage":"Failed to connect MQTT client [","messagePattern":"Failed to connect MQTT client \\[","errorType":"error_code","errorClass":"MqttConnectorException","httpStatus":null,"severity":"critical","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java","lineNumber":112,"sourceCode":"                            clientId,\n                            new MemoryPersistence());\n            this.mqttClient.setCallback(this);\n\n            MqttConnectOptions options = buildConnectOptions(pluginConfig);\n            this.mqttClient.connect(options);\n            log.info(\n                    \"MQTT sink writer [{}] connected to {}\",\n                    clientId,\n                    pluginConfig.get(MqttSinkOptions.URL));\n        } catch (MqttException e) {\n            if (this.mqttClient != null) {\n                try {\n                    this.mqttClient.close();\n                } catch (MqttException ignored) {\n                    // Best-effort cleanup; the original exception is more important.\n                }\n            }\n            throw new MqttConnectorException(\n                    MqttConnectorErrorCode.CONNECTION_FAILED,\n                    \"Failed to connect MQTT client [\" + clientId + \"]\",\n                    e);\n        }\n    }\n\n    @Override\n    public void write(SeaTunnelRow element) throws IOException {\n        byte[] payload = serializationSchema.serialize(element);\n        MqttMessage message = new MqttMessage(payload);\n        message.setQos(qos);\n\n        messageBuffer.add(message);\n        if (messageBuffer.size() >= batchSize) {\n            flushBuffer();\n        }\n    }\n","sourceCodeStart":94,"sourceCodeEnd":130,"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#L94-L130","documentation":"MqttConnectorException with CONNECTION_FAILED is thrown when the Paho MQTT client's connect() call fails during MqttSinkWriter construction. On failure the writer best-effort closes the partially-created client and rethrows with the clientId for diagnosis; the underlying MqttException is attached as cause (e.g. broker unreachable, auth failure, client ID collision).","triggerScenarios":"Constructor calls mqttClient.connect() (often with a timeout/retry helper) and the broker rejects or times out the connection; the catch block closes the client and throws CONNECTION_FAILED.","commonSituations":"Wrong broker host/port; broker down or firewalled; bad username/password or missing TLS config; duplicate clientId when another client uses the same generated ID (e.g. multiple jobs sharing config); SSL handshake issues.","solutions":["Read the cause MqttException reason code (e.g. 3 = server unavailable, 4/5 = bad credentials).","Verify host/port and network reachability from the SeaTunnel worker (telnet/nc to the broker port).","Check username/password and TLS/SSL settings (server URIs starting ssl:// need truststore configuration).","Ensure clientIds are unique — each writer appends subtask index + UUID, but brokers with restrictions or shared IDs can reject connections.","Confirm the broker allows the QoS and protocol version configured."],"exampleFix":"// before\nuri = \"tcp://broker.internal:1884\"\n// after (correct host/port, reachable and with auth)\nuri = \"tcp://broker.internal:1883\"\nusername = \"mqtt-user\"\npassword = \"********\"","handlingStrategy":"validation","validationCode":"// Validate broker reachability before job start:\n// (echo > /dev/tcp/broker-host/1883) && echo reachable || echo unreachable\n// and verify credentials with an MQTT client (e.g. mosquitto_pub -h host -p 1883 -u user -P pass)","typeGuard":null,"tryCatchPattern":"try {\n    new MqttSinkWriter(context, rowType, pluginConfig);\n} catch (MqttConnectorException e) {\n    if (\"CONNECTION_FAILED\".equals(e.getErrorCode().name())) {\n        // log e.getCause() (MqttException reason code) and retry with backoff or fail fast\n    }\n    throw e;\n}","preventionTips":["Pre-flight check broker host/port and auth from every worker.","Use unique clientIds per job (the writer appends subtask id + UUID).","Match TLS settings (ssl:// vs tcp://) with broker requirements.","Confirm broker connection limits/ACLs allow the SeaTunnel user."],"tags":["mqtt","connection","broker","authentication"],"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"}