This closes #4630.

Move metrics handlers under o.a.f.runtime.webmonitor.handlers

Move StaticFileServerHandler under o.a.f.runtime.webmonitor.files

This closes #4600.

With this change, the complete REST address (protocol://hostname:port) is retrieved
from the RestfulGateway. That way we decouple the RestHandlers from the underlying
RestServerEndpoint/WebRuntimeMonitor because they no longer have to know whether
HTTPs is enabled or not.

This closes #4599.

This commit creates the DispatcherRestEndpoint and integrates it with the
Dispatcher. The DispatcherRestEndpoint is created in the SessionClusterEntrypoint
and its address is passed to the Dispatcher such that it can answer the
requestRestAddress RPC.

This closes #4598.

By letting the AbstractRestHandler extend the RedirectHandler, we add redirection
capabilities to the AbstractRestHandler.

This closes #4597.

This closes #4662.

This closes #4677.

This closes #4663.

This fixes the netty check and makes all of them more robust against failures of
the executed commands counting the number of dependencies.

This closes #4653.

This closes #4646.

This closes #4644.

The TaskExecutorITCase does not wait for the leader election to have finished
before sending remote messages to the ResourceManager. Waiting on the completion
fixed the test instability.

The LeaderGatewayRetriever implementations, AkkaJobManagerRetriever and the
RpcGatewayRetriever, now automatically retry the gateway retrieval operation
for a fixed number of times with a retry delay before completing the gateway
future with an exception.

Retry AkkaJobManagerRetriever

Retry RpcGatewayRetriever

Add support for fenced components

This closes #4643.

Add test case

Only log LeaderGatewayRetriever exception on Debug log level

Properly fail outdated gateway retrieval operations

This closes #4602.

Introduce RedirectHandler which can be extended to add redirection functionality to all
SimpleInboundChannelHandlers. This allows to share the same functionality across the
StaticFileServerHandler and the RuntimeMonitorHandlerBase which could now be removed.
In the future, the AbstractRestHandler will also extend the RedirectHandler.

This closes #4551.

Introduce a generalized GatewayRetriever replacing the JobManagerRetriever. The
GatewayRetriever fulfills the same purpose as the JobManagerRetriever with the
ability to retrieve the gateway for an arbitrary endpoint type.

This closes #4549.

This closes #4673

This closes #4632

This closes #4562

This closes #4676

This closes #4575

Program output is written to TaskManager logs.

This closes #4641

The Simplify parameter should accept and set the parallelism when
calling the Simplify algorithms.

The LocalClusteringCoefficient "count triangles" reduce now uses the
assigned ("little") parallelism as this computation is proportional to
the number of vertices (the combine computation is proportional to the
potentially much larger number of triangles).

The ignored CombineHint on the HITS all-reduces have been removed.

This closes #4346

There exist some tests with empty graphs but the `EmptyGraph` in
`AsmTestBase` contained vertices but no edges. Add a new `EmptyGraph`
without vertices and test both empty graphs for each algorithm.

EmptyGraph now generates the proper TypeInformation (for Edge<> not
Tuple3) which had prevented adding edges due to a union incompatibility.

GraphGeneratorUtils#vertexSet now uses a hash-combine for distinct.

`PageRank` optionally includes zero-degree vertices in the results
(at a performance cost).

This closes #4405

[FLINK-7617] [runtime] Remove string format in BitSet to improve the performance of BuildSideOuterjoin

This closes #4668.

This closes #4622.

- Improve deprecation message in Javadocs
- Remove usage of ProducerConfigConstants in code wherever possible
- Remove usage of ProducerConfigConstants in documentation code snippets

This closes #4473.

[FLINK-7440] [kinesis] Eagerly check that provided schema and partitioner is serializable in FlinkKinesisProducer

This commit also adds a test to verify that the FlinkKinesisProducer is
serializable.

This closes #4537.

[FLINK-7440] [kinesis] Eagerly check serializability of deserialization schema in FlinkKinesisConsumer

This commit also adds tests for verifying that the FlinkKinesisConsumer
itself is serializable.

This closes #4526.

Previously, the AbstractPartitionDiscoverer tracked discovered
partitions by keeping only the largest discovered partition id. All
fetched partition metadata with ids smaller than this id would be
considered as discovered. This assumption of contiguous partition ids is
too naive for corner cases where there may be undiscovered partitions
that were temporariliy unavilable before and were shadowed by
discoverered partitions with largerer partition ids.

This commit changes to use a set to track seen partitions. This also
removes the need of pre-sorting fetched partitions.

Closing StreamOperators is still part of the StreamTask's running lifecycle,
because operators which perform asynchronous operations usually finish their
work when the StreamOperator is closed. Since this also entails that errors
can occur and that a checkpointing operation is triggered, we should only set
the StreamTask's isRunning to false after all StreamOperators have been closed.

Furthermore, this commit introduces a while guard for the waiting condition in
ContinuousFileReaderOperator#close.

This closes #4650.

FutureUtils.retryWithDelay executes the given operation of type
Callable<CompletableFuture<T>> n times and waits in between retries the given
delay. This allows to retry an operation with a specified delay.

Make retry and retry with delay future properly cancellable

This closes #4637.