[spark] Expose Paimon metrics through JMX

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala

@@ -32,6 +32,7 @@ import org.apache.paimon.io.{CompactIncrement, DataIncrement}
 import org.apache.paimon.manifest.FileKind
 import org.apache.paimon.spark.{SparkPostponeStagedCommitter, SparkRow}
 import org.apache.paimon.spark.catalog.functions.BucketFunction
+import org.apache.paimon.spark.metric.SparkMetricRegistry
 import org.apache.paimon.spark.schema.SparkSystemColumns.{BUCKET_COL, ROW_KIND_COL}
 import org.apache.paimon.spark.sort.TableSorter
 import org.apache.paimon.spark.util.OptionUtils.paimonExtensionEnabled
@@ -86,6 +87,8 @@ case class PaimonSparkWriter(
     }
   }
 
+  @transient private lazy val metricRegistry = SparkMetricRegistry()
+
   val postponeBatchWriteFixedBucket: Boolean =
     table.bucketMode() == POSTPONE_MODE && coreOptions.postponeBatchWriteFixedBucket()
 
@@ -471,6 +474,7 @@ case class PaimonSparkWriter(
     val activeWriteBuilder =
       Option(directPostponeWriteBuilder).getOrElse(writeBuilder)
     val tableCommit = activeWriteBuilder.newCommit()
+    tableCommit.withMetricRegistry(metricRegistry)
     if (operation != null) {
       tableCommit.withOperation(operation)
     }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricGroup.scala

@@ -0,0 +1,71 @@
+/*
+ * 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.paimon.spark.metric
+
+import org.apache.paimon.metrics.{Counter, Gauge, Histogram, MetricGroupImpl}
+
+import com.codahale.metrics.{Gauge => CodahaleGauge, MetricRegistry => CodahaleMetricRegistry}
+
+import java.util.{Map => JMap}
+
+import scala.collection.JavaConverters._
+
+/** Keeps Paimon's Spark UI metrics while publishing their current values to JMX. */
+class SparkMetricGroup(
+    groupName: String,
+    variables: JMap[String, String],
+    registry: CodahaleMetricRegistry)
+  extends MetricGroupImpl(groupName, variables) {
+
+  override def counter(name: String): Counter = {
+    val metric = super.counter(name)
+    register(name, () => java.lang.Long.valueOf(metric.getCount))
+    metric
+  }
+
+  override def gauge[T](name: String, gauge: Gauge[T]): Gauge[T] = {
+    val metric = super.gauge(name, gauge)
+    if (metric != null) {
+      register(name, () => metric.getValue.asInstanceOf[AnyRef])
+    }
+    metric
+  }
+
+  override def histogram(name: String, windowSize: Int): Histogram = {
+    val metric = super.histogram(name, windowSize)
+    register(name, () => java.lang.Double.valueOf(metric.getStatistics.getMean))
+    metric
+  }
+
+  private def register(name: String, value: () => AnyRef): Unit = {
+    val path = (Seq(groupName) ++ variables.asScala.toSeq.sortBy(_._1).flatMap {
+      case (key, variable) => Seq(key, variable)
+    } :+ name).mkString(".")
+    val jmxGauge = new CodahaleGauge[AnyRef] {
+      override def getValue: AnyRef = value()
+    }
+
+    // Keep the latest value available to periodic scrapers; repeated operations on the same
+    // table replace their metrics rather than increasing the registry's cardinality.
+    registry.synchronized {
+      registry.remove(path)
+      registry.register(path, jmxGauge)
+    }
+  }
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/metric/SparkMetricRegistry.scala

@@ -18,10 +18,11 @@
 
 package org.apache.paimon.spark.metric
 
-import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricGroupImpl, MetricRegistry}
+import org.apache.paimon.metrics.{Gauge, Metric, MetricGroup, MetricRegistry}
 import org.apache.paimon.operation.metrics.{CommitMetrics, ScanMetrics, WriterBufferMetric}
 import org.apache.paimon.spark._
 
+import org.apache.spark.metrics.source.PaimonMetricsSource
 import org.apache.spark.sql.connector.metric.CustomTaskMetric
 
 import java.util.{Map => JMap}
@@ -35,7 +36,7 @@ case class SparkMetricRegistry() extends MetricRegistry {
   override def createMetricGroup(
       groupName: String,
       variables: JMap[String, String]): MetricGroup = {
-    val metricGroup = new MetricGroupImpl(groupName, variables)
+    val metricGroup = new SparkMetricGroup(groupName, variables, PaimonMetricsSource.metricRegistry)
     metricGroups.put(groupName, metricGroup)
     metricGroup
   }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/metrics/source/PaimonMetricsSource.scala

@@ -0,0 +1,64 @@
+/*
+ * 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.spark.metrics.source
+
+import com.codahale.metrics.{MetricRegistry => CodahaleMetricRegistry}
+import com.codahale.metrics.jmx.JmxReporter
+import org.apache.spark.SparkEnv
+import org.apache.spark.metrics.MetricsSystem
+
+/** Spark source for Paimon metrics. The reporter observes metrics added after source creation. */
+class PaimonMetricsSource extends Source {
+
+  override val sourceName: String = "paimon"
+
+  override def metricRegistry: CodahaleMetricRegistry = PaimonMetricsSource.sharedRegistry
+}
+
+object PaimonMetricsSource {
+
+  private val sharedRegistry = new CodahaleMetricRegistry()
+
+  private val reporter = JmxReporter.forRegistry(sharedRegistry).inDomain("paimon").build()
+  reporter.start()
+
+  // Spark snapshots a source's registry when it is registered, so the reporter above also
+  // observes metrics created later by scans and commits.
+  private val source = new PaimonMetricsSource
+
+  @volatile private var registeredSystem: MetricsSystem = _
+
+  def metricRegistry: CodahaleMetricRegistry = {
+    Option(SparkEnv.get).foreach {
+      env =>
+        val system = env.metricsSystem
+        if (registeredSystem ne system) {
+          synchronized {
+            if (registeredSystem ne system) {
+              if (system.getSourcesByName(source.sourceName).isEmpty) {
+                system.registerSource(source)
+              }
+              registeredSystem = system
+            }
+          }
+        }
+    }
+    sharedRegistry
+  }
+}

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonMetricTest.scala

@@ -20,10 +20,12 @@ package org.apache.paimon.spark.sql
 
 import org.apache.paimon.spark.PaimonMetrics.{RESULTED_TABLE_FILES, SCANNED_SNAPSHOT_ID, SKIPPED_TABLE_FILES}
 import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.metric.SparkMetricRegistry
 import org.apache.paimon.spark.read.PaimonSplitScan
 import org.apache.paimon.spark.util.ScanPlanHelper
 import org.apache.paimon.table.source.DataSplit
 
+import org.apache.spark.metrics.source.PaimonMetricsSource
 import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.PaimonUtils.createDataset
@@ -34,8 +36,76 @@ import org.apache.spark.sql.execution.metric.SQLMetric
 import org.apache.spark.sql.paimon.Utils
 import org.junit.jupiter.api.Assertions
 
+import javax.management.ObjectName
+
+import java.lang.management.ManagementFactory
+
 class PaimonMetricTest extends PaimonSparkTestBase with ScanPlanHelper {
 
+  test("Paimon Metric: registered metrics are exposed through JMX") {
+    val tableName = "jmx-metric-test-" + System.nanoTime()
+    val registry = SparkMetricRegistry()
+    val group = registry.createTableMetricGroup("commit", tableName)
+    group.gauge("lastCommitDuration", () => 42L)
+    val counter = group.counter("recordsWritten")
+    counter.inc(3)
+    val histogram = group.histogram("commitDuration", 10)
+    histogram.update(7)
+
+    val server = ManagementFactory.getPlatformMBeanServer
+    val names = server.queryNames(new ObjectName("paimon:*"), null)
+    val metricName = names.toArray.collectFirst {
+      case name: ObjectName
+          if name.toString.contains(tableName) &&
+            name.toString.contains("lastCommitDuration") =>
+        name
+    }
+    assert(metricName.isDefined, s"JMX metric missing for $tableName: $names")
+    assert(server.getAttribute(metricName.get, "Value") == 42L)
+    assert(group.getMetrics.get("lastCommitDuration") != null)
+
+    val counterName = names.toArray.collectFirst {
+      case name: ObjectName
+          if name.toString.contains(tableName) &&
+            name.toString.contains("recordsWritten") =>
+        name
+    }.get
+    assert(server.getAttribute(counterName, "Value") == 3L)
+    val histogramName = names.toArray.collectFirst {
+      case name: ObjectName
+          if name.toString.contains(tableName) &&
+            name.toString.contains("commitDuration") =>
+        name
+    }.get
+    assert(server.getAttribute(histogramName, "Value") == 7.0)
+
+    val metricCount = PaimonMetricsSource.metricRegistry.getMetrics.size()
+    val nextGroup = registry.createTableMetricGroup("commit", tableName)
+    nextGroup.gauge("lastCommitDuration", () => 99L)
+    assert(server.getAttribute(metricName.get, "Value") == 99L)
+    assert(PaimonMetricsSource.metricRegistry.getMetrics.size() == metricCount)
+  }
+
+  test("Paimon Metric: V1 commit metrics are exposed through JMX") {
+    withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+      withTable("T_V1_JMX") {
+        sql("CREATE TABLE T_V1_JMX (id INT)")
+        sql("INSERT INTO T_V1_JMX VALUES (1), (2)")
+
+        val server = ManagementFactory.getPlatformMBeanServer
+        val names = server.queryNames(new ObjectName("paimon:*"), null)
+        val commitMetric = names.toArray.collectFirst {
+          case name: ObjectName
+              if name.toString.toLowerCase.contains("t_v1_jmx") &&
+                name.toString.contains("lastTableFilesAdded") =>
+            name
+        }
+        assert(commitMetric.isDefined, s"V1 commit JMX metric missing: $names")
+        assert(server.getAttribute(commitMetric.get, "Value").asInstanceOf[Long] > 0L)
+      }
+    }
+  }
+
   test(s"Paimon Metric: scan driver metric") {
     // Spark support reportDriverMetrics since Spark 3.4
     if (gteqSpark3_4) {