{"record":{"id":"674ea17f98d113ac","repo":"apache/seatunnel","slug":"mqttconnectorerrorcode-invalid-config","errorCode":"MqttConnectorErrorCode.INVALID_CONFIG","errorMessage":"PluginName: %s, Message: MQTT source only supports streaming job mode","messagePattern":"PluginName: (.+?), Message: MQTT source only supports streaming job mode","errorType":"error_code","errorClass":"MqttConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSource.java","lineNumber":50,"sourceCode":"\nimport java.util.Collections;\nimport java.util.List;\n\npublic class MqttSource extends AbstractSingleSplitSource<SeaTunnelRow> {\n\n    private final MqttSourceConfig sourceConfig;\n    private final CatalogTable catalogTable;\n    private JobContext jobContext;\n\n    public MqttSource(ReadonlyConfig pluginConfig) {\n        this.sourceConfig = new MqttSourceConfig(pluginConfig);\n        this.catalogTable = CatalogTableUtil.buildWithConfig(pluginConfig);\n    }\n\n    @Override\n    public Boundedness getBoundedness() {\n        if (jobContext != null && !JobMode.STREAMING.equals(jobContext.getJobMode())) {\n            throw new MqttConnectorException(\n                    MqttConnectorErrorCode.INVALID_CONFIG,\n                    String.format(\n                            \"PluginName: %s, Message: MQTT source only supports streaming job mode\",\n                            getPluginName()));\n        }\n        return Boundedness.UNBOUNDED;\n    }\n\n    @Override\n    public String getPluginName() {\n        return MqttSourceOptions.CONNECTOR_IDENTITY;\n    }\n\n    @Override\n    public void setJobContext(JobContext jobContext) {\n        this.jobContext = jobContext;\n    }\n","sourceCodeStart":32,"sourceCodeEnd":68,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-mqtt/src/main/java/org/apache/seatunnel/connectors/seatunnel/mqtt/source/MqttSource.java#L32-L68","documentation":"MqttSource.getBoundedness throws MqttConnectorException (INVALID_CONFIG) when the job mode is not STREAMING, because the MQTT source is a continuous unbounded source. It returns Boundedness.UNBOUNDED otherwise. Batch mode jobs cannot use this connector.","triggerScenarios":"Submitting a job with env job.mode = BATCH (or any non-streaming mode) that contains an MqttSource; getBoundedness is called during source initialization and immediately fails.","commonSituations":"Reusing a batch-oriented config template with the MQTT source; switching a pipeline from a bounded source (e.g. file) to MQTT without changing job mode; running local tests with default batch mode.","solutions":["Set env { job.mode = \"STREAMING\" } in the job config","If batch semantics are required, use a bounded source connector instead of MQTT","Update automated job templates that default to BATCH mode when they reference the MQTT source"],"exampleFix":"// before\nenv {\n  job.mode = \"BATCH\"\n}\n// after\nenv {\n  job.mode = \"STREAMING\"\n}","handlingStrategy":"validation","validationCode":"if (!\"STREAMING\".equalsIgnoreCase(jobMode)) {\n  throw new IllegalArgumentException(\"MQTT source requires env { job.mode = \\\"STREAMING\\\" }\");\n}","typeGuard":"null","tryCatchPattern":"try {\n  // submit job\n} catch (MqttConnectorException e) {\n  if (e.getCode() == MqttConnectorErrorCode.INVALID_CONFIG) {\n    LOG.error(\"Fix job.mode in env config: {}\", e.getMessage());\n  }\n  throw e;\n}","preventionTips":["Always set env.job.mode = STREAMING in configs using the MQTT source","Lint job configs for source/mode compatibility before submission","Document bounded-source alternatives for batch pipelines"],"tags":["mqtt","job-mode","streaming","config"],"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"}