@Override
    public void open() throws Exception {
        super.open();

        checkState(
                this.serializer != null,
                "The serializer has not been set. "
                        + "Probably the setOutputType() was not called. Please report it.");

        this.state = ReaderState.IDLE;
        if (this.format instanceof RichInputFormat) {
            ((RichInputFormat<?, ?>) this.format).setRuntimeContext(getRuntimeContext());
        }
        this.format.configure(new Configuration());

        this.sourceContext =
                StreamSourceContexts.getSourceContext(
                        getOperatorConfig().getTimeCharacteristic(),
                        getProcessingTimeService(),
                        new Object(), // no actual locking needed
                        output,
                        getExecutionConfig().getAutoWatermarkInterval(),
                        -1,
                        true);

        this.reusedRecord = serializer.createInstance();
        this.completedSplitsCounter = getMetricGroup().counter("numSplitsProcessed");
        this.splits = this.splits == null ? new PriorityQueue<>() : this.splits;

        if (!splits.isEmpty()) {
            enqueueProcessRecord();
        }
    }
@Test
    void testExceptionRethrownFromClose() throws Exception {
        OneInputStreamOperatorTestHarness<TimestampedFileInputSplit, String> harness =
                createHarness(failingFormat());
        harness.getExecutionConfig().setAutoWatermarkInterval(10);
        harness.setTimeCharacteristic(TimeCharacteristic.IngestionTime);

        assertThatThrownBy(
                        () -> {
                            try (OneInputStreamOperatorTestHarness<
                                            TimestampedFileInputSplit, String>
                                    tester = harness) {
                                tester.open();
                            }
                        })
                .isInstanceOf(ExpectedTestException.class);
    }