[FLINK-38703][runtime] Update slot manager metrics in thread-safety manner

[ztison]

What is the purpose of the change

This PR resolves a race condition that leads to a ConcurrentModificationException when OpenTelemetry (OTel) metrics are collected simultaneously with updates to the tracked task managers in the SlotManager.

Brief change log

• The new fields for the metrics are introduced
• These fields are periodically updated on the main thread every 1s
• OpenTelemetry reads these static fields

Verifying this change

This change added tests and can be verified as follows:

• New test was added that verifies the fields are periodically updated and values are propagated to the OpenTelemetry collector.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[flinkbot]

CI report:

• 1ae156a Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[ztison]

CI is not passing due to a failing integration test.

Failures: 
Nov 20 17:36:29 17:36:29.834 [ERROR]   JobManagerMetricsITCase.testJobManagerMetrics:136 expected:<1> but was:<0>

The reason is that the failing test reads metrics before the fields were updated (interval 1s).
Update: I added a sleep to the test.

[ztison]

@flinkbot run azur

[davidradl]

I wonder if it would not be simpler to synchronise on slotManagerMetricGroup around the original 2 lines. WDYT?

[ztison]

Hi @davidradl, thanks for looking at this PR.

Can you elaborate more on what you mean?

If I understand that correctly then a synchronization on slotManagerMetricGroup level wouldn't help because the ConcurrentModificationException is still possible if the FineGrainedTaskManagerTracker#addTaskManager or the FineGrainedTaskManagerTracker#removeTaskManager is called from the main thread at the same time.

[davidradl]

@ztison I see that this change is ensuring that some fields in FineGrainedManager are updated on the same thread, so they cannot be concurrently modified. If the code modifying these fields were synchronized on the FineGrainedManager object, then thread 1 would get the lock and make the updates, then if thread 2 tries to update the same object, it would also request the lock, which it would only get, when thread 1 completed its updates. So it would ensure only one thread ever updated the fields.

[ztison]

@davidradl
The easiest fix would be to make the FineGrainedTaskManagerTracker.taskManagerRegistrations map a concurrent map. But what about the other maps in the FineGrainedTaskManagerTracker? My point is that this component doesn't claim to be thread-safe. I don't want to make it partially thread-safe, as that could confuse future readers of this class. That's why I moved the synchronization logic up.

[davidradl]

@ztison Personally I would go with the synchronise or concurrent map as it solves the same problem without the future, but let's see what others have to say. If others are ok with your approach - I will back it.

[ztison]

Yes, sure. I have already discussed it with @zentol internally. So let's wait for his feedback.
Thanks anyway.

[snuyanzin]

fyi: to have green ci, rebase to the latest master
e2e was fixed at FLINK-38700