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; }