Migrate logic from
org.apache.flink.runtime.rest.handler.legacy.metrics.SubtaskMetricsHandler to
new handler. Add new handler to DispatcherRestEndpoint.

[FLINK-8143][flip6] Assert that SubtaskIndexPathParameter is mandatory

[FLINK-8143][flip6] Use path parameter constants in SubtaskMetricsHandlerTest

This closes #5082.

Migrate logic in
org.apache.flink.runtime.rest.handler.legacy.metrics.JobManagerMetricsHandler
to new handler and add new handler to DispatcherRestEndpoint.

[FLINK-7716][Javadoc] Deprecate method MetricStore#getJobManager().

There is a semantically equivalent method in MetricStore.

This closes #5083.

Migrate logic in
org.apache.flink.runtime.rest.handler.legacy.metrics.TaskManagerMetricsHandler
to new handler, and add new handler to DispatcherRestEndpoint.

[FLINK-7717][flip6] Use taskmanagerid constant in TaskManagerMetricsHandlerTest

This closes #5081.

Before, the TaskExecutor's InstanceID was exposed as TaskExecutor id. This was wrong
since the InstanceID is bound the registration of a TaskExecutor whereas the
ResourceID is bound to the lifetime of the TaskExecutor. Thus, it is better to identify
the TaskExecutor by its ResourceID which does not change.

This commit changes the behaviour accordingly on the ResourceManager and the
TaskManagerDetailsHandler.

This closes #5093.

This closes #5001.

This closes #4574

When printing the job plan the operator description is typically
formatted as the operator name followed by the user given or generated
function name in parenthesis. For example, "Reduce (My Function)".

This normalizes the node names to include a space between the operator
and function names.

This closes #4383

Add "Remote connection to [null] failed with java.nio.channels.NotYetConnectedException"
to the list of whitelisted log statements in YarnTestBase. This logging statement
seems to appear since we moved from Flakka to Akka 2.4.0.

This closes #5085.

This closes #4748

This closes #4946

This closes #5034

This closes #5067

This closes #5061

This closes #5077.

This closes #5049.

Increasing timeout for reading the records from 30s to 60s seems to solve the issue
for failing at-least-one tests.

Migrate logic in
org.apache.flink.runtime.rest.handler.legacy.metrics.JobVertexMetricsHandler to
new handler and add new handler to DispatcherRestEndpoint. Add common classes
for remaining implementations of
org.apache.flink.runtime.rest.handler.legacy.metrics.AbstractMetricsHandler,
which require migration as well.

[FLINK-7718] [flip6] Clean up JobVertexMetricsHandlerHeaders

[FLINK-7718] [flip6] Assert that HTTP code is 404 if metric is unknown

[FLINK-7718] [flip6] Minor fixes in Javadocs

[FLINK-7718] [flip6] Add unit test for AbstractMetricsHandlerHeaders

[FLINK-7718] [flip6] Let unit tests inherit from TestLogger

[FLINK-7718] [flip6] Re-format Metric constructor

[FLINK-7718] [flip6] Fix mistake in Javadoc of AbstractMetricsHandlerHeaders

[FLINK-7718] [flip6] Rename AbstractMetricsHandlerHeaders to AbstractMetricsHeaders

Strip the term Handler from the Header class. Also rename its subclasses.

[FLINK-7718] [flip6] No longer return HTTP 404 if metric is unknown

[FLINK-7718] [flip6] Make JobVertexMetricsHeaders class final

[FLINK-7718] [flip6] Introduce MetricsHandlerTestBase for future MetricHandlers

[FLINK-7718] [flip6] Always return same MessageParameter objects in JobVertexMetricsMessageParameters

This closes #5055.

This is another step at using or own (off-heap) buffers for network
communication that we pass through netty in order to avoid unnecessary buffer
copies.

This closes #4481.

This was causing the snapshot deployment to fail.

[FLINK-5465] [streaming] Wait for pending timer threads to finish or to exceed a time limit in exceptional stream task shutdown.

This closes #5058.

This closes #5056.

Previously faulty scenario with producer pool of 2.

1. started transaction 1 with producerA, written record 42
2. checkpoint 1 triggered, pre committing txn1, started txn2 with producerB, written record 43
3. checkpoint 1 completed, committing txn1, returning producerA to the pool
4. checkpoint 2 triggered , committing txn2, started txn3 with producerA, written record 44
5. crash....
6. recover to checkpoint 1, txn1 from producerA found to "pendingCommitTransactions", attempting to recoverAndCommit(txn1)
7. unfortunately txn1 and txn3 from the same producers are identical from KafkaBroker perspective and thus txn3 is being committed

result is that both records 42 and 44 are committed.

With this fix, after re-initialization txn3 will have different producerId/epoch counters compared to txn1.

This closes #5043.