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

	@Override
	public void initializeState(StateInitializationContext context) throws Exception {
		super.initializeState(context);
		final ListState<byte[]> rawState = context.getOperatorStateStore().getListState(SPLITS_STATE_DESC);
		readerState = new SimpleVersionedListState<>(rawState, splitSerializer);
	}

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

}