This closes #3367

Performin TestingListener#waitForNewLeader under the lock which is also hold when
updating the leader information makes sure that leader changes won't go unnoticed.
This led before to failing test cases due to timeouts.

This closes #3853.

This closes #3736.

This closes #3816.

This closes #3789.

This closes #3831.

This closes #3552.

This closes #3824.

This closes #1033
This closes #1944
This closes #1979
This closes #2029
This closes #2809
This closes #3043
This closes #3627

This closes #3619.

This closes #3318

This closes #3849.

This closes #3850.

This closes #3815.

This closes #3841.

This closes #3823

This closes #3821

This closes #3803.

[FLINK-6475] [checkpoint] Incremental snapshots in RocksDB should not hold lock during async file upload

[FLINK-6471] [checkpoint] Fix RocksDBStateBackendTest#testCancelRunningSnapshot failing sporadically

Previously, the CompatibilityResult class accepts a full-blown
TypeSerializer for its convert deserializer, which will actually only
ever be used for deserialization.

This commit narrows down the interface by introducing a new
TypeDeserializer interface that contains only the read methods.

This closes #3834.
This closes #3804.

This commit adds the functionality of allowing serializer upgrades for
Flink's managed state. It consists of 2 major changes: 1) new
user-facing API in `TypeSerializer`, and 2) activate serializer upgrades
in state backends.

For 1) new user-facing API for `TypeSerializer`, the following is added:
- new class: TypeSerializerConfigSnapshot
- new class: CompatibilityResult
- new method: TypeSerializer#snapshotConfiguration()
- new method:
  TypeSerializer#ensureCompatibility(TypeSerializerConfigSnapshot)

Generally speaking, configuration snapshots contains a point-in-time
view of a serializer's state / configuration, and is persisted along
with checkpoints. On restore, the configuration is confronted with the
new serializer of the state to check for compatibility, which may
introduce reconfiguration of the new serializer to be compatible.

This compatibility check is integrated in the state backends' restore
flow in 2). Currently, if the check results in the need to perform state
migration, the restore simply fails as the state migration feature isn't
yet available.

Previously, the TypeSerializerSerializationProxy only uses the dummy
ClassNotFoundDummyTypeSerializer as a placeholder in the case where the
user uses a completely new serializer and deletes the old one.

There is also the case where the user changes the original serializer's
implementation and results in an InvalidClassException when trying to
deserialize the serializer. We should also use the
ClassNotFoundDummyTypeSerializer as a temporary placeholder in this
case.

This closes #3793.

This closes #3697.

This commit adds some general improvements to the rescalable
implementation of FlinkKinesisConsumer, including:
- Refactor setup procedures in KinesisDataFetcher so that duplicate work
  isn't done on a restored run
- Strengthen corner cases where fetcher was not fully seeded with
  initial state when snapshot is taken

This closes #3001.

This method exploits the fact that verties are already in topological order.

This closes #3773

[FLINK-5867] [flip-1] Support restarting only pipelined sub-regions of the ExecutionGraph on task failure

This closes #3822

This closes #3828

This preserves old config keys as deprecated keys where the key was already present
in an earlier release.

This also re-arranges config options to form logical sections in the file
and harmonized JavaDoc formatting style.