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

        // watermark and timestamp should start from 0
        this.currentWatermark = 0;
        this.watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
        long now = getProcessingTimeService().getCurrentProcessingTime();
        this.lastWatermarkPeriodicEmitTime = now;
        this.timeSinceLastIdleCheck = now;

        if (watermarkInterval > 0 || idleTimeout > 0) {
            this.timerInterval =
                    calculateProcessingTimeTimerInterval(watermarkInterval, idleTimeout);
            getProcessingTimeService().registerTimer(now + timerInterval, this);
        }

        FunctionUtils.setFunctionRuntimeContext(watermarkGenerator, getRuntimeContext());
        FunctionUtils.openFunction(watermarkGenerator, DefaultOpenContext.INSTANCE);
    }
@Test
    public void testIdleTimeoutUnderBackpressure() throws Exception {
        long idleTimeout = 100;

        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(0, WATERMARK_GENERATOR, idleTimeout);
        testHarness.getExecutionConfig().setAutoWatermarkInterval(idleTimeout);
        testHarness.open();

        TaskIOMetricGroup taskIOMetricGroup =
                testHarness.getEnvironment().getMetricGroup().getIOMetricGroup();
        taskIOMetricGroup.getHardBackPressuredTimePerSecond().markStart();

        stepProcessingTime(testHarness, 0, idleTimeout * 10, idleTimeout / 10);
        assertThat(testHarness.getOutput()).isEmpty();

        taskIOMetricGroup.getHardBackPressuredTimePerSecond().markEnd();
        taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markStart();

        stepProcessingTime(testHarness, idleTimeout * 10, idleTimeout * 20, idleTimeout / 10);
        assertThat(testHarness.getOutput()).isEmpty();

        taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markEnd();

        stepProcessingTime(testHarness, idleTimeout * 20, idleTimeout * 30, idleTimeout / 10);
        assertThat(testHarness.getOutput()).containsExactly(WatermarkStatus.IDLE);
    }