class ContinuousFileReaderOperator extends AbstractStreamOperator<OUT> implements OneInputStreamOperator<T, OUT>, OutputTypeConfigurable<OUT> {

	@Override
	public void open() throws Exception {
		super.open();

		checkState(this.serializer != null, "The serializer has not been set. " +
			"Probably the setOutputType() was not called. Please report it.");

		this.state = ReaderState.IDLE;
		if (this.format instanceof RichInputFormat) {
			((RichInputFormat) this.format).setRuntimeContext(getRuntimeContext());
		}
		this.format.configure(new Configuration());

		this.sourceContext = StreamSourceContexts.getSourceContext(
			getOperatorConfig().getTimeCharacteristic(),
			getProcessingTimeService(),
			new Object(), // no actual locking needed
			getContainingTask().getStreamStatusMaintainer(),
			output,
			getRuntimeContext().getExecutionConfig().getAutoWatermarkInterval(),
			-1);

		this.reusedRecord = serializer.createInstance();
		this.completedSplitsCounter = getMetricGroup().counter("numSplitsProcessed");
		this.splits = this.splits == null ? new PriorityQueue<>() : this.splits;

		if (!splits.isEmpty()) {
			enqueueProcessRecord();
		}
	}

	  ContinuousFileReaderOperator(
		InputFormat<OUT, ? super T> format,
		ProcessingTimeService processingTimeService,
		MailboxExecutor mailboxExecutor);

}