apache/seatunnel · error · MqttConnectorException

MqttConnectorErrorCode.INVALID_CONFIG

MqttConnectorErrorCode.INVALID_CONFIG

Error message

PluginName: %s, Message: MQTT source only supports streaming job mode

What it means

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.

Source

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

import java.util.Collections;
import java.util.List;

public class MqttSource extends AbstractSingleSplitSource<SeaTunnelRow> {

    private final MqttSourceConfig sourceConfig;
    private final CatalogTable catalogTable;
    private JobContext jobContext;

    public MqttSource(ReadonlyConfig pluginConfig) {
        this.sourceConfig = new MqttSourceConfig(pluginConfig);
        this.catalogTable = CatalogTableUtil.buildWithConfig(pluginConfig);
    }

    @Override
    public Boundedness getBoundedness() {
        if (jobContext != null && !JobMode.STREAMING.equals(jobContext.getJobMode())) {
            throw new MqttConnectorException(
                    MqttConnectorErrorCode.INVALID_CONFIG,
                    String.format(
                            "PluginName: %s, Message: MQTT source only supports streaming job mode",
                            getPluginName()));
        }
        return Boundedness.UNBOUNDED;
    }

    @Override
    public String getPluginName() {
        return MqttSourceOptions.CONNECTOR_IDENTITY;
    }

    @Override
    public void setJobContext(JobContext jobContext) {
        this.jobContext = jobContext;
    }

View on GitHub (pinned to cf67b549a7)

Solutions

1. Set env { job.mode = "STREAMING" } in the job config
2. If batch semantics are required, use a bounded source connector instead of MQTT
3. Update automated job templates that default to BATCH mode when they reference the MQTT source

Example fix

// before
env {
  job.mode = "BATCH"
}
// after
env {
  job.mode = "STREAMING"
}

if (!"STREAMING".equalsIgnoreCase(jobMode)) {
  throw new IllegalArgumentException("MQTT source requires env { job.mode = \"STREAMING\" }");
}

null

try {
  // submit job
} catch (MqttConnectorException e) {
  if (e.getCode() == MqttConnectorErrorCode.INVALID_CONFIG) {
    LOG.error("Fix job.mode in env config: {}", e.getMessage());
  }
  throw e;
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• '%s' source don't support off-line job.
• CONFIGURATION_FAILED
• UNSUPPORTED_OPERATION
• Unsupported MQTT sink format:
• client_id is required when clean_session=false for MQTT sour

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