It adds a method failExternally() to the StreamTask, so that custom Operators
can make their containing task fail when needed.

This adds a new interface called CheckpointableInputFormat
which describes input formats whose state is queryable,
i.e. getCurrentChannelState() returns where the reader is
in the underlying source, and they can resume reading from
a user-specified position.

This functionality is not yet leveraged by current readers.

This is meant to replace the different file
reading sources in Flink streaming. Now there is
one monitoring source with DOP 1 monitoring a
directory and assigning input split to downstream
readers.

In addition, it makes the new features added by
FLINK-3717 work together with the aforementioned entities
(the monitor and the readers) in order to have
fault tolerant file sources and exactly once guarantees.

This does not replace the old API calls. This
will be done in a future commit.

This commit takes the changes from the previous
commits and wires them into the API, both Java and Scala.

While doing so, some changes were introduced to the
classes actually doing the work, either as bug fixes, or
as new design choices.