apache/flink · error · InvalidProgramException

The data sink {} has no input.

Error message

The data sink {} has no input.

What it means

Thrown by CollectionExecutor.executeDataSink() when sink.getInput() returns null. A data sink must have a data source to write; an unconnected sink has nothing to consume. This is an InvalidProgramException indicating a malformed job plan — the sink was created but never had an input wired to it.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java:180

            result = Collections.emptyList();
        } else {
            throw new RuntimeException("Cannot execute operator " + operator.getClass().getName());
        }

        this.intermediateResults.put(operator, result);

        return result;
    }

    // --------------------------------------------------------------------------------------------
    //  Operator class specific execution methods
    // --------------------------------------------------------------------------------------------

    private <IN> void executeDataSink(GenericDataSinkBase<?> sink, int superStep, JobInfo jobInfo)
            throws Exception {
        Operator<?> inputOp = sink.getInput();
        if (inputOp == null) {
            throw new InvalidProgramException("The data sink " + sink.getName() + " has no input.");
        }

        @SuppressWarnings("unchecked")
        List<IN> input = (List<IN>) execute(inputOp, jobInfo);

        @SuppressWarnings("unchecked")
        GenericDataSinkBase<IN> typedSink = (GenericDataSinkBase<IN>) sink;

        // build the runtime context and compute broadcast variables, if necessary
        TaskInfo taskInfo = new TaskInfoImpl(typedSink.getName(), 1, 0, 1, 0);
        RuntimeUDFContext ctx;

        if (RichOutputFormat.class.isAssignableFrom(
                typedSink.getUserCodeWrapper().getUserCodeClass())) {
            ctx = createContext(superStep, taskInfo, jobInfo);
        } else {
            ctx = null;
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Ensure the sink's input operator/DataSet is set before execution: verify the chain from source to sink is complete.
2. Check that the DataSet variable feeding the sink is not null (e.g., print the plan or use env.createPlanAsJSON() to inspect).
3. If using the low-level API, call sink.setInput(sourceOperator) explicitly.

Example fix

// before — sink input is null or unconnected
dataSet.writeAsText(path);  // dataSet is null
// after — ensure the transformation chain is complete
DataSet<String> dataSet = env.fromElements("a", "b").map(x -> x);
dataSet.writeAsText(path);

// Ensure sink input is connected before execute()
if (sink.getInput() == null) {
    throw new IllegalStateException("Sink '" + sink.getName() + "' has no input — check the plan");
}
env.execute();

When it happens

Trigger: Creating a GenericDataSinkBase (or using the high-level API output()/write() method) without connecting it to a data source. In the high-level API this typically happens if a DataSet that was supposed to feed the sink is null or was produced by an operation that returned null.

Common situations: A developer constructs a plan where a sink's input DataSet was never assigned, or a transformation chain was broken (e.g., the result of a map/filter was discarded and the sink references a stale or null DataSet). Most commonly seen when programmatically building operator graphs.

Related errors

• The unary operation {} has no input.
• The binary operation {} has no first input.
• The binary operation {} has no second input.
• The iteration {} has no input (initial partial solution).
• The delta iteration {} has no initial solution set.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/45b9b79885d423e2. Report an issue: GitHub.