apache/seatunnel · error · UnsupportedOperationException

Multiple input tables are not supported in flink plugin

Error message

Multiple input tables are not supported in flink plugin

What it means

FlinkAbstractPluginExecuteProcessor.fromSourceTable resolves which upstream DataStream(s) a plugin consumes via the 'plugin_input' option. The Flink starter only supports single-input plugins, so it explicitly rejects configurations listing more than one input table identifier with an UnsupportedOperationException.

Solutions

Reduce the job to a single upstream input per plugin and add intermediate transform stages instead of multi-input joins
Run the job on the Zeta (SeaTunnel) engine, which supports multiple input tables
Combine upstream streams before the plugin so only one DataStream reaches it
Remove duplicate plugin_input entries if the multiple values were unintentional

Example fix

// before
transform {
  Sql = {
    plugin_input = ["source_a", "source_b"]
  }
}
// after
transform {
  Sql = {
    plugin_input = "source_a"
  }
}

Defensive patterns

Strategy: validation

Validate before calling

List<String> inputs = readonlyConfig.get(PLUGIN_INPUT);
if (inputs != null && inputs.size() > 1) {
    throw new IllegalArgumentException("Flink starter supports at most one plugin_input, got: " + inputs);
}

Type guard

boolean supportsMultiInput(JobMode mode) {
    return mode == JobMode.ZETA; // only Zeta supports multiple input tables
}

Try / catch

try {
    processor.execute(...);
} catch (UnsupportedOperationException e) {
    // fall back to single-input config or Zeta engine
}

Prevention

Keep one plugin_input per plugin in Flink jobs
Use the Zeta engine when a job needs multi-input/join semantics
Lint job configs to reject plugin_input lists longer than 1 for Flink execution

When it happens

Trigger: A job config specifies 'plugin_input' with more than one table identifier (or 'plugin_input = ["t1","t2"]') for a Flink-executed source/transform/sink plugin.

Common situations: Users porting multi-input jobs from the Zeta engine to Flink; copy-pasting configs that feed two upstream transforms into one plugin; joining two streams in a Flink-launched SeaTunnel job.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

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

Appendix: source

Thrown at seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/src/main/java/org/apache/seatunnel/core/starter/flink/execution/FlinkAbstractPluginExecuteProcessor.java:101

        this.pluginConfigs = pluginConfigs;
        this.jobContext = jobContext;
        this.plugins = initializePlugins(jarPaths, pluginConfigs);
        this.envConfig = envConfig;
    }

    @Override
    public void setRuntimeEnvironment(FlinkRuntimeEnvironment flinkRuntimeEnvironment) {
        this.flinkRuntimeEnvironment = flinkRuntimeEnvironment;
    }

    protected Optional<DataStreamTableInfo> fromSourceTable(
            Config pluginConfig, List<DataStreamTableInfo> upstreamDataStreams) {
        ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(pluginConfig);

        if (readonlyConfig.getOptional(PLUGIN_INPUT).isPresent()) {
            List<String> pluginInputIdentifiers = readonlyConfig.get(PLUGIN_INPUT);
            if (pluginInputIdentifiers.size() > 1) {
                throw new UnsupportedOperationException(
                        "Multiple input tables are not supported in flink plugin");
            }

            String tableName = pluginInputIdentifiers.get(0);
            DataStreamTableInfo dataStreamTableInfo =
                    upstreamDataStreams.stream()
                            .filter(info -> tableName.equals(info.getTableName()))
                            .findFirst()
                            .orElseThrow(
                                    () ->
                                            new SeaTunnelException(
                                                    String.format(
                                                            "table %s not found", tableName)));
            return Optional.of(
                    new DataStreamTableInfo(
                            dataStreamTableInfo.getDataStream(),
                            dataStreamTableInfo.getCatalogTables(),
                            tableName));

View on GitHub (pinned to cf67b549a7)