[FLINK-19901][metrics] Fix caching offset for variables

deepfield · Nov 3, 2020 · 477d37d · 477d37d
1 parent 6ed81d9
commit 477d37d
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 3 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java
@@ -120,11 +120,14 @@ public Map<String, String> getAllVariables() {
  }
 
  public Map<String, String> getAllVariables(int reporterIndex, Set<String> excludedVariables) {
- // offset cache location to account for general cache at position 0
- reporterIndex += 1;
  if (reporterIndex < 0 || reporterIndex >= logicalScopeStrings.length) {
- reporterIndex = 0;
+ // invalid reporter index; either a programming mistake, or we try to retrieve variables outside of a reporter
+ reporterIndex = -1;
  }
+
+ // offset cache location to account for general cache at position 0
+ reporterIndex += 1;
+
  // if no variables are excluded (which is the default!) we re-use the general variables map to save space
  return internalGetAllVariables(excludedVariables.isEmpty() ? 0 : reporterIndex, excludedVariables);
  }

diff --git a/...untime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java b/...untime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java
@@ -43,9 +43,11 @@
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsNot.not;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -93,6 +95,26 @@ public void testGetAllVariablesWithExclusions() {
  assertEquals(group.getAllVariables(-1, Collections.singleton(ScopeFormat.SCOPE_HOST)).size(), 0);
  }
 
+ @Test
+ public void testGetAllVariablesWithExclusionsForReporters() {
+ MetricRegistry registry = TestingMetricRegistry.builder().setNumberReporters(2).build();
+
+ AbstractMetricGroup<?> group = new GenericMetricGroup(registry, null, "test") {
+ @Override
+ protected void putVariables(Map<String, String> variables) {
+ variables.put("k1", "v1");
+ variables.put("k2", "v2");
+ }
+ };
+
+ group.getAllVariables(-1, Collections.emptySet());
+
+ assertThat(group.getAllVariables(0, Collections.singleton("k1")), not(IsMapContaining.hasKey("k1")));
+ assertThat(group.getAllVariables(0, Collections.singleton("k1")), IsMapContaining.hasKey("k2"));
+ assertThat(group.getAllVariables(1, Collections.singleton("k2")), IsMapContaining.hasKey("k1"));
+ assertThat(group.getAllVariables(1, Collections.singleton("k2")), not(IsMapContaining.hasKey("k2")));
+ }
+
  // ========================================================================
  // Scope Caching
  // ========================================================================