@Override
    public void emitWatermark(Watermark watermark) {
        final long newWatermark = watermark.getTimestamp();
        if (newWatermark <= maxWatermarkSoFar) {
            return;
        }

        maxWatermarkSoFar = newWatermark;
        watermarkEmitted.updateCurrentEffectiveWatermark(maxWatermarkSoFar);

        try {
            markActiveInternally();

            output.emitWatermark(
                    new org.apache.flink.streaming.api.watermark.Watermark(newWatermark));
        } catch (ExceptionInChainedOperatorException e) {
            throw e;
        } catch (Exception e) {
            throw new ExceptionInChainedOperatorException(e);
        }
    }
@Test
    void testInitialZeroWatermark() {
        final CollectingDataOutput<Object> testingOutput = new CollectingDataOutput<>();
        final WatermarkToDataOutput wmOutput = new WatermarkToDataOutput(testingOutput);

        wmOutput.emitWatermark(new org.apache.flink.api.common.eventtime.Watermark(0L));

        assertThat(testingOutput.events).contains(new Watermark(0L));
    }