[HUDI-1113] Add user define metrics reporter (#1851)

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

@@ -58,6 +58,10 @@ public class HoodieMetricsConfig extends DefaultHoodieConfig {
 
   public static final String GRAPHITE_METRIC_PREFIX = GRAPHITE_PREFIX + ".metric.prefix";
 
+  // User defined
+  public static final String METRICS_REPORTER_CLASS = METRIC_PREFIX + ".reporter.class";
+  public static final String DEFAULT_METRICS_REPORTER_CLASS = "";
+
   private HoodieMetricsConfig(Properties props) {
     super(props);
   }
@@ -117,6 +121,11 @@ public class HoodieMetricsConfig extends DefaultHoodieConfig {
       return this;
     }
 
+    public Builder withReporterClass(String className) {
+      props.setProperty(METRICS_REPORTER_CLASS, className);
+      return this;
+    }
+
     public HoodieMetricsConfig build() {
       HoodieMetricsConfig config = new HoodieMetricsConfig(props);
       setDefaultOnCondition(props, !props.containsKey(METRICS_ON), METRICS_ON, String.valueOf(DEFAULT_METRICS_ON));
@@ -133,6 +142,9 @@ public class HoodieMetricsConfig extends DefaultHoodieConfig {
       MetricsReporterType reporterType = MetricsReporterType.valueOf(props.getProperty(METRICS_REPORTER_TYPE));
       setDefaultOnCondition(props, reporterType == MetricsReporterType.DATADOG,
           HoodieMetricsDatadogConfig.newBuilder().fromProperties(props).build());
+      setDefaultOnCondition(props, !props.containsKey(METRICS_REPORTER_CLASS),
+              METRICS_REPORTER_CLASS, DEFAULT_METRICS_REPORTER_CLASS);
+
       return config;
     }
   }

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

@@ -599,6 +599,10 @@ public class HoodieWriteConfig extends DefaultHoodieConfig {
         HoodieMetricsDatadogConfig.DATADOG_METRIC_TAGS).split("\\s*,\\s*")).collect(Collectors.toList());
   }
 
+  public String getMetricReporterClassName() {
+    return props.getProperty(HoodieMetricsConfig.METRICS_REPORTER_CLASS);
+  }
+
   /**
    * memory configs.
    */

hudi-client/src/main/java/org/apache/hudi/metrics/MetricsReporterFactory.java

@@ -18,13 +18,19 @@
 
 package org.apache.hudi.metrics;
 
+import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.metrics.datadog.DatadogMetricsReporter;
 
 import com.codahale.metrics.MetricRegistry;
+import org.apache.hudi.metrics.userdefined.AbstractUserDefinedMetricsReporter;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
+import java.util.Properties;
+
 /**
  * Factory class for creating MetricsReporter.
  */
@@ -35,6 +41,18 @@ public class MetricsReporterFactory {
   public static MetricsReporter createReporter(HoodieWriteConfig config, MetricRegistry registry) {
     MetricsReporterType type = config.getMetricsReporterType();
     MetricsReporter reporter = null;
+
+    if (!StringUtils.isNullOrEmpty(config.getMetricReporterClassName())) {
+      Object instance = ReflectionUtils
+              .loadClass(config.getMetricReporterClassName(),
+                      new Class<?>[] {Properties.class, MetricRegistry.class}, config.getProps(), registry);
+      if (!(instance instanceof AbstractUserDefinedMetricsReporter)) {
+        throw new HoodieException(config.getMetricReporterClassName()
+                + " is not a subclass of AbstractUserDefinedMetricsReporter");
+      }
+      return (MetricsReporter) instance;
+    }
+
     switch (type) {
       case GRAPHITE:
         reporter = new MetricsGraphiteReporter(config, registry);

hudi-client/src/main/java/org/apache/hudi/metrics/userdefined/AbstractUserDefinedMetricsReporter.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.metrics.userdefined;
+
+import com.codahale.metrics.MetricRegistry;
+import org.apache.hudi.metrics.MetricsReporter;
+import java.util.Properties;
+
+/**
+ * Abstract class of user defined metrics reporter.
+ */
+public abstract class AbstractUserDefinedMetricsReporter extends MetricsReporter {
+  private Properties props;
+  private MetricRegistry registry;
+
+  public AbstractUserDefinedMetricsReporter(Properties props, MetricRegistry registry) {
+    this.props = props;
+    this.registry = registry;
+  }
+
+  public Properties getProps() {
+    return props;
+  }
+
+  public MetricRegistry getRegistry() {
+    return registry;
+  }
+}

hudi-client/src/test/java/org/apache/hudi/metrics/TestMetricsReporterFactory.java

@@ -22,11 +22,18 @@ package org.apache.hudi.metrics;
 import org.apache.hudi.config.HoodieWriteConfig;
 
 import com.codahale.metrics.MetricRegistry;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metrics.userdefined.AbstractUserDefinedMetricsReporter;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
+import java.io.Closeable;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.when;
 
@@ -45,4 +52,53 @@ public class TestMetricsReporterFactory {
     MetricsReporter reporter = MetricsReporterFactory.createReporter(config, registry);
     assertTrue(reporter instanceof InMemoryMetricsReporter);
   }
+
+  @Test
+  public void metricsReporterFactoryShouldReturnUserDefinedReporter() {
+    when(config.getMetricReporterClassName()).thenReturn(DummyMetricsReporter.class.getName());
+
+    Properties props = new Properties();
+    props.setProperty("testKey", "testValue");
+
+    when(config.getProps()).thenReturn(props);
+    MetricsReporter reporter = MetricsReporterFactory.createReporter(config, registry);
+    assertTrue(reporter instanceof AbstractUserDefinedMetricsReporter);
+    assertEquals(props, ((DummyMetricsReporter) reporter).getProps());
+    assertEquals(registry, ((DummyMetricsReporter) reporter).getRegistry());
+  }
+
+  @Test
+  public void metricsReporterFactoryShouldThrowExceptionWhenMetricsReporterClassIsIllegal() {
+    when(config.getMetricReporterClassName()).thenReturn(IllegalTestMetricsReporter.class.getName());
+    when(config.getProps()).thenReturn(new Properties());
+    assertThrows(HoodieException.class, () -> MetricsReporterFactory.createReporter(config, registry));
+  }
+
+  public static class DummyMetricsReporter extends AbstractUserDefinedMetricsReporter {
+
+    public DummyMetricsReporter(Properties props, MetricRegistry registry) {
+      super(props, registry);
+    }
+
+    @Override
+    public void start() {}
+
+    @Override
+    public void report() {}
+
+    @Override
+    public Closeable getReporter() {
+      return null;
+    }
+
+    @Override
+    public void stop() {}
+  }
+
+  public static class IllegalTestMetricsReporter {
+
+    public IllegalTestMetricsReporter(Properties props, MetricRegistry registry) {}
+  }
 }
+
+