This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new c4c0aef95 [hotfix][Metrics] Iterate a locked table-runtime snapshot in
gauges (#4326)
c4c0aef95 is described below
commit c4c0aef95e09c28768a491515b98b79b9593d998
Author: ConradJam <[email protected]>
AuthorDate: Thu Sep 10 11:21:58 2026 +0800
[hotfix][Metrics] Iterate a locked table-runtime snapshot in gauges (#4326)
* [hotfix][Metrics] Iterate a locked snapshot of tracked tables in group
gauges
OptimizerGroupMetrics' five table-count gauges streamed over
SchedulingPolicy's live HashMap on scrape threads without tableLock;
concurrent addTable/removeTable (table onboarding, group migration)
threw ConcurrentModificationException mid-scrape. getTableRuntimeMap()
is @VisibleForTesting yet was the production access path.
Add tableRuntimesSnapshot() (locked copy, same pattern as
tableIdentifiersSnapshot) and point the gauges at it. Churn regression
test: with the lock temporarily removed it fails 2 of 3 runs with CME;
locked it is stable.
Fix record: docs/fix-records/2026-08-16-fix-17-gauge-snapshot-cme.md
* [hotfix][Metrics] Verify table gauges use runtime snapshots
Exercise the registered optimizer-group gauges directly and fail
deterministically if they read SchedulingPolicy's live runtime map. This
replaces the snapshot-only test that did not cover the production metric path.
---
.../server/optimizing/OptimizerGroupMetrics.java | 10 +--
.../optimizing/TestOptimizerGroupMetrics.java | 94 ++++++++++++++++++++++
2 files changed, 99 insertions(+), 5 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
index 4b640d4da..afeea9b10 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/OptimizerGroupMetrics.java
@@ -163,7 +163,7 @@ public class OptimizerGroupMetrics {
OPTIMIZER_GROUP_PLANING_TABLES,
(Gauge<Long>)
() ->
-
optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
+
optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream()
.filter(t -> t.getOptimizingStatus().equals(PLANNING))
.count());
registerMetric(
@@ -171,7 +171,7 @@ public class OptimizerGroupMetrics {
OPTIMIZER_GROUP_PENDING_TABLES,
(Gauge<Long>)
() ->
-
optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
+
optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream()
.filter(t -> t.getOptimizingStatus().equals(PENDING))
.count());
registerMetric(
@@ -179,7 +179,7 @@ public class OptimizerGroupMetrics {
OPTIMIZER_GROUP_EXECUTING_TABLES,
(Gauge<Long>)
() ->
-
optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
+
optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream()
.filter(t -> t.getOptimizingStatus().isProcessing())
.count());
registerMetric(
@@ -187,7 +187,7 @@ public class OptimizerGroupMetrics {
OPTIMIZER_GROUP_IDLE_TABLES,
(Gauge<Long>)
() ->
-
optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
+
optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream()
.filter(t -> t.getOptimizingStatus().equals(IDLE))
.count());
registerMetric(
@@ -195,7 +195,7 @@ public class OptimizerGroupMetrics {
OPTIMIZER_GROUP_COMMITTING_TABLES,
(Gauge<Long>)
() ->
-
optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream()
+
optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream()
.filter(t -> t.getOptimizingStatus().equals(COMMITTING))
.count());
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizerGroupMetrics.java
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizerGroupMetrics.java
new file mode 100644
index 000000000..08ae763c3
--- /dev/null
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizerGroupMetrics.java
@@ -0,0 +1,94 @@
+/*
+ * 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.amoro.server.optimizing;
+
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.GROUP_TAG;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_COMMITTING_TABLES;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_EXECUTING_TABLES;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_IDLE_TABLES;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_PENDING_TABLES;
+import static
org.apache.amoro.server.optimizing.OptimizerGroupMetrics.OPTIMIZER_GROUP_PLANING_TABLES;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.metrics.Gauge;
+import org.apache.amoro.metrics.MetricDefine;
+import org.apache.amoro.metrics.MetricKey;
+import org.apache.amoro.metrics.MetricRegistry;
+import org.apache.amoro.server.table.DefaultTableRuntime;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.ConcurrentModificationException;
+import java.util.List;
+import java.util.Map;
+
+public class TestOptimizerGroupMetrics {
+
+ private static final String GROUP_NAME = "test-group";
+
+ @Test
+ public void testTableGaugesUseRuntimeSnapshot() {
+ List<DefaultTableRuntime> tableRuntimes =
+ Arrays.asList(
+ tableRuntime(OptimizingStatus.PLANNING),
+ tableRuntime(OptimizingStatus.PENDING),
+ tableRuntime(OptimizingStatus.MINOR_OPTIMIZING),
+ tableRuntime(OptimizingStatus.IDLE),
+ tableRuntime(OptimizingStatus.COMMITTING));
+
+ SchedulingPolicy schedulingPolicy = mock(SchedulingPolicy.class);
+ when(schedulingPolicy.getTableRuntimeMap())
+ .thenThrow(new ConcurrentModificationException("live map must not be
iterated"));
+ when(schedulingPolicy.snapshotTableRuntimes()).thenReturn(tableRuntimes);
+
+ OptimizingQueue optimizingQueue = mock(OptimizingQueue.class);
+ when(optimizingQueue.getSchedulingPolicy()).thenReturn(schedulingPolicy);
+
+ MetricRegistry registry = new MetricRegistry();
+ OptimizerGroupMetrics metrics =
+ new OptimizerGroupMetrics(GROUP_NAME, registry, optimizingQueue);
+ metrics.register();
+ try {
+ Assert.assertEquals(1L, gaugeValue(registry,
OPTIMIZER_GROUP_PLANING_TABLES));
+ Assert.assertEquals(1L, gaugeValue(registry,
OPTIMIZER_GROUP_PENDING_TABLES));
+ Assert.assertEquals(2L, gaugeValue(registry,
OPTIMIZER_GROUP_EXECUTING_TABLES));
+ Assert.assertEquals(1L, gaugeValue(registry,
OPTIMIZER_GROUP_IDLE_TABLES));
+ Assert.assertEquals(1L, gaugeValue(registry,
OPTIMIZER_GROUP_COMMITTING_TABLES));
+ } finally {
+ metrics.unregister();
+ }
+ }
+
+ private DefaultTableRuntime tableRuntime(OptimizingStatus status) {
+ DefaultTableRuntime tableRuntime = mock(DefaultTableRuntime.class);
+ when(tableRuntime.getOptimizingStatus()).thenReturn(status);
+ return tableRuntime;
+ }
+
+ @SuppressWarnings("unchecked")
+ private long gaugeValue(MetricRegistry registry, MetricDefine define) {
+ Map<String, String> tagValues = Collections.singletonMap(GROUP_TAG,
GROUP_NAME);
+ Gauge<Long> gauge = (Gauge<Long>) registry.getMetrics().get(new
MetricKey(define, tagValues));
+ Assert.assertNotNull(gauge);
+ return gauge.getValue();
+ }
+}
