public interface SourceReader<T, SplitT extends SourceSplit> extends Serializable, AutoCloseable {

	void start();

	Status pollNext(SourceOutput<T> sourceOutput) throws Exception;

	List<SplitT> snapshotState();

	CompletableFuture<Void> isAvailable();

	

interface SourceReader {

    void start() throws IOException;

    CompletableFuture<?> isAvailable() throws IOException;

    Status pollNext(SourceOutput<E> output) throws IOException;

    void addSplits(List<SplitT> splits) throws IOException;

	default void handleSourceEvents(SourceEvent sourceEvent)  List<SplitT> snapshotState();
}{
		// Do nothing.
	}

	enum Status {
		/** The next record is available right now. */
		AVAILABLE_NOW,
		/** The next record will be available later. */
		AVAILABLE_LATER,
		/** The source reader has completed all the reading work. */
		FINISHED
	}
}

/**
 * The interface for Source. It acts like a factory class that helps construct
 * the {@link SplitEnumerator} and {@link SourceReader} and corresponding
 * serializers.
 *
 * @param <T>        The type of records produced by the source.
 * @param <SplitT>   The type of splits handled by the source.
 * @param <EnumChkT> The type of the enumerator checkpoints.
 */
public interface Source<T, SplitT extends SourceSplit, EnumChkT> extends Serializable {

	/**
	 * Checks whether the source supports the given boundedness.
	 *
	 * <p>Some sources might only support either continuous unbounded streams, or
	 * bounded streams.
	 *
	 * @param boundedness The boundedness to check.
	 * @return <code>true</code> if the given boundedness is supported, <code>false</code> otherwise.
	 */
	boolean supportsBoundedness(Boundedness boundedness);

	/**
	 * Creates a new reader to read data from the spits it gets assigned.
	 * The reader starts fresh and does not have any state to resume.
	 *
	 * @param config A flat config for this source operator.
	 * @param readerContext The {@link SourceReaderContext context} for the source reader.
	 * @return A new SourceReader.
	 */
	SourceReader<T, SplitT> createReader(
			Configuration config,
			SourceReaderContext readerContext);

	/**
	 * Creates a new SplitEnumerator for this source, starting a new input.
	 *
	 * @param config The configuration for this operator.
	 * @param enumContext The {@link SplitEnumeratorContext context} for the split enumerator.
	 * @return A new SplitEnumerator.
	 */
	SplitEnumerator<SplitT, EnumChkT> createEnumerator(
			Configuration config,
			SplitEnumeratorContext<SplitT> enumContext);

	/**
	 * Restores an enumerator from a checkpoint.
	 *
	 * @param config The configuration of this operator.
	 * @param enumContext The {@link SplitEnumeratorContext context} for the restored split enumerator.
	 * @param checkpoint The checkpoint to restore the SplitEnumerator from.
	 * @return A SplitEnumerator restored from the given checkpoint.
	 */
	SplitEnumerator<SplitT, EnumChkT> restoreEnumerator(
			Configuration config,
			SplitEnumeratorContext<SplitT> enumContext,
			EnumChkT checkpoint) throws IOException;

	// ------------------------------------------------------------------------
	//  serializers for the metadata
	// ------------------------------------------------------------------------

	/**
	 * Creates a serializer for the source splits. Splits are serialized when sending them
	 * from enumerator to reader, and when checkpointing the reader's current state.
	 *
	 * @return The serializer for the split type.
	 */
	SimpleVersionedSerializer<SplitT> getSplitSerializer();

	/**
	 * Creates the serializer for the {@link SplitEnumerator} checkpoint.
	 * The serializer is used for the result of the {@link SplitEnumerator#snapshotState()}
	 * method.
	 *
	 * @return The serializer for the SplitEnumerator checkpoint.
	 */
	SimpleVersionedSerializer<EnumChkT> getEnumeratorCheckpointSerializer();
}


/**
 * The boundedness of the source: "bounded" for the currently available data (batch style),
 * "continuous unbounded" for a continuous streaming style source.
 */
public enum Boundedness {

	/**
	 * A bounded source processes the data that is currently available and will end after that.
	 *
	 * <p>When a source produces a bounded stream, the runtime may activate additional optimizations
	 * that are suitable only for bounded input. Incorrectly producing unbounded data when the source
	 * is set to produce a bounded stream will often result in programs that do not output any results
	 * and may eventually fail due to runtime errors (out of memory or storage).
	 */
	BOUNDED,

	/**
	 * A continuous unbounded source continuously processes all data as it comes.
	 *
	 * <p>The source may run forever (until the program is terminated) or might actually end at some point,
	 * based on some source-specific conditions. Because that is not transparent to the runtime,
	 * the runtime will use an execution mode for continuous unbounded streams whenever this mode
	 * is chosen.
	 */
	CONTINUOUS_UNBOUNDED
}

/**
 * An interface for all the Split types to implement.
 */
public interface SourceSplit {

	/**
	 * Get the split id of this source split.
	 * @return id of this source split.
	 */
	String splitId();
}

public interface SourceOutput<E> extends WatermarkOutput {

	void emitRecord(E record);

	void emitRecord(E record, long timestamp);
}

/**
 * A context for the source reader. It allows the source reader to get the context information and
 * allows the SourceReader to send source event to its split enumerator.
 */
public interface SourceReaderContext {

	/**
	 * Returns Createsthe ametric serializergroup for thethis sourceparallel splits. Splits are serialized when sending themsubtask.
	 *
	 * @return metric group for this parallel subtask.
	 * from enumerator to reader, and when checkpointing the reader's current state/
	MetricGroup getMetricGroup();

	/**
	 * Send a source event to the corresponding SplitEnumerator.
	 *
	 * @param @returnevent The serializersource forevent theto split typesend.
	 */
	SimpleVersionedSerializer<SplitT>void getSplitSerializersendEventToEnumerator(SourceEvent event);
}

	/**
	 * A interface of a split enumerator responsible for the followings:
 * Creates1. discover the serializersplits for the {@link SplitEnumeratorSourceReader} checkpointto read.
	 * The serializer is used for the result of the {@link SplitEnumerator#snapshotState()}
	 * method.
	 *
	 * @return The serializer for the SplitEnumerator checkpoint.
	 */
	SimpleVersionedSerializer<EnumChkT> getEnumeratorCheckpointSerializer();
}


/**
 * The boundedness of the source: "bounded" for the currently available data (batch style),
 * "continuous unbounded" for a continuous streaming style source.
 */
public enum Boundedness {2. assign the splits to the source reader.
 */
public interface SplitEnumerator<SplitT extends SourceSplit, CheckpointT> extends AutoCloseable {

	/**
	 * Start the split enumerator.
	 *
	 * <p>The default behavior does nothing.
	 */
	default void start() {
		// By default do nothing.
	}

	/**
	 * AHandles boundedthe source processesevent from the data that is currently available and will end after thatsource reader.
	 *
	 * @param <p>WhensubtaskId athe sourcesubtask producesid aof boundedthe stream,source thereader runtimewho maysent activatethe additionalsource optimizationsevent.
	 * @param thatsourceEvent the aresource suitableevent onlyfrom forthe boundedsource inputreader. Incorrectly producing unbounded data when the source
	 */
	void handleSourceEvent(int subtaskId, SourceEvent sourceEvent);

	/**
	 * Add isa split setback to producethe asplit boundedenumerator. streamIt will oftenonly resulthappen inwhen programsa that{@link do not output any resultsSourceReader} fails
	 * and maythere eventuallyare failsplits dueassigned to runtimeit errorsafter (outthe oflast memory or storage)successful checkpoint.
	 */
	BOUNDED,

	/**
	 * A continuous unbounded source continuously processes all data as it comes.
	 * * @param splits The split to add back to the enumerator for reassignment.
	 * <p>The@param sourcesubtaskId mayThe runid foreverof (until the programsubtask isto terminated)which orthe mightreturned actually end at some point,
	 * based on some source-specific conditions. Because that is not transparent to the runtime,splits belong.
	 */
	void addSplitsBack(List<SplitT> splits, int subtaskId);

	/**
	 * Add a new source reader with the given subtask ID.
	 *
	 * the@param runtimesubtaskId willthe usesubtask an execution mode for continuous unbounded streams whenever this mode
	 * is chosenID of the new source reader.
	 */
	CONTINUOUS_UNBOUNDED
}

/**
	 * Handles theAn source event fromfor the SourceReaders sourceand readerEnumerators.
	 */
	public *interface @paramSourceEvent subtaskIdextends the subtask id of the source reader who sent the source event.
	 * @param sourceEvent the source event from the source reader.
	 */
	void handleSourceEvent(int subtaskId, SourceEvent sourceEvent);

	/**
	 * Add a split back to the split enumerator. It will only happen when a {@link SourceReader} fails
	 * and there are splits assigned to it after the last successful checkpoint.
	 *
	 * @param splits The split to add back to the enumerator for reassignment.
	 * @param subtaskId The id of the subtask to which the returned splits belong.
	 */
	void addSplitsBack(List<SplitT> splits, int subtaskId);

	/**
	 * Add a new source reader with the given subtask ID.
	 *
	 * @param subtaskId the subtask ID of the new source reader.
	 */
	void addReader(int subtaskId);

	/**
	 * Checkpoints the state of this split enumerator.
	 */
	CheckpointT snapshotState();

	/**
	 * Called to close the enumerator, in case it holds on to any resources, like threads or
	 * network connections.
	 */
	@Override
	void close() throws IOException;
}

public class StreamExecutionEnvironment {
...
    public <T> DataStream<T> continuousSource(Source<T, ?, ?> source) {...}

    public <T> DataStream<T> continuousSource(Source<T, ?, ?> source, TypeInformation<T> type) {...}

    public <T> DataStream<T> boundedSource(Source<T, ?, ?> source) {...}

    public <T> DataStream<T> boundedSource(Source<T, ?, ?> source, TypeInformation<T> type) {...}
...
}

/**
 * An interface used to read from splits. The implementation could either read from a single split or from
 * multiple splits.
 *
 * @param <E> the element type.
 * @param <SplitT> the split type.
 */
public interface SplitReader<E, SplitT extends SourceSplit> {

	/**
	 * Fetch elements into the blocking queue for the given splits. The fetch call could be blocking
	 * but it should get unblocked when {@link #wakeUp()} is invoked. In that case, the implementation
	 * may either decide to return without throwing an exception, or it can just throw an interrupted
	 * exception. In either case, this method should be reentrant, meaning that the next fetch call
	 * should just resume from where the last fetch call was waken up or interrupted

/**
 * A context class for the {@link SplitEnumerator}. This class serves the following purposes:
 * 1. Host information necessary for the SplitEnumerator to make split assignment decisions.
 * 2. Accept and track the split assignment from the enumerator.
 * 3. Provide a managed threading model so the split enumerators do not need to create their
 *    own internal threads.
 *
 * @param <SplitT> the type of the splits.
 */
public interface SplitEnumeratorContext<SplitT extends SourceSplit> {

	MetricGroup metricGroup();

	/**
	 * Send a source event to a source reader. The source reader is identified by its subtask id.
	 *
	 * @param subtaskId@return the subtask idIds of the source reader to send this event to.
	 * @param event the source event to send finished splits.
	 *
	 * @throws InterruptedException when interrupted
	 */
	RecordsWithSplitIds<E> fetch() throws InterruptedException;

	/**
	 * Handle the split changes. This call should be non-blocking.
	 * @return
	 * @param splitsChanges a completablequeue futurewith whichsplit willchanges bethat completedhas whennot thebeen eventhandled isby successfullythis sentSplitReader.
	 */
	void sendEventToSourceReader(int subtaskId, SourceEvent eventhandleSplitsChanges(Queue<SplitsChange<SplitT>> splitsChanges);

	/**
	 * Get Wake up the split reader in case the numberfetcher thread of subtasks.
	 *is blocking in
	 * @return the number of subtasks{@link #fetch()}.
	 */
	intvoid numSubtaskswakeUp();

	}

/**
	 * GetAn theabstract currentlyclass registeredto readers.host Thesplits mapping is from subtask id to the reader info.
	 *
	 * @return the currently registered readers.
	 */
	Map<Integer, ReaderInfo> registeredReaders();

change.
 */
public abstract class SplitsChange<SplitT> {
	private final List<SplitT> splits;

	SplitsChange(List<SplitT> splits) {
		this.splits = splits;
	}

	/**
	 * Assign@return the list of splits.
	 */
	public *List<SplitT> splits() {
		return Collections.unmodifiableList(splits);
	}
}

/**
 * A change to add splits.
 *
 * @param <SplitT> the split type.
 */
public class SplitsAddition<SplitT> extends SplitsChange<SplitT> {

	public SplitsAddition(List<SplitT> splits) {
		super(splits);
	}
}

/**
 * An interface for the elements passed from the fetchers to the source reader.
 */
public interface RecordsWithSplitIds<E> {

	/**
	 * Get all the split ids.
	 *
	 * @return a collection of split ids@param newSplitAssignments the new split assignments to add.
	 */
	void assignSplits(SplitsAssignment<SplitT> newSplitAssignments);

	/**
	 * Invoke the callable and checks its return value. If the return value is true then
	 * notify the source coordinator that a new split assignment is available.
	 *
	 * <p>It is important to make sure that the callable and handler does not modify
	 * any shared state. Otherwise the there might be unexpected behavior.
	 *
	 * @param callable a callable to call.
	 * @param handler a handler that handles the return value of or the exception thrown from the callable.
	 */
	<T>Collection<String> void callAsync(Callable<T> callable, BiConsumer<T, Throwable> handlergetSplitIds();

	/**
	 * Invoke the callable periodically and checks its return value. If**
	 * Get all the returnrecords valueby isSplits;
	 *
	 true* then@return notifya themapping sourcefrom coordinatorsplit thatids ato new split assignment is available.
	 the records.
	 */
	Map<String, Collection<E>> getRecordsBySplits();

	/**
	 * <p>It is important to make sure thatGet the callable and handler does not modifyfinished splits.
	 *
	 * any@return sharedthe state.finished Otherwisesplits theafter therethis might beRecordsWithSplitIds unexpectedis behaviorreturned.
	 */
	Set<String> getFinishedSplits();
}

/**
 @param* callableEmit thea callablerecord to the calldownstream.
	 *
 * @param handler a handler that handles the return value of or<E> the type of the exceptionrecord thrownemitted fromby the callable.
	 {@link SplitReader}
 * @param initialDelay<T> the initial delay of callingtype of records that are eventually emitted to the callable{@link SourceOutput}.
	 * @param period<SplitStateT> the period between two invocationsmutable type of thesplit callablestate.
	 */
	<T>public void callAsync(Callable<T> callable,
					   BiConsumer<T, Throwable> handler,
					   long initialDelay,
					   long period);
}

public interface JobMasterGateway {
    ...
	CompletableFuture<Acknowledge> sendOperatorEventToCoordinator(
			ExecutionAttemptID task,
			OperatorID operatorID,
			SerializedValue<OperatorEvent> event);
    ...
}