Skip to content

include global metrics in updateMetrics for Flink Runner #155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yananhao12 merged 5 commits into from
Dec 3, 2025
Merged

include global metrics in updateMetrics for Flink Runner #155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
16a2aa1
add global metrics to updateMetrics for Flink Runner
yananhao12 Dec 2, 2025
b8daf3b
update beam version
yananhao12 Dec 2, 2025
8aa0cca
fix style
yananhao12 Dec 3, 2025
da2da6b
remove asAttemptedOnlyMetricResultsForSteps
yananhao12 Dec 3, 2025
4b09658
fix style
yananhao12 Dec 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -398,7 +398,7 @@ class BeamModulePlugin implements Plugin<Project> {

// Automatically use the official release version if we are performing a release
// otherwise append '-SNAPSHOT'
project.version = '2.45.47'
project.version = '2.45.48'
if (isLinkedin(project)) {
project.ext.mavenGroupId = 'com.linkedin.beam'
}
Expand Down
4 changes: 2 additions & 2 deletions gradle.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,8 +30,8 @@ signing.gnupg.useLegacyGpg=true
# buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy.
# To build a custom Beam version make sure you change it in both places, see
# https://github.com/apache/beam/issues/21302.
version=2.45.47
sdk_version=2.45.47
version=2.45.48
sdk_version=2.45.48

javaVersion=1.8

Expand Down
11 changes: 7 additions & 4 deletions ...s/flink/src/main/java/org/apache/beam/runners/flink/metrics/FlinkMetricContainerBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@
*/
package org.apache.beam.runners.flink.metrics;

import static org.apache.beam.runners.core.metrics.MetricsContainerStepMap.asAttemptedOnlyMetricResults;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -84,9 +82,14 @@ public void updateMetrics(String stepName, List<MetricsApi.MonitoringInfo> monit
* given step.
*/
void updateMetrics(String stepName) {
MetricResults metricResults = asAttemptedOnlyMetricResults(metricsContainers);
MetricResults metricResults =
MetricsContainerStepMap.asAttemptedOnlyMetricResults(metricsContainers);
MetricQueryResults metricQueryResults =
metricResults.queryMetrics(MetricsFilter.builder().addStep(stepName).build());
metricResults.queryMetrics(
MetricsFilter.builder()
.addStep(stepName)
.addStep(GlobalMetricsUtils.GLOBAL_CONTAINER_STEP_NAME)
.build());
updateCounters(metricQueryResults.getCounters());
updateDistributions(metricQueryResults.getDistributions());
updateGauge(metricQueryResults.getGauges());
Expand Down
2 changes: 1 addition & 1 deletion runners/flink/src/main/java/org/apache/beam/runners/flink/metrics/GlobalMetricsUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@

/** Utility class for managing global metrics in a Flink environment. */
public class GlobalMetricsUtils {
private static final String GLOBAL_CONTAINER_STEP_NAME = "GLOBAL_METRICS";
static final String GLOBAL_CONTAINER_STEP_NAME = "GLOBAL_METRICS";

/**
* Sets the global metrics container if it is not already set.
Expand Down