class SourceOperator extends AbstractStreamOperator<OUT> implements OperatorEventHandler, PushingAsyncDataInput<OUT> {

	@Override
	public void snapshotState(StateSnapshotContext context) throws Exception {
		LOG.debug("Taking a snapshot for checkpoint {}", context.getCheckpointId());
		readerState.update(sourceReader.snapshotState());
	}

	public  SourceOperator(
			Function<SourceReaderContext, SourceReader<OUT, SplitT>> readerFactory,
			OperatorEventGateway operatorEventGateway,
			SimpleVersionedSerializer<SplitT> splitSerializer,
			WatermarkStrategy<OUT> watermarkStrategy,
			ProcessingTimeService timeService,
			Configuration configuration,
			String localHostname);

}