{"record":{"id":"45b9b79885d423e2","repo":"apache/flink","slug":"the-data-sink-has-no-input","errorCode":null,"errorMessage":"The data sink {} has no input.","messagePattern":"The data sink (.+?) has no input\\.","errorType":"validation","errorClass":"InvalidProgramException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java","lineNumber":180,"sourceCode":"            result = Collections.emptyList();\n        } else {\n            throw new RuntimeException(\"Cannot execute operator \" + operator.getClass().getName());\n        }\n\n        this.intermediateResults.put(operator, result);\n\n        return result;\n    }\n\n    // --------------------------------------------------------------------------------------------\n    //  Operator class specific execution methods\n    // --------------------------------------------------------------------------------------------\n\n    private <IN> void executeDataSink(GenericDataSinkBase<?> sink, int superStep, JobInfo jobInfo)\n            throws Exception {\n        Operator<?> inputOp = sink.getInput();\n        if (inputOp == null) {\n            throw new InvalidProgramException(\"The data sink \" + sink.getName() + \" has no input.\");\n        }\n\n        @SuppressWarnings(\"unchecked\")\n        List<IN> input = (List<IN>) execute(inputOp, jobInfo);\n\n        @SuppressWarnings(\"unchecked\")\n        GenericDataSinkBase<IN> typedSink = (GenericDataSinkBase<IN>) sink;\n\n        // build the runtime context and compute broadcast variables, if necessary\n        TaskInfo taskInfo = new TaskInfoImpl(typedSink.getName(), 1, 0, 1, 0);\n        RuntimeUDFContext ctx;\n\n        if (RichOutputFormat.class.isAssignableFrom(\n                typedSink.getUserCodeWrapper().getUserCodeClass())) {\n            ctx = createContext(superStep, taskInfo, jobInfo);\n        } else {\n            ctx = null;\n        }","sourceCodeStart":162,"sourceCodeEnd":198,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java#L162-L198","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure the sink's input operator/DataSet is set before execution: verify the chain from source to sink is complete.","Check that the DataSet variable feeding the sink is not null (e.g., print the plan or use env.createPlanAsJSON() to inspect).","If using the low-level API, call sink.setInput(sourceOperator) explicitly."],"exampleFix":"// before — sink input is null or unconnected\ndataSet.writeAsText(path);  // dataSet is null\n// after — ensure the transformation chain is complete\nDataSet<String> dataSet = env.fromElements(\"a\", \"b\").map(x -> x);\ndataSet.writeAsText(path);","handlingStrategy":"validation","validationCode":"// Ensure sink input is connected before execute()\nif (sink.getInput() == null) {\n    throw new IllegalStateException(\"Sink '\" + sink.getName() + \"' has no input — check the plan\");\n}\nenv.execute();","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always chain a source transformation before calling writeAsText()/output() on a DataSet.","Use env.createPlanAsJSON() to verify the full plan before execution.","Never leave a sink unconnected — every sink must trace back to a source."],"tags":["data-sink","plan-validation","collection-executor","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}