This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix/jvm-thread-metrics-npe in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0522ab578dd8f116c9e6ac183f3d689b135d34d7 Author: Tian Jiang <[email protected]> AuthorDate: Wed Oct 21 10:47:31 2026 +0800 Fix concurrent JVM thread state metric refresh --- .../metrics/metricsets/jvm/JvmThreadMetrics.java | 3 +- .../metricsets/jvm/JvmThreadMetricsTest.java | 143 +++++++++++++++++++++ 2 files changed, 145 insertions(+), 1 deletion(-) diff --git a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java index cd8fcbd6cd0..5109d7f6f43 100644 --- a/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java +++ b/iotdb-core/metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetrics.java @@ -104,7 +104,8 @@ public class JvmThreadMetrics implements IMetricSet { } // VisibleForTesting - static long getThreadStateCount(ThreadMXBean threadBean, Thread.State state) { + static synchronized long getThreadStateCount(ThreadMXBean threadBean, Thread.State state) { + // Keep reads and refreshes under the same lock to avoid observing a partially rebuilt EnumMap. checkAndUpdate(threadBean); return threadStateCountMap.getOrDefault(state, 0); } diff --git a/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetricsTest.java b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetricsTest.java new file mode 100644 index 00000000000..4c7a4f9b421 --- /dev/null +++ b/iotdb-core/metrics/interface/src/test/java/org/apache/iotdb/metrics/metricsets/jvm/JvmThreadMetricsTest.java @@ -0,0 +1,143 @@ +/* + * 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.iotdb.metrics.metricsets.jvm; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.lang.reflect.Field; +import java.lang.reflect.Proxy; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; +import java.util.function.Supplier; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class JvmThreadMetricsTest { + + @Before + @After + public void expireCache() throws Exception { + Field lastUpdateTime = JvmThreadMetrics.class.getDeclaredField("lastUpdateTime"); + lastUpdateTime.setAccessible(true); + synchronized (JvmThreadMetrics.class) { + lastUpdateTime.setLong(null, 0L); + } + } + + // Count two live threads, ignore a terminated thread's null info, and return zero for an absent + // state without resampling within the cache interval. + @Test + public void testThreadStateCountsAndMissingState() { + ThreadInfo info = + ManagementFactory.getThreadMXBean().getThreadInfo(Thread.currentThread().getId()); + AtomicInteger samples = new AtomicInteger(); + ThreadMXBean threadBean = + threadBean( + () -> { + samples.incrementAndGet(); + return new ThreadInfo[] {info, info, null}; + }); + + assertEquals(2L, JvmThreadMetrics.getThreadStateCount(threadBean, info.getThreadState())); + assertEquals(0L, JvmThreadMetrics.getThreadStateCount(threadBean, Thread.State.TERMINATED)); + assertEquals(1, samples.get()); + } + + // Pause a refresh after clearing the cache, then read the same state concurrently. Both calls + // must return the complete sample rather than a temporarily absent entry. + @Test + public void testConcurrentReadDuringRefresh() throws Exception { + ThreadInfo info = + ManagementFactory.getThreadMXBean().getThreadInfo(Thread.currentThread().getId()); + CountDownLatch refreshStarted = new CountDownLatch(1); + CountDownLatch resumeRefresh = new CountDownLatch(1); + CountDownLatch readStarted = new CountDownLatch(1); + AtomicReference<Thread> readerThread = new AtomicReference<>(); + ThreadMXBean threadBean = + threadBean( + () -> { + refreshStarted.countDown(); + try { + assertTrue(resumeRefresh.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + return new ThreadInfo[] {info}; + }); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<Long> refresh = + executor.submit( + () -> JvmThreadMetrics.getThreadStateCount(threadBean, info.getThreadState())); + assertTrue(refreshStarted.await(5, TimeUnit.SECONDS)); + Future<Long> read = + executor.submit( + () -> { + readerThread.set(Thread.currentThread()); + readStarted.countDown(); + return JvmThreadMetrics.getThreadStateCount(threadBean, info.getThreadState()); + }); + assertTrue(readStarted.await(5, TimeUnit.SECONDS)); + + // Ensure the reader reaches the cache lookup before allowing the refresh to finish. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (!read.isDone() + && readerThread.get().getState() != Thread.State.BLOCKED + && System.nanoTime() < deadline) { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1)); + } + assertTrue(read.isDone() || readerThread.get().getState() == Thread.State.BLOCKED); + resumeRefresh.countDown(); + + assertEquals(1L, refresh.get(5, TimeUnit.SECONDS).longValue()); + assertEquals(1L, read.get(5, TimeUnit.SECONDS).longValue()); + } finally { + resumeRefresh.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + private static ThreadMXBean threadBean(Supplier<ThreadInfo[]> threadInfos) { + return (ThreadMXBean) + Proxy.newProxyInstance( + JvmThreadMetricsTest.class.getClassLoader(), + new Class<?>[] {ThreadMXBean.class}, + (proxy, method, args) -> + switch (method.getName()) { + case "getAllThreadIds" -> new long[] {1L, 2L, 3L}; + case "getThreadInfo" -> threadInfos.get(); + default -> throw new UnsupportedOperationException(method.getName()); + }); + } +}
