{"record":{"id":"4c284b4f043b45fd","repo":"apache/seatunnel","slug":"mqtt-qos-must-be-0-at-most-once-or-1-at-least-o","errorCode":null,"errorMessage":"MQTT QoS must be 0 (at-most-once) or 1 (at-least-once), got: ","messagePattern":"MQTT QoS must be 0 \\(at-most-once\\) or 1 \\(at-least-once\\), got: ","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/sink/MqttSinkWriter.java","lineNumber":69,"sourceCode":"public class MqttSinkWriter implements SinkWriter<SeaTunnelRow, Void, Void>, MqttCallback {\n\n    private static final String CLIENT_ID_PREFIX = \"seatunnel_mqtt_sink_task_\";\n    private static final long RETRY_BACKOFF_MS = 200L;\n\n    private final String topic;\n    private final int qos;\n    private final int retryTimeoutMs;\n    private final int batchSize;\n    private final SerializationSchema serializationSchema;\n    private final List<MqttMessage> messageBuffer;\n    private MqttClient mqttClient;\n\n    public MqttSinkWriter(\n            SinkWriter.Context context, SeaTunnelRowType rowType, ReadonlyConfig pluginConfig) {\n        this.topic = pluginConfig.get(MqttSinkOptions.TOPIC);\n        this.qos = pluginConfig.get(MqttSinkOptions.QOS);\n        if (this.qos < 0 || this.qos > 1) {\n            throw new IllegalArgumentException(\n                    \"MQTT QoS must be 0 (at-most-once) or 1 (at-least-once), got: \" + this.qos);\n        }\n        this.retryTimeoutMs = pluginConfig.get(MqttSinkOptions.RETRY_TIMEOUT);\n        this.batchSize = pluginConfig.get(MqttSinkOptions.BATCH_SIZE);\n        if (this.batchSize < 1) {\n            throw new IllegalArgumentException(\"batch_size must be >= 1, got: \" + this.batchSize);\n        }\n        this.messageBuffer = new ArrayList<>(this.batchSize);\n        this.serializationSchema = createSerializationSchema(rowType, pluginConfig);\n\n        // Each subtask appends its index and a random UUID to guarantee a globally unique client\n        // ID,\n        // preventing mutual disconnections and connection hijacking when running parallel jobs.\n        String clientId =\n                CLIENT_ID_PREFIX\n                        + context.getIndexOfSubtask()\n                        + \"-\"\n                        + java.util.UUID.randomUUID().toString();","sourceCodeStart":51,"sourceCodeEnd":87,"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#L51-L87","documentation":"The MQTT sink writer validates the configured qos option in its constructor and rejects values other than 0 or 1 with an IllegalArgumentException. The sink deliberately forbids QoS 2 (exactly-once) and negative values because its publish/ack handling only supports at-most-once and at-least-once delivery.","triggerScenarios":"Setting `qos = 2`, `qos = -1`, or any value outside [0,1] in the MQTT sink configuration; MqttSinkWriter constructor reads MqttSinkOptions.QOS and fails validation.","commonSituations":"Users copying QoS 2 settings from other MQTT tools; assuming the sink supports exactly-once; typos producing out-of-range numbers.","solutions":["Set qos = 0 for at-most-once delivery.","Set qos = 1 for at-least-once delivery (recommended for sinks).","Remove the qos option to use the default instead of an unsupported value.","If exactly-once is required, MQTT QoS 2 is not supported by this sink — deduplicate downstream."],"exampleFix":"// before\nMqttSink {\n  qos = 2\n}\n// after\nMqttSink {\n  qos = 1\n}","handlingStrategy":"validation","validationCode":"int qos = config.getInt(\"qos\", 1);\nif (qos < 0 || qos > 1) {\n    throw new IllegalArgumentException(\"qos must be 0 or 1, got: \" + qos);\n}","typeGuard":null,"tryCatchPattern":"try {\n    new MqttSinkWriter(context, rowType, pluginConfig);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().startsWith(\"MQTT QoS must be\")) {\n        // fall back to qos = 1 or surface config error to user\n    }\n    throw e;\n}","preventionTips":["Only set qos to 0 or 1 in MQTT sink configs.","Document that QoS 2 is unsupported by this sink.","Validate the whole sink config at job-design time."],"tags":["mqtt","config-validation","qos"],"backgroundTag":"invalid-config-value","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"}