{"record":{"id":"4d7f3cb8c0d3c9e8","repo":"apache/seatunnel","slug":"batch-size-must-be-1-got","errorCode":null,"errorMessage":"batch_size must be >= 1, got: ","messagePattern":"batch_size must be >= 1, 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":75,"sourceCode":"    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();\n\n        try {\n            // MemoryPersistence avoids file-system I/O; ideal for containerized deployments.\n            this.mqttClient =\n                    new MqttClient(\n                            pluginConfig.get(MqttSinkOptions.URL),","sourceCodeStart":57,"sourceCodeEnd":93,"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#L57-L93","documentation":"The MQTT sink writer validates the batch_size option in its constructor and rejects values below 1 with an IllegalArgumentException. batch_size controls how many messages are buffered before flushing; zero or negative values would break the batching loop.","triggerScenarios":"Setting `batch_size = 0` or a negative number in the MQTT sink configuration; the constructor reads MqttSinkOptions.BATCH_SIZE and fails the `batchSize < 1` check.","commonSituations":"Users trying to disable batching by setting batch_size = 0; copy-paste config errors; template placeholders left unfilled.","solutions":["Set batch_size >= 1 (e.g. 10–100 depending on throughput).","To send messages immediately, use batch_size = 1 rather than 0.","Remove the option to fall back to the default batch size."],"exampleFix":"// before\nbatch_size = 0\n// after\nbatch_size = 1","handlingStrategy":"validation","validationCode":"int batchSize = config.getInt(\"batch_size\", 1000);\nif (batchSize < 1) {\n    throw new IllegalArgumentException(\"batch_size must be >= 1, got: \" + batchSize);\n}","typeGuard":null,"tryCatchPattern":"try {\n    new MqttSinkWriter(context, rowType, pluginConfig);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage().startsWith(\"batch_size must be >= 1\")) {\n        // coerce to default batch size or abort job config load\n    }\n    throw e;\n}","preventionTips":["Set batch_size >= 1; use 1 for unbuffered publishing.","Never use 0 to 'disable' batching — it is invalid.","Lint config templates for unfilled/zero placeholders."],"tags":["mqtt","config-validation","batching"],"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"}