{"record":{"id":"7944be34c41bb8ea","repo":"apache/seatunnel","slug":"unsupported-mqtt-source-format-format","errorCode":null,"errorMessage":"Unsupported MQTT source format: ${format}","messagePattern":"Unsupported MQTT source format: (.+?)","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java","lineNumber":74,"sourceCode":"        String configuredClientId = config.get(MqttSourceOptions.CLIENT_ID);\n        if (!cleanSession && isBlank(configuredClientId)) {\n            throw new IllegalArgumentException(\n                    \"client_id is required when clean_session=false for MQTT source\");\n        }\n        this.clientId =\n                isBlank(configuredClientId)\n                        ? CLIENT_ID_PREFIX + UUID.randomUUID().toString()\n                        : configuredClientId;\n\n        validate();\n    }\n\n    private void validate() {\n        if (qos < 0 || qos > 1) {\n            throw new IllegalArgumentException(\"MQTT source qos must be 0 or 1, got: \" + qos);\n        }\n        if (!\"json\".equalsIgnoreCase(format) && !\"text\".equalsIgnoreCase(format)) {\n            throw new IllegalArgumentException(\"Unsupported MQTT source format: \" + format);\n        }\n        if (reconnectTimeout <= 0) {\n            throw new IllegalArgumentException(\n                    \"reconnect_timeout must be greater than 0, got: \" + reconnectTimeout);\n        }\n        if (maxQueueSize <= 0) {\n            throw new IllegalArgumentException(\n                    \"max_queue_size must be greater than 0, got: \" + maxQueueSize);\n        }\n    }\n\n    private static boolean isBlank(String value) {\n        return value == null || value.trim().isEmpty();\n    }\n\n    public String getUrl() {\n        return url;\n    }","sourceCodeStart":56,"sourceCodeEnd":92,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSourceConfig.java#L56-L92","documentation":"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.","triggerScenarios":"Configuring the MQTT source with format = csv, xml, or a misspelled value; validate() runs during MqttSourceConfig construction and fails immediately.","commonSituations":"Publishing binary or CSV payloads and hoping the connector handles them; format names copied from other connectors (e.g. canal, avro); typos like 'jsion'.","solutions":["Set format = \"json\" or format = \"text\" in the source config","If payload is another format, add a transform or custom deserialization before/after, or publish as JSON from the producer","Match casing freely — comparison is case-insensitive, but the value itself must be json or text"],"exampleFix":"// before\nMqtt {\n  format = csv\n}\n// after\nMqtt {\n  format = json\n}","handlingStrategy":"validation","validationCode":"if (!\"json\".equalsIgnoreCase(format) && !\"text\".equalsIgnoreCase(format)) {\n  throw new IllegalArgumentException(\"Unsupported MQTT source format: \" + format);\n}","typeGuard":"null","tryCatchPattern":"null","preventionTips":["Use only json or text for MQTT source payloads","Convert other payload formats at the producer side","Add schema/enum validation to config templates"],"tags":["mqtt","format","validation","config"],"backgroundTag":"invalid-enum-value","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}