protected List<OUT> executeOnCollections(RuntimeContext ctx, ExecutionConfig executionConfig)
            throws Exception {
        @SuppressWarnings("unchecked")
        InputFormat<OUT, InputSplit> inputFormat =
                (InputFormat<OUT, InputSplit>) this.formatWrapper.getUserCodeObject();
        // configure the input format
        inputFormat.configure(this.parameters);

        // open the input format
        if (inputFormat instanceof RichInputFormat) {
            ((RichInputFormat) inputFormat).setRuntimeContext(ctx);
            ((RichInputFormat) inputFormat).openInputFormat();
        }

        List<OUT> result = new ArrayList<OUT>();

        // splits
        InputSplit[] splits = inputFormat.createInputSplits(1);
        TypeSerializer<OUT> serializer =
                getOperatorInfo()
                        .getOutputType()
                        .createSerializer(executionConfig.getSerializerConfig());

        for (InputSplit split : splits) {
            inputFormat.open(split);

            while (!inputFormat.reachedEnd()) {
                OUT next = inputFormat.nextRecord(serializer.createInstance());
                if (next != null) {
                    result.add(serializer.copy(next));
                }
            }

            inputFormat.close();
        }

        // close the input format
        if (inputFormat instanceof RichInputFormat) {
            ((RichInputFormat) inputFormat).closeInputFormat();
        }

        return result;
    }
@Test
    void testDataSourceWithRuntimeContext() throws Exception {
        TestRichInputFormat in = new TestRichInputFormat();
        GenericDataSourceBase<String, TestRichInputFormat> source =
                new GenericDataSourceBase<>(
                        in,
                        new OperatorInformation<>(BasicTypeInfo.STRING_TYPE_INFO),
                        "testSource");

        final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<>();
        final HashMap<String, Future<Path>> cpTasks = new HashMap<>();
        final TaskInfo taskInfo = new TaskInfoImpl("test_source", 1, 0, 1, 0);

        ExecutionConfig executionConfig = new ExecutionConfig();
        executionConfig.disableObjectReuse();
        assertThat(in.hasBeenClosed()).isFalse();
        assertThat(in.hasBeenOpened()).isFalse();

        List<String> resultMutableSafe =
                source.executeOnCollections(
                        new RuntimeUDFContext(
                                taskInfo,
                                null,
                                executionConfig,
                                cpTasks,
                                accumulatorMap,
                                UnregisteredMetricsGroup.createOperatorMetricGroup()),
                        executionConfig);

        assertThat(in.hasBeenClosed()).isTrue();
        assertThat(in.hasBeenOpened()).isTrue();

        in.reset();
        executionConfig.enableObjectReuse();
        assertThat(in.hasBeenClosed()).isFalse();
        assertThat(in.hasBeenOpened()).isFalse();

        List<String> resultRegular =
                source.executeOnCollections(
                        new RuntimeUDFContext(
                                taskInfo,
                                null,
                                executionConfig,
                                cpTasks,
                                accumulatorMap,
                                UnregisteredMetricsGroup.createOperatorMetricGroup()),
                        executionConfig);

        assertThat(in.hasBeenClosed()).isTrue();
        assertThat(in.hasBeenOpened()).isTrue();

        assertThat(resultMutableSafe).isEqualTo(asList(TestIOData.RICH_NAMES));
        assertThat(resultRegular).isEqualTo(asList(TestIOData.RICH_NAMES));
    }