Add updateGlobalMetric methods #154
Conversation
| public static synchronized void updateGlobalMetrics() { | ||
| FlinkMetricContainer container = GLOBAL_FLINK_METRIC_CONTAINER.get(); | ||
| if (container == null) { | ||
| LOG.warn("Cannot update global metrics: FlinkMetricContainer reference not set"); |
There was a problem hiding this comment.
Yes, it is possible, since we use dedicated thread to updateGlobalMetircs, and setGlobalMetrics is inited in a different thread. It is possible that the gloableMetrics is not inited and then we call updateGlobalMetircs
| * update metrics concurrently, which could lead to incorrect metric values or duplicate metric | ||
| * registrations in Flink's metrics system. | ||
| */ | ||
| public static synchronized void updateGlobalMetrics() { |
There was a problem hiding this comment.
can updateGlobalMetrics get called before setGlobalMetrics?
There was a problem hiding this comment.
Yes, it is possible, since we use dedicated thread to updateGlobalMetircs, and setGlobalMetrics is inited in a different thread. It is possible that the gloableMetrics is not inited and then we call updateGlobalMetircs
| } | ||
|
|
||
| try { | ||
| container.updateMetrics(GLOBAL_CONTAINER_STEP_NAME); |
There was a problem hiding this comment.
where does container.updateMetrics gets called in the current implementation?
There was a problem hiding this comment.
It will be called at beam-runner library. Will share you PR.
| private static final Logger LOG = LoggerFactory.getLogger(GlobalMetricsUtils.class); | ||
| private static final String GLOBAL_CONTAINER_STEP_NAME = "GLOBAL_METRICS"; | ||
|
|
||
| // Maintain a reference to the FlinkMetricContainer for updating global metrics |
There was a problem hiding this comment.
can we add more context on why this is needed?
2 participants
Summary:
These changes enable the periodic publishing of global metrics from non-main threads (async callbacks, background tasks) to Flink's metrics system. The updateGlobalMetrics() method is the key addition that allows a background timer to push accumulated metrics to Flink at regular intervals.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI.