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

        timestampAssigner = watermarkStrategy.createTimestampAssigner(this::getMetricGroup);
        watermarkGenerator =
                emitProgressiveWatermarks
                        ? watermarkStrategy.createWatermarkGenerator(
                                new WatermarkGeneratorSupplier.Context() {
                                    @Override
                                    public MetricGroup getMetricGroup() {
                                        return this.getMetricGroup();
                                    }

                                    @Override
                                    public RelativeClock getInputActivityClock() {
                                        return inputActivityClock;
                                    }
                                })
                        : new NoWatermarksGenerator<>();

        wmOutput = new WatermarkEmitter(output);

        watermarkInterval = getExecutionConfig().getAutoWatermarkInterval();
        if (watermarkInterval > 0 && emitProgressiveWatermarks) {
            final long now = getProcessingTimeService().getCurrentProcessingTime();
            getProcessingTimeService().registerTimer(now + watermarkInterval, this);
        }
    }
@Test
    void watermarksWithIdlenessUnderBackpressure() throws Exception {
        long idleTimeout = 100;

        TimestampsAndWatermarksOperator<Tuple2<Boolean, Long>> operator =
                new TimestampsAndWatermarksOperator<>(
                        WatermarkStrategy.forGenerator((ctx) -> new PunctuatedWatermarkGenerator())
                                .withTimestampAssigner((ctx) -> new TupleExtractor())
                                .withIdleness(Duration.ofMillis(idleTimeout)),
                        true);

        OneInputStreamOperatorTestHarness<Tuple2<Boolean, Long>, Tuple2<Boolean, Long>>
                testHarness = new OneInputStreamOperatorTestHarness<>(operator);
        testHarness.open();

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

        for (int i = 0; i < 10; i++) {
            testHarness.advanceTime(idleTimeout);
        }
        assertThat(testHarness.getOutput()).isEmpty();

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

        for (int i = 10; i < 20; i++) {
            testHarness.advanceTime(idleTimeout);
        }
        assertThat(testHarness.getOutput()).isEmpty();

        taskIOMetricGroup.getSoftBackPressuredTimePerSecond().markEnd();

        for (int i = 20; i < 30; i++) {
            testHarness.advanceTime(idleTimeout);
        }
        assertThat(testHarness.getOutput()).containsExactly(WatermarkStatus.IDLE);
    }