class WatermarkToDataOutput implements WatermarkOutput {

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

		maxWatermarkSoFar = newWatermark;

		try {
			if (isIdle) {
				output.emitStreamStatus(StreamStatus.ACTIVE);
				isIdle = false;
			}

			output.emitWatermark(new org.apache.flink.streaming.api.watermark.Watermark(newWatermark));
		} catch (ExceptionInChainedOperatorException e) {
			throw e;
		} catch (Exception e) {
			throw new ExceptionInChainedOperatorException(e);
		}
	}

	public  WatermarkToDataOutput(PushingAsyncDataInput.DataOutput<?> output);

}