apache/seatunnel · critical · MqttConnectorException

MqttConnectorErrorCode.CONNECTION_FAILED

MqttConnectorErrorCode.CONNECTION_FAILED

Error message

Failed to connect MQTT client [

What it means

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).

Source

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

                            clientId,
                            new MemoryPersistence());
            this.mqttClient.setCallback(this);

            MqttConnectOptions options = buildConnectOptions(pluginConfig);
            this.mqttClient.connect(options);
            log.info(
                    "MQTT sink writer [{}] connected to {}",
                    clientId,
                    pluginConfig.get(MqttSinkOptions.URL));
        } catch (MqttException e) {
            if (this.mqttClient != null) {
                try {
                    this.mqttClient.close();
                } catch (MqttException ignored) {
                    // Best-effort cleanup; the original exception is more important.
                }
            }
            throw new MqttConnectorException(
                    MqttConnectorErrorCode.CONNECTION_FAILED,
                    "Failed to connect MQTT client [" + clientId + "]",
                    e);
        }
    }

    @Override
    public void write(SeaTunnelRow element) throws IOException {
        byte[] payload = serializationSchema.serialize(element);
        MqttMessage message = new MqttMessage(payload);
        message.setQos(qos);

        messageBuffer.add(message);
        if (messageBuffer.size() >= batchSize) {
            flushBuffer();
        }
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Read the cause MqttException reason code (e.g. 3 = server unavailable, 4/5 = bad credentials).
2. Verify host/port and network reachability from the SeaTunnel worker (telnet/nc to the broker port).
3. Check username/password and TLS/SSL settings (server URIs starting ssl:// need truststore configuration).
4. Ensure clientIds are unique — each writer appends subtask index + UUID, but brokers with restrictions or shared IDs can reject connections.
5. Confirm the broker allows the QoS and protocol version configured.

Example fix

// before
uri = "tcp://broker.internal:1884"
// after (correct host/port, reachable and with auth)
uri = "tcp://broker.internal:1883"
username = "mqtt-user"
password = "********"

// Validate broker reachability before job start:
// (echo > /dev/tcp/broker-host/1883) && echo reachable || echo unreachable
// and verify credentials with an MQTT client (e.g. mosquitto_pub -h host -p 1883 -u user -P pass)

try {
    new MqttSinkWriter(context, rowType, pluginConfig);
} catch (MqttConnectorException e) {
    if ("CONNECTION_FAILED".equals(e.getErrorCode().name())) {
        // log e.getCause() (MqttException reason code) and retry with backoff or fail fast
    }
    throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

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

• CONNECTION_FAILED
• Error closing MQTT client
• Error closing MQTT source client
• CREATE_ACTIVEMQ_CLIENT_FAILED
• AmazonDocumentDB option 'uri' must include authentication cr

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