{"record":{"id":"bbfcc668c8c3e99a","repo":"apache/flink","slug":"failed-to-open-the-generatorfunction","errorCode":null,"errorMessage":"Failed to open the GeneratorFunction","messagePattern":"Failed to open the GeneratorFunction","errorType":"exception","errorClass":"FlinkRuntimeException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-datagen/src/main/java/org/apache/flink/connector/datagen/source/DoubleEmittingSourceReaderWithCheckpointsInBetween.java","lineNumber":83,"sourceCode":"        super(context);\n        this.generatorFunction = checkNotNull(generatorFunction);\n        this.allowedToExit = allowedToExit;\n    }\n\n    public DoubleEmittingSourceReaderWithCheckpointsInBetween(\n            SourceReaderContext context, GeneratorFunction<E, O> generatorFunction) {\n        super(context);\n        this.generatorFunction = checkNotNull(generatorFunction);\n    }\n\n    // ------------------------------------------------------------------------\n\n    @Override\n    public void start(SourceReaderContext context) {\n        try {\n            generatorFunction.open(context);\n        } catch (Exception e) {\n            throw new FlinkRuntimeException(\"Failed to open the GeneratorFunction\", e);\n        }\n    }\n\n    @Override\n    public InputStatus pollNext(ReaderOutput<O> output) {\n        // This is the termination path after the test data has been emitted twice\n        if (done) {\n            if (allowedToExit != null) { // Termination is controlled externally\n                return allowedToExit.getAsBoolean()\n                        ? InputStatus.END_OF_INPUT\n                        : InputStatus.NOTHING_AVAILABLE;\n            } else {\n                return InputStatus.END_OF_INPUT;\n            }\n        }\n        // This is the initial path\n        if (currentSplit == null) {\n            InputStatus inputStatus = tryMoveToNextSplit();","sourceCodeStart":65,"sourceCodeEnd":101,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-datagen/src/main/java/org/apache/flink/connector/datagen/source/DoubleEmittingSourceReaderWithCheckpointsInBetween.java#L65-L101","documentation":"Thrown by start() in DoubleEmittingSourceReaderWithCheckpointsInBetween when the user-provided GeneratorFunction.open(context) throws an exception. The original exception is wrapped in FlinkRuntimeException. This source reader is @Experimental and designed for testing purposes (checkpoint-synchronized double emission).","triggerScenarios":"A custom GeneratorFunction whose open() method throws — e.g., failing to initialize resources, connecting to an unreachable service, or throwing a runtime exception during setup. The source reader calls generatorFunction.open(context) during its start() lifecycle.","commonSituations":"Generator function trying to open a database or network connection in open() that fails; misconfigured SourceReaderContext; resource initialization failure (file handle, thread pool).","solutions":["Inspect the wrapped cause exception in the FlinkRuntimeException for the root failure.","Fix the root cause in the GeneratorFunction's open() method (e.g., add connection retry, validate configuration).","Ensure the SourceReaderContext is properly configured and accessible.","If the open() method needs external resources, verify they are available before job submission."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Validate open() does not throw by testing the GeneratorFunction separately\nGeneratorFunction<E, O> fn = ...;\nfn.open(testContext); // mock or real SourceReaderContext\nfn.close();","typeGuard":null,"tryCatchPattern":"try {\n    sourceReader.start(context);\n} catch (FlinkRuntimeException e) {\n    if (e.getMessage().equals(\"Failed to open the GeneratorFunction\")) {\n        // inspect e.getCause() for the root failure in open()\n    }\n    throw e;\n}","preventionTips":["Unit-test GeneratorFunction.open() with a mock SourceReaderContext.","Ensure external resources are available before job submission.","Add retry logic in open() for transient connection failures."],"tags":["datagen","source-reader","generator-function","lifecycle","test"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}