@Override
    public long currentWatermark() {
        return currentWatermark;
    }
@Test
    void testCurrentWatermark() throws Exception {
        BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
                new BatchExecutionKeyedStateBackend<>(
                        KEY_SERIALIZER, new KeyGroupRange(0, 1), new ExecutionConfig());
        InternalTimeServiceManager<Integer> timeServiceManager =
                BatchExecutionInternalTimeServiceManager.create(
                        UnregisteredMetricGroups.createUnregisteredTaskMetricGroup()
                                .getIOMetricGroup(),
                        keyedStatedBackend,
                        this.getClass().getClassLoader(),
                        new DummyKeyContext(),
                        new TestProcessingTimeService(),
                        Collections.emptyList(),
                        StreamTaskCancellationContext.alwaysRunning());

        List<Long> timers = new ArrayList<>();
        TriggerWithTimerServiceAccess<Integer, VoidNamespace> eventTimeTrigger =
                TriggerWithTimerServiceAccess.eventTimeTrigger(
                        (timer, timerService) -> {
                            assertThat(timerService.currentWatermark()).isEqualTo(Long.MAX_VALUE);
                            timers.add(timer.getTimestamp());
                        });
        InternalTimerService<VoidNamespace> timerService =
                timeServiceManager.getInternalTimerService(
                        "test", KEY_SERIALIZER, new VoidNamespaceSerializer(), eventTimeTrigger);
        eventTimeTrigger.setTimerService(timerService);

        assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE);
        keyedStatedBackend.setCurrentKey(1);
        timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123);
        assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE);

        // advancing the watermark to a value different than Long.MAX_VALUE should have no effect
        timeServiceManager.advanceWatermark(new Watermark(1000));
        assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE);

        // changing the current key fires all timers
        keyedStatedBackend.setCurrentKey(2);
        assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE);
        timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 124);

        // advancing the watermark to Long.MAX_VALUE should fire remaining key
        timeServiceManager.advanceWatermark(Watermark.MAX_WATERMARK);

        assertThat(timers).containsExactly(123L, 124L);
    }