[FLINK-24775][coordination] move JobStatus-related metrics out of ExecutionGraph

[zentol]

Based on #17722.

The down-/up-/restartTime metrics are now setup in the schedulersinstead of the ExecutionGraph, similar to the numRestart metrics.
To this end the AdaptiveScheduler now maintains its own set of state timestamps, according to the job state transitions that the scheduler advertises (i.e., they are not based on the ExecutionGraph).

This prevents collisions upon rescaling as they are now only registered once.

[flinkbot]

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit 075f8ee (Tue Nov 09 12:28:16 UTC 2021)

Warnings:

• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
• @flinkbot approve all to approve all aspects
• @flinkbot approve-until architecture to approve everything until architecture
• @flinkbot attention @username1 [@username2 ..] to require somebody's attention
• @flinkbot disapprove architecture to remove an approval you gave earlier

[flinkbot]

CI report:

• 4db2506 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[dmvk]

The change LGTM overall 👍 My main question would be whether we could unify the JobStatus metrics between DefaultScheduler / AdaptiveScheduler / ExecutionGraph a bit more, by reusing the new JobStatusStore and registerMetrics().

[dmvk]

What do you think about passing a Clock instance to Adaptive scheduler instead of relying on System.currentTimeMillis() to simplify the test?

[zentol]

I generally like doing that, but I'm wondering if this would work properly for the AdaptiveScheduler in that truly all time-measurements go through the clock. For smaller self-contained components it is easy to ensure that, but this isn't the case here because we re-use some parts of the SchedulerBase/DefaultScheduler, there are multiple state classes, then internally there is the EG, ....
It would be a bit unsatisfactory to introduce a clock but only use it in one place :/

[dmvk]

LGTM, great job ;)

[dmvk]

Does numRestarts need to be volatile? If I understand that correctly if we access the metric eg. using JMX, then it gets accessed by a different thread. Or is there some synchronization in the metrics system that I'm missing?

[zentol]

Does numRestarts need to be volatile?

Technically yes, but we generally don't do it. It would be too expensive on the hot code paths (aka, we can't be consistent about it anyway), and we haven't had issues so far 🤷

Or is there some synchronization in the metrics system that I'm missing?

There is none.