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 testDataSourcePlain() throws Exception {
        TestNonRichInputFormat in = new TestNonRichInputFormat();
        GenericDataSourceBase<String, TestNonRichInputFormat> source =
                new GenericDataSourceBase<>(
                        in,
                        new OperatorInformation<>(BasicTypeInfo.STRING_TYPE_INFO),
                        "testSource");

        ExecutionConfig executionConfig = new ExecutionConfig();
        executionConfig.disableObjectReuse();
        List<String> resultMutableSafe = source.executeOnCollections(null, executionConfig);

        in.reset();
        executionConfig.enableObjectReuse();
        List<String> resultRegular = source.executeOnCollections(null, executionConfig);
        assertThat(resultMutableSafe).isEqualTo(asList(TestIOData.NAMES));
        assertThat(resultRegular).isEqualTo(asList(TestIOData.NAMES));
    }