[HUDI-2974] Make the prefix for metrics name configurable

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -1628,6 +1628,10 @@ public boolean getPushGatewayRandomJobNameSuffix() {
     return getBoolean(HoodieMetricsPrometheusConfig.PUSHGATEWAY_RANDOM_JOBNAME_SUFFIX);
   }
 
+  public String getMetricReporterMetricsNamePrefix() {
+    return getStringOrDefault(HoodieMetricsConfig.METRICS_REPORTER_PREFIX);
+  }
+
   /**
    * memory configs.
    */

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/metrics/HoodieMetricsConfig.java

@@ -22,6 +22,8 @@
 import org.apache.hudi.common.config.ConfigGroups;
 import org.apache.hudi.common.config.ConfigProperty;
 import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieMetricsCloudWatchConfig;
 import org.apache.hudi.metrics.MetricsReporterType;
 
@@ -63,6 +65,18 @@ public class HoodieMetricsConfig extends HoodieConfig {
       .sinceVersion("0.6.0")
       .withDocumentation("");
 
+  public static final ConfigProperty<String> METRICS_REPORTER_PREFIX = ConfigProperty
+      .key(METRIC_PREFIX + ".reporter.metricsname.prefix")
+      .defaultValue("")
+      .sinceVersion("0.11.0")
+      .withInferFunction(cfg -> {
+        if (cfg.contains(HoodieTableConfig.NAME)) {
+          return Option.of(cfg.getString(HoodieTableConfig.NAME));
+        }
+        return Option.empty();
+      })
+      .withDocumentation("The prefix given to the metrics names.");
+
   // Enable metrics collection from executors
   public static final ConfigProperty<String> EXECUTOR_METRICS_ENABLE = ConfigProperty
       .key(METRIC_PREFIX + ".executor.enable")

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java

@@ -215,7 +215,7 @@ public void updateIndexMetrics(final String action, final long durationInMs) {
   }
 
   String getMetricsName(String action, String metric) {
-    return config == null ? null : String.format("%s.%s.%s", tableName, action, metric);
+    return config == null ? null : String.format("%s.%s.%s", config.getMetricReporterMetricsNamePrefix(), action, metric);
   }
 
   /**

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/Metrics.java

@@ -46,7 +46,7 @@ public class Metrics {
 
   private Metrics(HoodieWriteConfig metricConfig) {
     registry = new MetricRegistry();
-    commonMetricPrefix = metricConfig.getTableName();
+    commonMetricPrefix = metricConfig.getMetricReporterMetricsNamePrefix();
     reporter = MetricsReporterFactory.createReporter(metricConfig, registry);
     if (reporter == null) {
       throw new RuntimeException("Cannot initialize Reporter.");

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerMetrics.java

@@ -74,7 +74,7 @@ private Timer createTimer(String name) {
   }
 
   String getMetricsName(String action, String metric) {
-    return config == null ? null : String.format("%s.%s.%s", tableName, action, metric);
+    return config == null ? null : String.format("%s.%s.%s", config.getMetricReporterMetricsNamePrefix(), action, metric);
   }
 
   public void updateDeltaStreamerMetrics(long durationInNs) {