From 8d7beaed4726cefad2e546c7e319ba5a2a16be14 Mon Sep 17 00:00:00 2001 From: pallavithegod Date: Wed, 14 Oct 2026 21:38:54 +0530 Subject: [PATCH] Fix compactor thread and metrics leak in RocksDBKVSpace Shut down per-space compactionExecutor in doClose and doDestroy, and remove its ExecutorServiceMetrics meters. Previously each range leaked a non-daemon thread, blocking JVM exit and growing without bound on split/merge. Fixes #290 --- .../localengine/rocksdb/RocksDBKVSpace.java | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpace.java b/base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpace.java index adae463e9..637f294e2 100644 --- a/base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpace.java +++ b/base-kv/base-kv-local-engine-rocksdb/src/main/java/org/apache/bifromq/basekv/localengine/rocksdb/RocksDBKVSpace.java @@ -49,6 +49,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; import lombok.SneakyThrows; import org.apache.bifromq.baseenv.EnvProvider; import org.apache.bifromq.basekv.localengine.AbstractKVSpace; @@ -69,6 +70,7 @@ abstract class RocksDBKVSpace extends AbstractKVSpace protected final IWriteStatsRecorder writeStats; private final File keySpaceDBDir; private final ExecutorService compactionExecutor; + private final Tags compactionMetricTags; private final AtomicBoolean compacting; private final ISyncContext.IRefresher metadataRefresher; private SpaceMetrics spaceMetrics; @@ -89,10 +91,11 @@ public RocksDBKVSpace(String id, syncContext = new SyncContext(); metadataRefresher = syncContext.refresher(); compacting = new AtomicBoolean(false); + compactionMetricTags = Tags.of(tags); compactionExecutor = ExecutorServiceMetrics.monitor(Metrics.globalRegistry, new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), EnvProvider.INSTANCE.newThreadFactory("kvspace-compactor-" + id)), - "compactor", "kvspace", Tags.of(tags)); + "compactor", "kvspace", compactionMetricTags); if (boolVal(conf, MANUAL_COMPACTION)) { int minKeys = (int) numVal(conf, COMPACT_MIN_TOMBSTONE_KEYS); int minRanges = (int) numVal(conf, COMPACT_MIN_TOMBSTONE_RANGES); @@ -130,6 +133,8 @@ protected void publishMetadata(Map metadataUpdates) { @Override protected void doClose() { logger.debug("Close key range[{}]", id); + compactionExecutor.shutdownNow(); + unregisterCompactionMetrics(); if (spaceMetrics != null) { spaceMetrics.close(); } @@ -137,6 +142,8 @@ protected void doClose() { @Override protected void doDestroy() { + compactionExecutor.shutdownNow(); + unregisterCompactionMetrics(); // Destroy the whole space root directory, including pointer file and all generations. try { if (keySpaceDBDir.exists()) { @@ -147,6 +154,15 @@ protected void doDestroy() { } } + private void unregisterCompactionMetrics() { + Metrics.globalRegistry.getMeters().stream() + .filter(m -> m.getId().getName().startsWith("kvspace.executor")) + .filter(m -> compactionMetricTags.stream() + .allMatch(t -> t.getValue().equals(m.getId().getTag(t.getKey())))) + .collect(Collectors.toList()) + .forEach(Metrics.globalRegistry::remove); + } + protected File spaceRootDir() { return keySpaceDBDir; }