[HUDI-209] Implement JMX metrics reporter

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

@@ -45,6 +45,14 @@ public class HoodieMetricsConfig extends DefaultHoodieConfig {
   public static final String GRAPHITE_SERVER_PORT = GRAPHITE_PREFIX + ".port";
   public static final int DEFAULT_GRAPHITE_SERVER_PORT = 4756;
 
+  // Jmx
+  public static final String JMX_PREFIX = METRIC_PREFIX + ".jmx";
+  public static final String JMX_HOST = JMX_PREFIX + ".host";
+  public static final String DEFAULT_JMX_HOST = "localhost";
+
+  public static final String JMX_PORT = JMX_PREFIX + ".port";
+  public static final int DEFAULT_JMX_PORT = 9889;
+
   public static final String GRAPHITE_METRIC_PREFIX = GRAPHITE_PREFIX + ".metric.prefix";
 
   private HoodieMetricsConfig(Properties props) {

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

@@ -471,6 +471,14 @@ public String getGraphiteMetricPrefix() {
     return props.getProperty(HoodieMetricsConfig.GRAPHITE_METRIC_PREFIX);
   }
 
+  public String getJmxHost() {
+    return props.getProperty(HoodieMetricsConfig.JMX_HOST);
+  }
+
+  public int getJmxPort() {
+    return Integer.parseInt(props.getProperty(HoodieMetricsConfig.JMX_PORT));
+  }
+
   /**
    * memory configs
    */

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

@@ -0,0 +1,85 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import java.io.Closeable;
+
+import java.lang.management.ManagementFactory;
+import java.rmi.registry.LocateRegistry;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXConnectorServerFactory;
+import javax.management.remote.JMXServiceURL;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+/**
+ * Implementation of Jmx reporter, which used to report jmx metric.
+ */
+public class JmxMetricsReporter extends MetricsReporter {
+
+  private static Logger logger = LogManager.getLogger(JmxMetricsReporter.class);
+  private final JMXConnectorServer connector;
+  private String host;
+  private int port;
+
+  public JmxMetricsReporter(HoodieWriteConfig config) {
+    try {
+      // Check the host and port here
+      this.host = config.getJmxHost();
+      this.port = config.getJmxPort();
+      if (host == null || port == 0) {
+        throw new RuntimeException(
+            String.format("Jmx cannot be initialized with host[%s] and port[%s].",
+                host, port));
+      }
+      LocateRegistry.createRegistry(port);
+      String serviceUrl =
+          "service:jmx:rmi://" + host + ":" + port + "/jndi/rmi://" + host + ":" + port + "/jmxrmi";
+      JMXServiceURL url = new JMXServiceURL(serviceUrl);
+      this.connector = JMXConnectorServerFactory
+          .newJMXConnectorServer(url, null, ManagementFactory.getPlatformMBeanServer());
+    } catch (Exception e) {
+      String msg = "Jmx initialize failed: ";
+      logger.error(msg, e);
+      throw new HoodieException(msg, e);
+    }
+  }
+
+  @Override
+  public void start() {
+    try {
+      Preconditions.checkNotNull(connector, "Cannot start as the jmxReporter is null.");
+      connector.start();
+    } catch (Exception e) {
+      throw new HoodieException(e);
+    }
+  }
+
+  @Override
+  public void report() {
+  }
+
+  @Override
+  public Closeable getReporter() {
+    return null;
+  }
+}

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

@@ -40,6 +40,9 @@ public static MetricsReporter createReporter(HoodieWriteConfig config, MetricReg
       case INMEMORY:
         reporter = new InMemoryMetricsReporter();
         break;
+      case JMX:
+        reporter = new JmxMetricsReporter(config);
+        break;
       default:
         logger.error("Reporter type[" + type + "] is not supported.");
         break;

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

@@ -22,5 +22,5 @@
  * Types of the reporter. Right now we only support Graphite. We can include JMX and CSV in the future.
  */
 public enum MetricsReporterType {
-  GRAPHITE, INMEMORY
+  GRAPHITE, INMEMORY, JMX
 }

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

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import static org.apache.hudi.metrics.Metrics.registerGauge;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.hudi.config.HoodieMetricsConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.junit.Test;
+
+/**
+ * Test for the Jmx metrics report.
+ */
+public class TestHoodieJmxMetrics extends TestHoodieMetrics {
+
+  @Override
+  public void start() {
+    HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+    when(config.isMetricsOn()).thenReturn(true);
+    when(config.getMetricsReporterType()).thenReturn(MetricsReporterType.JMX);
+    when(config.getJmxHost()).thenReturn(HoodieMetricsConfig.DEFAULT_JMX_HOST);
+    when(config.getJmxPort()).thenReturn(HoodieMetricsConfig.DEFAULT_JMX_PORT);
+    new HoodieMetrics(config, "raw_table");
+  }
+
+  @Test
+  public void testRegisterGauge() {
+    registerGauge("jmx_metric", 123L);
+    assertTrue(Metrics.getInstance().getRegistry().getGauges()
+        .get("jmx_metric").getValue().toString().equals("123"));
+  }
+}

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

@@ -29,14 +29,12 @@
 
 public class TestHoodieMetrics {
 
-  private HoodieMetrics metrics = null;
-
   @Before
   public void start() {
     HoodieWriteConfig config = mock(HoodieWriteConfig.class);
     when(config.isMetricsOn()).thenReturn(true);
     when(config.getMetricsReporterType()).thenReturn(MetricsReporterType.INMEMORY);
-    metrics = new HoodieMetrics(config, "raw_table");
+    new HoodieMetrics(config, "raw_table");
   }
 
   @Test

pom.xml

@@ -83,7 +83,7 @@
     <hadoop.version>2.7.3</hadoop.version>
     <hive.groupid>org.apache.hive</hive.groupid>
     <hive.version>2.3.1</hive.version>
-    <metrics.version>4.0.2</metrics.version>
+    <metrics.version>4.1.1</metrics.version>
     <spark.version>2.1.0</spark.version>
     <avro.version>1.7.7</avro.version>
     <scala.version>2.11.8</scala.version>