apache/seatunnel · error · IllegalArgumentException

Unsupported MQTT source format: ${format}

Error message

Unsupported MQTT source format: ${format}

What it means

MqttSourceConfig.validate throws IllegalArgumentException when format is neither 'json' nor 'text' (case-insensitive). The source can only deserialize payloads as JSON or plain text, so any other format fails at startup. This is the source-side counterpart of the sink's format switch.

Source

Thrown at seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java:74

        String configuredClientId = config.get(MqttSourceOptions.CLIENT_ID);
        if (!cleanSession && isBlank(configuredClientId)) {
            throw new IllegalArgumentException(
                    "client_id is required when clean_session=false for MQTT source");
        }
        this.clientId =
                isBlank(configuredClientId)
                        ? CLIENT_ID_PREFIX + UUID.randomUUID().toString()
                        : configuredClientId;

        validate();
    }

    private void validate() {
        if (qos < 0 || qos > 1) {
            throw new IllegalArgumentException("MQTT source qos must be 0 or 1, got: " + qos);
        }
        if (!"json".equalsIgnoreCase(format) && !"text".equalsIgnoreCase(format)) {
            throw new IllegalArgumentException("Unsupported MQTT source format: " + format);
        }
        if (reconnectTimeout <= 0) {
            throw new IllegalArgumentException(
                    "reconnect_timeout must be greater than 0, got: " + reconnectTimeout);
        }
        if (maxQueueSize <= 0) {
            throw new IllegalArgumentException(
                    "max_queue_size must be greater than 0, got: " + maxQueueSize);
        }
    }

    private static boolean isBlank(String value) {
        return value == null || value.trim().isEmpty();
    }

    public String getUrl() {
        return url;
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Set format = "json" or format = "text" in the source config
2. If payload is another format, add a transform or custom deserialization before/after, or publish as JSON from the producer
3. Match casing freely — comparison is case-insensitive, but the value itself must be json or text

Example fix

// before
Mqtt {
  format = csv
}
// after
Mqtt {
  format = json
}

if (!"json".equalsIgnoreCase(format) && !"text".equalsIgnoreCase(format)) {
  throw new IllegalArgumentException("Unsupported MQTT source format: " + format);
}

null

null

When it happens

Trigger: Configuring the MQTT source with format = csv, xml, or a misspelled value; validate() runs during MqttSourceConfig construction and fails immediately.

Common situations: Publishing binary or CSV payloads and hoping the connector handles them; format names copied from other connectors (e.g. canal, avro); typos like 'jsion'.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• Invalid endpoint: %s, expected format host:port
• Invalid endpoint port in endpoint: %s
• Unsupported MQTT sink format:
• client_id is required when clean_session=false for MQTT sour
• MQTT source qos must be 0 or 1, got: ${qos}

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