This closes #10899.

[FLINK-17218][tests] Adding recoverable failures and correctness checks to UnalignedCheckpointITCase.

This closes #12037

It fixes the behavior: when we take the union of two side outputs. Flink
was repeating the data from one twice.

This closes #11983

[FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint

During recovery process for unaligned checkpoint, the input channel state should also be recovered besides with existing operator states.

We considered three guarantees during the implementation:
1. Make input recovery happen after the output recovery for providing more floating buffers on output side firstly.
2. Make partition request happen after input recovery for avoiding new data overtaking the previous state data.
3. Introduce a dedicated single IO executor for unspilling the channel state one by one, to avoid potential random IO.

This closes #11687.

Apparently, it's causing deadlocks.

This commit is the first part of implementation to solve the dead lock problem when reducing the exclusive buffer of receiver side to 0.

Reducing the number of exclusive buffers of receiver side to 0 can bring several advantages (may at the cost of some performance regression). One is that memory can be saved from the reduced network buffer usage. Another important benefit is that the in-flight data can be reduced so we can speed up checkpoint in cases of back pressure. However, for the current implementation, reducing the exclusive buffer of receiver side can incur deadlock problem because all the floating buffers might be requested away by some blocked input channels and never recycled until barrier alignment.

To solve the problem, this commit mainly makes the following changes:
1. At sender side, after sending a checkpoint barrier when aligned exactly-once checkpoint mode is used, the outgoing channel will be blocked and no data will be sent out until the channel is unblocked.
2. At receiver side, no buffer will be stored in BufferStorage any more and after a checkpoint is completed or canceled, the receiver side will resume data consumption and unblock the upstream by sending a special event to the sender side.

Note that after this patch we still can't set the exclusive buffer of receiver side to 0 because there is still deadlock problem which will be totally solved in the following up patches.

This closes #11841

This reverts commit 300263e1.

[FLINK-16587][checkpointing] Fix abortion of ChannelStateWriter and add test/IT cases for unaligned checkpoint.

In ChannelStateCheckpointWriter#fail, DataOutputStream flushes on close, which is not possible when checkpointStream is closed before. That can happen either directly in fail or through a previous finishWriteAndResult.
DataOutputStream#close does not need to be closed in both cases as long as checkpointStream is closed.
checkpointStream#close is idempotent.

This closes #11624

StateBackendFactories do not need a full read and write access to the
Configuration object. It's sufficient to have read only access. Moreover
the ReadableConfig is a lightweight interface that can be implemented in
other ways, not just through the Configuration. Lastly we exposed this
lightweight interface as a configuration entry point for
ExecutionEnvironments. This change will make it possible to pass the
ReadableConfig directly to the StateBackendFactories without fragile
adapters.

There is a race condition in `LocalExecutor` between (a) shutting down the
cluster when the job has finished and (b) the client which retrieves the result
of the job execution.

This was observed in Beam, running a large test suite with the Flink Runner.

We should make sure the job result retrieval and the cluster shutdown do not
interfere. This adds a PerJobMiniClusterClient which guarantees that.

Improve message for running flag state checks in MiniCluster

Additionally check for the JobID in PerJobMiniClusterClient

Introduce PerJobMiniCluster and a corresponding JobClient

Add TestLogger to test

Convert shutdown methods to be async

This closes #11473.

Because the classloader is now closed when the task fails the UDF only has access to the bootstrap classloader, which doesn't contain our own test classes.

[FLINK-16317][operators] Provide support for key selector and latency marker in MultipleInputStreamOperator

[FLINK-16316][operators] Implement new AbstractStreamOperatorV2 as a replacement for AbstractStreamOperator

The new base class for operators tries to address couple of limitations in the AbstractStreamOperator like:
- lack of support for multiple inputs
- setup(...) method

New POJO class will make Public and PublicEvolving interfaces more stable
and easier to use. User will not have to pass n parameters, but just this
one POJO.

This removes another dependency on the StreamTask from AbstractStreamOperator

At the moment there are four fields initialized in the method of StreamTask#beforeInvoke, such as `stateBackend`, `checkpointStorage`, `timerService`, `asyncOperationsThreadPool`.

In general it is suggested to use final fields to get known benefits. So we can refactor the StreamTask to initialize these fields in the constructor instead.

This closes #11486

[FLINK-15911][runtime] Refactor to create AkkaRpsService with builder class, in order to reduce the number of nested creating methods.

The way of detecting external checkpoint is done or not is not strict.
The finalization of checkpoint might be interrupted by the cancallation.