@Override
    public void open() throws Exception {
        mainInputActivityClock = new PausableRelativeClock(getProcessingTimeService().getClock());
        TaskIOMetricGroup taskIOMetricGroup =
                getContainingTask().getEnvironment().getMetricGroup().getIOMetricGroup();
        taskIOMetricGroup.registerBackPressureListener(mainInputActivityClock);

        initReader();

        // in the future when we this one is migrated to the "eager initialization" operator
        // (StreamOperatorV2), then we should evaluate this during operator construction.
        if (emitProgressiveWatermarks) {
            eventTimeLogic =
                    TimestampsAndWatermarks.createProgressiveEventTimeLogic(
                            watermarkStrategy,
                            sourceMetricGroup,
                            getProcessingTimeService(),
                            getExecutionConfig().getAutoWatermarkInterval(),
                            mainInputActivityClock,
                            getProcessingTimeService().getClock(),
                            taskIOMetricGroup);
        } else {
            eventTimeLogic =
                    TimestampsAndWatermarks.createNoOpEventTimeLogic(
                            watermarkStrategy, sourceMetricGroup, mainInputActivityClock);
        }

        // restore the state if necessary.
        final List<SplitT> splits = CollectionUtil.iterableToList(readerState.get());
        if (!splits.isEmpty()) {
            LOG.info("Restoring state for {} split(s) to reader.", splits.size());
            splitsToInitializeOutput.addAll(splits);
            sourceReader.addSplits(splits);
        }

        // Register the reader to the coordinator.
        registerReader();

        sourceMetricGroup.idlingStarted();
        // Start the reader after registration, sending messages in start is allowed.
        sourceReader.start();

        eventTimeLogic.startPeriodicWatermarkEmits();
    }
@Test
    void testOpen() throws Exception {
        // Initialize the operator.
        operator.initializeState(context.createStateContext());
        // Open the operator.
        operator.open();
        // The source reader should have been assigned a split.
        assertThat(mockSourceReader.getAssignedSplits())
                .containsExactly(SourceOperatorTestContext.MOCK_SPLIT);
        // The source reader should have started.
        assertThat(mockSourceReader.isStarted()).isTrue();

        // A ReaderRegistrationRequest should have been sent.
        assertThat(mockGateway.getEventsSent()).hasSize(1);
        OperatorEvent operatorEvent = mockGateway.getEventsSent().get(0);
        assertThat(operatorEvent).isInstanceOf(ReaderRegistrationEvent.class);
        assertThat(((ReaderRegistrationEvent) operatorEvent).subtaskId())
                .isEqualTo(SourceOperatorTestContext.SUBTASK_INDEX);
    }