feat(metrics): emit hoodie.table.version as gauge on client init

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieClient.java

@@ -196,14 +196,19 @@ public HoodieEngineContext getEngineContext() {
   }
 
   protected HoodieTableMetaClient createMetaClient(boolean loadActiveTimelineOnLoad) {
-    return HoodieTableMetaClient.builder()
+    HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
         .setConf(storageConf.newInstance())
         .setBasePath(config.getBasePath())
         .setLoadActiveTimelineOnLoad(loadActiveTimelineOnLoad)
         .setConsistencyGuardConfig(config.getConsistencyGuardConfig())
         .setTimeGeneratorConfig(config.getTimeGeneratorConfig())
         .setFileSystemRetryConfig(config.getFileSystemRetryConfig())
         .setMetaserverConfig(config.getProps()).build();
+    if (metrics != null && config.isMetricsOn()) {
+      metrics.emitTableVersionMetric(
+          metaClient.getTableConfig().getTableVersion().versionCode());
+    }
+    return metaClient;
   }
 
   /**

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

@@ -89,6 +89,7 @@ public class HoodieMetrics {
   public static final String POST_COMMIT_STR = "postCommit";
   public static final String SUCCESS_EXTENSION = ".success";
   public static final String FAILURE_EXTENSION = ".failure";
+  public static final String TABLE_VERSION_STR = "tableVersion";
 
   public static final String TIMER_METRIC = "timer";
   public static final String COUNTER_METRIC = "counter";
@@ -272,6 +273,21 @@ public Timer.Context getConflictResolutionCtx() {
     return conflictResolutionTimer == null ? null : conflictResolutionTimer.time();
   }
 
+  /**
+   * Emits the current Hudi table version as a gauge metric. Intended to be called once at
+   * client init time so observability backends can alert when a table version drifts (for
+   * example when a deployment that intends to stay pinned at an older table version is
+   * accidentally promoted by an auto-upgrade path).
+   *
+   * @param versionCode integer version code from {@code HoodieTableVersion.versionCode()}
+   */
+  public void emitTableVersionMetric(int versionCode) {
+    if (!config.isMetricsOn()) {
+      return;
+    }
+    metrics.registerGauge(getMetricsName("table", TABLE_VERSION_STR), versionCode);
+  }
+
   public void updateMetricsForEmptyData(String actionType) {
     if (!config.isMetricsOn() || !config.getMetricsReporterType().equals(MetricsReporterType.PROMETHEUS_PUSHGATEWAY)) {
       // No-op if metrics are not of type PROMETHEUS_PUSHGATEWAY.

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

@@ -705,4 +705,24 @@ public void testConflictResolutionByCategoryMetrics() {
         HoodieWriteConflictException.ConflictCategory.TABLE_SERVICE_VS_TABLE_SERVICE);
     assertEquals(1, metrics.getRegistry().getCounters().get(tableServiceVsTableService).getCount());
   }
+
+  @Test
+  public void testEmitTableVersionMetric() {
+    int versionCode = 6;
+    hoodieMetrics.emitTableVersionMetric(versionCode);
+
+    String metricName = hoodieMetrics.getMetricsName("table", HoodieMetrics.TABLE_VERSION_STR);
+    assertEquals((long) versionCode, metrics.getRegistry().getGauges().get(metricName).getValue());
+  }
+
+  @Test
+  public void testEmitTableVersionMetricWhenMetricsDisabled() {
+    HoodieMetrics disabledMetrics = buildMetricsOff();
+    disabledMetrics.emitTableVersionMetric(6);
+
+    String metricName = disabledMetrics.getMetricsName("table", HoodieMetrics.TABLE_VERSION_STR);
+    if (disabledMetrics.getMetrics() != null) {
+      assertNull(disabledMetrics.getMetrics().getRegistry().getGauges().get(metricName));
+    }
+  }
 }