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

	@Override
	public void processElement(final StreamRecord<T> element) throws Exception {
		final T event = element.getValue();
		final long previousTimestamp = element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE;
		final long newTimestamp = timestampAssigner.extractTimestamp(event, previousTimestamp);

		element.setTimestamp(newTimestamp);
		output.collect(element);
		watermarkGenerator.onEvent(event, newTimestamp, wmOutput);
	}

	public  TimestampsAndWatermarksOperator(
			WatermarkStrategy<T> watermarkStrategy);

}