[FLINK-14993][metrics] Store delimiter in reporter settings

flink-metrics/flink-metrics-jmx/src/test/java/org/apache/flink/metrics/jmx/JMXReporterTest.java

@@ -129,8 +129,8 @@ public Integer getValue() {
  }
  };
 
- rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(new ReporterScopedSettings(0), mg));
- rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(new ReporterScopedSettings(0), mg));
+ rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(createReporterScopedSettings(0), mg));
+ rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(createReporterScopedSettings(0), mg));
 
  MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
 
@@ -183,9 +183,9 @@ public Integer getValue() {
  }
  };
 
- rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(new ReporterScopedSettings(0), mg));
+ rep1.notifyOfAddedMetric(g1, "rep1", new FrontMetricGroup<>(createReporterScopedSettings(0), mg));
 
- rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(new ReporterScopedSettings(1), mg));
+ rep2.notifyOfAddedMetric(g2, "rep2", new FrontMetricGroup<>(createReporterScopedSettings(1), mg));
 
  ObjectName objectName1 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep1", JMXReporter.generateJmxTable(mg.getAllVariables()));
  ObjectName objectName2 = new ObjectName(JMX_DOMAIN_PREFIX + "taskmanager.rep2", JMXReporter.generateJmxTable(mg.getAllVariables()));
@@ -304,4 +304,8 @@ public void testMeterReporting() throws Exception {
  }
  }
  }
+
+ private static ReporterScopedSettings createReporterScopedSettings(int reporterIndex) {
+ return new ReporterScopedSettings(reporterIndex, ',');
+ }
 }

flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusReporterTest.java

@@ -87,7 +87,7 @@ public void setupReporter() {
  MetricRegistryConfiguration.defaultMetricRegistryConfiguration(),
  Collections.singletonList(createReporterSetup("test1", portRangeProvider.next())));
  metricGroup = new FrontMetricGroup<>(
- new ReporterScopedSettings(0),
+ createReporterScopedSettings(),
  new TaskManagerMetricGroup(registry, HOST_NAME, TASK_MANAGER));
  reporter = (PrometheusReporter) registry.getReporters().get(0);
  }
@@ -176,13 +176,13 @@ public void metricIsRemovedWhenCollectorIsNotUnregisteredYet() throws UnirestExc
 
  Counter metric1 = new SimpleCounter();
  FrontMetricGroup<TaskManagerJobMetricGroup> metricGroup1 = new FrontMetricGroup<>(
- new ReporterScopedSettings(0),
+ createReporterScopedSettings(),
  new TaskManagerJobMetricGroup(registry, tmMetricGroup, JobID.generate(), "job_1"));
  reporter.notifyOfAddedMetric(metric1, metricName, metricGroup1);
 
  Counter metric2 = new SimpleCounter();
  FrontMetricGroup<TaskManagerJobMetricGroup> metricGroup2 = new FrontMetricGroup<>(
- new ReporterScopedSettings(0),
+ createReporterScopedSettings(),
  new TaskManagerJobMetricGroup(registry, tmMetricGroup, JobID.generate(), "job_2"));
  reporter.notifyOfAddedMetric(metric2, metricName, metricGroup2);
 
@@ -345,4 +345,8 @@ public String next() {
  return String.valueOf(lowEnd) + "-" + String.valueOf(highEnd);
  }
  }
+
+ private static ReporterScopedSettings createReporterScopedSettings() {
+ return new ReporterScopedSettings(0, ',');
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistry.java

@@ -37,14 +37,6 @@ public interface MetricRegistry {
  */
  char getDelimiter();
 
- /**
- * Returns the configured delimiter for the reporter with the given index.
- *
- * @param index index of the reporter whose delimiter should be used
- * @return configured reporter delimiter, or global delimiter if index is invalid
- */
- char getDelimiter(int index);
-
  /**
  * Returns the number of registered reporters.
  */

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricRegistryImpl.java

@@ -71,7 +71,6 @@ public class MetricRegistryImpl implements MetricRegistry {
 
  private final ScopeFormats scopeFormats;
  private final char globalDelimiter;
- private final List<Character> delimiters;
 
  private final CompletableFuture<Void> terminationFuture;
 
@@ -98,7 +97,6 @@ public MetricRegistryImpl(MetricRegistryConfiguration config, Collection<Reporte
  this.maximumFramesize = config.getQueryServiceMessageSizeLimit();
  this.scopeFormats = config.getScopeFormats();
  this.globalDelimiter = config.getDelimiter();
- this.delimiters = new ArrayList<>(10);
  this.terminationFuture = new CompletableFuture<>();
  this.isShutdown = false;
 
@@ -147,14 +145,18 @@ public MetricRegistryImpl(MetricRegistryConfiguration config, Collection<Reporte
  } else {
  LOG.info("Reporting metrics for reporter {} of type {}.", namedReporter, className);
  }
- reporters.add(new ReporterAndSettings(reporterInstance, new ReporterScopedSettings(reporters.size())));
 
  String delimiterForReporter = reporterSetup.getDelimiter().orElse(String.valueOf(globalDelimiter));
  if (delimiterForReporter.length() != 1) {
  LOG.warn("Failed to parse delimiter '{}' for reporter '{}', using global delimiter '{}'.", delimiterForReporter, namedReporter, globalDelimiter);
  delimiterForReporter = String.valueOf(globalDelimiter);
  }
- this.delimiters.add(delimiterForReporter.charAt(0));
+
+ reporters.add(new ReporterAndSettings(
+ reporterInstance,
+ new ReporterScopedSettings(
+ reporters.size(),
+ delimiterForReporter.charAt(0))));
  }
  catch (Throwable t) {
  LOG.error("Could not instantiate metrics reporter {}. Metrics might not be exposed/reported.", namedReporter, t);
@@ -223,10 +225,10 @@ public char getDelimiter() {
  return this.globalDelimiter;
  }
 
- @Override
- public char getDelimiter(int reporterIndex) {
+ @VisibleForTesting
+ char getDelimiter(int reporterIndex) {
  try {
- return delimiters.get(reporterIndex);
+ return reporters.get(reporterIndex).getSettings().getDelimiter();
  } catch (IndexOutOfBoundsException e) {
  LOG.warn("Delimiter for reporter index {} not found, returning global delimiter.", reporterIndex);
  return this.globalDelimiter;

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/NoOpMetricRegistry.java

@@ -40,11 +40,6 @@ public char getDelimiter() {
  return delimiter;
  }
 
- @Override
- public char getDelimiter(int index) {
- return delimiter;
- }
-
  @Override
  public int getNumberReporters() {
  return 0;

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java

@@ -242,7 +242,7 @@ public String getMetricIdentifier(String metricName) {
  */
  @Override
  public String getMetricIdentifier(String metricName, CharacterFilter filter) {
- return getMetricIdentifier(metricName, filter, -1);
+ return getMetricIdentifier(metricName, filter, -1, registry.getDelimiter());
  }
 
  /**
@@ -252,11 +252,11 @@ public String getMetricIdentifier(String metricName, CharacterFilter filter) {
  * @param metricName metric name
  * @param filter character filter which is applied to the scope components if not null.
  * @param reporterIndex index of the reporter whose delimiter should be used
+ * @param delimiter delimiter to use
  * @return fully qualified metric name
  */
- public String getMetricIdentifier(String metricName, CharacterFilter filter, int reporterIndex) {
+ public String getMetricIdentifier(String metricName, CharacterFilter filter, int reporterIndex, char delimiter) {
  if (scopeStrings.length == 0 || (reporterIndex < 0 || reporterIndex >= scopeStrings.length)) {
- char delimiter = registry.getDelimiter();
  String newScopeString;
  if (filter != null) {
  newScopeString = ScopeFormat.concat(filter, delimiter, scopeComponents);
@@ -266,7 +266,6 @@ public String getMetricIdentifier(String metricName, CharacterFilter filter, int
  }
  return newScopeString + delimiter + metricName;
  } else {
- char delimiter = registry.getDelimiter(reporterIndex);
  if (scopeStrings[reporterIndex] == null) {
  if (filter != null) {
  scopeStrings[reporterIndex] = ScopeFormat.concat(filter, delimiter, scopeComponents);

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/FrontMetricGroup.java

@@ -39,16 +39,16 @@ public FrontMetricGroup(ReporterScopedSettings settings, P reference) {
 
  @Override
  public String getMetricIdentifier(String metricName) {
- return parentMetricGroup.getMetricIdentifier(metricName, null, this.settings.getReporterIndex());
+ return parentMetricGroup.getMetricIdentifier(metricName, null, this.settings.getReporterIndex(), this.settings.getDelimiter());
  }
 
  @Override
  public String getMetricIdentifier(String metricName, CharacterFilter filter) {
- return parentMetricGroup.getMetricIdentifier(metricName, filter, this.settings.getReporterIndex());
+ return parentMetricGroup.getMetricIdentifier(metricName, filter, this.settings.getReporterIndex(), this.settings.getDelimiter());
  }
 
  public String getLogicalScope(CharacterFilter filter) {
- return parentMetricGroup.getLogicalScope(filter);
+ return parentMetricGroup.getLogicalScope(filter, this.settings.getDelimiter());
  }
 
  public String getLogicalScope(CharacterFilter filter, char delimiter) {

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/ReporterScopedSettings.java

@@ -26,12 +26,19 @@ public class ReporterScopedSettings {
 
  private final int reporterIndex;
 
- public ReporterScopedSettings(int reporterIndex) {
+ private final char delimiter;
+
+ public ReporterScopedSettings(int reporterIndex, char delimiter) {
  Preconditions.checkArgument(reporterIndex >= 0);
  this.reporterIndex = reporterIndex;
+ this.delimiter = delimiter;
  }
 
  public int getReporterIndex() {
  return reporterIndex;
  }
+
+ public char getDelimiter() {
+ return delimiter;
+ }
 }

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroupTest.java

@@ -266,8 +266,8 @@ public void testScopeGenerationWithoutReporters() throws Exception {
  // no caching should occur
  assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B));
  // invalid reporter indices do not throw errors
- assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B, -1));
- assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B, 2));
+ assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B, -1, '.'));
+ assertEquals("A.X.C.D.1", group.getMetricIdentifier("1", FILTER_B, 2, '.'));
  } finally {
  testRegistry.shutdown().get();
  }
@@ -326,11 +326,6 @@ public char getDelimiter() {
  return 0;
  }
 
- @Override
- public char getDelimiter(int index) {
- return 0;
- }
-
  @Override
  public int getNumberReporters() {
  return 0;

flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/RocksDBNativeMetricMonitorTest.java

@@ -168,11 +168,6 @@ public char getDelimiter() {
  return 0;
  }
 
- @Override
- public char getDelimiter(int index) {
- return 0;
- }
-
  @Override
  public int getNumberReporters() {
  return 0;

flink-streaming-java/src/test/java/org/apache/flink/streaming/util/LatencyStatsTest.java

@@ -186,11 +186,6 @@ public char getDelimiter() {
  return '.';
  }
 
- @Override
- public char getDelimiter(int index) {
- return 0;
- }
-
  @Override
  public int getNumberReporters() {
  return 0;