class TimestampsAndWatermarksOperator extends AbstractStreamOperator<T> implements OneInputStreamOperator<T, T>, ProcessingTimeCallback {

	@Override
	public void processWatermark(org.apache.flink.streaming.api.watermark.Watermark mark) throws Exception {
		// if we receive a Long.MAX_VALUE watermark we forward it since it is used
		// to signal the end of input and to not block watermark progress downstream
		if (mark.getTimestamp() == Long.MAX_VALUE) {
			wmOutput.emitWatermark(Watermark.MAX_WATERMARK);
		}
	}

	public  TimestampsAndWatermarksOperator(
			WatermarkStrategy<T> watermarkStrategy);

}