From e7c666d317e512151640dab28737ce72ca506a65 Mon Sep 17 00:00:00 2001 From: ConradJam Date: Sun, 16 Aug 2026 14:14:12 +0800 Subject: [PATCH 1/2] [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 --- .../optimizing/OptimizerGroupMetrics.java | 10 ++-- .../optimizing/TestSchedulingPolicy.java | 53 +++++++++++++++++++ 2 files changed, 58 insertions(+), 5 deletions(-) create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java 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 4b640d4da6..afeea9b104 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 void register() { OPTIMIZER_GROUP_PLANING_TABLES, (Gauge) () -> - optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream() + optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream() .filter(t -> t.getOptimizingStatus().equals(PLANNING)) .count()); registerMetric( @@ -171,7 +171,7 @@ public void register() { OPTIMIZER_GROUP_PENDING_TABLES, (Gauge) () -> - optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream() + optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream() .filter(t -> t.getOptimizingStatus().equals(PENDING)) .count()); registerMetric( @@ -179,7 +179,7 @@ public void register() { OPTIMIZER_GROUP_EXECUTING_TABLES, (Gauge) () -> - optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream() + optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream() .filter(t -> t.getOptimizingStatus().isProcessing()) .count()); registerMetric( @@ -187,7 +187,7 @@ public void register() { OPTIMIZER_GROUP_IDLE_TABLES, (Gauge) () -> - optimizingQueue.getSchedulingPolicy().getTableRuntimeMap().values().stream() + optimizingQueue.getSchedulingPolicy().snapshotTableRuntimes().stream() .filter(t -> t.getOptimizingStatus().equals(IDLE)) .count()); registerMetric( @@ -195,7 +195,7 @@ public void register() { OPTIMIZER_GROUP_COMMITTING_TABLES, (Gauge) () -> - 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/TestSchedulingPolicy.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java new file mode 100644 index 0000000000..c079821372 --- /dev/null +++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java @@ -0,0 +1,53 @@ +/* + * 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.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.apache.amoro.ServerTableIdentifier; +import org.apache.amoro.TableFormat; +import org.apache.amoro.resource.ResourceGroup; +import org.apache.amoro.server.table.DefaultTableRuntime; +import org.apache.amoro.table.TableIdentifier; +import org.junit.Assert; +import org.junit.Test; + +import java.util.List; + +public class TestSchedulingPolicy { + + @Test + public void testTableRuntimesSnapshotIsIndependentFromLiveMap() { + SchedulingPolicy policy = + new SchedulingPolicy(new ResourceGroup.Builder("test", "local").build()); + DefaultTableRuntime runtime = mock(DefaultTableRuntime.class); + ServerTableIdentifier identifier = + ServerTableIdentifier.of( + TableIdentifier.of("catalog", "db", "table"), TableFormat.ICEBERG); + when(runtime.getTableIdentifier()).thenReturn(identifier); + policy.addTable(runtime); + + List snapshot = policy.snapshotTableRuntimes(); + policy.removeTable(runtime); + + Assert.assertEquals(1, snapshot.size()); + Assert.assertTrue(policy.snapshotTableRuntimes().isEmpty()); + } +} From 3e41beb9e073fc8df4e0860859a0dcaa22b29a54 Mon Sep 17 00:00:00 2001 From: ConradJam Date: Fri, 28 Aug 2026 15:21:21 +0800 Subject: [PATCH 2/2] [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. --- .../optimizing/TestOptimizerGroupMetrics.java | 94 +++++++++++++++++++ .../optimizing/TestSchedulingPolicy.java | 53 ----------- 2 files changed, 94 insertions(+), 53 deletions(-) create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizerGroupMetrics.java delete mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java 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 0000000000..08ae763c31 --- /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 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 tagValues = Collections.singletonMap(GROUP_TAG, GROUP_NAME); + Gauge gauge = (Gauge) registry.getMetrics().get(new MetricKey(define, tagValues)); + Assert.assertNotNull(gauge); + return gauge.getValue(); + } +} diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java deleted file mode 100644 index c079821372..0000000000 --- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestSchedulingPolicy.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * 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.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import org.apache.amoro.ServerTableIdentifier; -import org.apache.amoro.TableFormat; -import org.apache.amoro.resource.ResourceGroup; -import org.apache.amoro.server.table.DefaultTableRuntime; -import org.apache.amoro.table.TableIdentifier; -import org.junit.Assert; -import org.junit.Test; - -import java.util.List; - -public class TestSchedulingPolicy { - - @Test - public void testTableRuntimesSnapshotIsIndependentFromLiveMap() { - SchedulingPolicy policy = - new SchedulingPolicy(new ResourceGroup.Builder("test", "local").build()); - DefaultTableRuntime runtime = mock(DefaultTableRuntime.class); - ServerTableIdentifier identifier = - ServerTableIdentifier.of( - TableIdentifier.of("catalog", "db", "table"), TableFormat.ICEBERG); - when(runtime.getTableIdentifier()).thenReturn(identifier); - policy.addTable(runtime); - - List snapshot = policy.snapshotTableRuntimes(); - policy.removeTable(runtime); - - Assert.assertEquals(1, snapshot.size()); - Assert.assertTrue(policy.snapshotTableRuntimes().isEmpty()); - } -}