diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakDetector.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakDetector.java
new file mode 100644
index 000000000000..67f5c2f2bbc5
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakDetector.java
@@ -0,0 +1,101 @@
+/*
+ * 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.hadoop.hdds.utils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.ref.ReferenceQueue;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Simple general resource leak detector using {@link ReferenceQueue} and {@link java.lang.ref.WeakReference} to
+ * observe resource object life-cycle and assert proper resource closure before they are GCed.
+ *
+ * <p>
+ * Example usage:
+ *
+ * <pre> {@code
+ * class MyResource implements AutoClosable {
+ *   static final LeakDetector LEAK_DETECTOR = new LeakDetector("MyResource");
+ *
+ *   private final LeakTracker leakTracker = LEAK_DETECTOR.track(this, () -> {
+ *      // report leaks, don't refer to the original object (MyResource) here.
+ *      System.out.println("MyResource is not closed before being discarded.");
+ *   });
+ *
+ *   @Override
+ *   public void close() {
+ *     // proper resources cleanup...
+ *     // inform tracker that this object is closed properly.
+ *     leakTracker.close();
+ *   }
+ * }
+ *
+ * }</pre>
+ */
+public class LeakDetector {
+  public static final Logger LOG = LoggerFactory.getLogger(LeakDetector.class);
+  private final ReferenceQueue<Object> queue = new ReferenceQueue<>();
+  private final Set<LeakTracker> allLeaks = Collections.newSetFromMap(new ConcurrentHashMap<>());
+  private final String name;
+
+  public LeakDetector(String name) {
+    this.name = name;
+    start();
+  }
+
+  private void start() {
+    Thread t = new Thread(this::run);
+    t.setName(LeakDetector.class.getSimpleName() + "-" + name);
+    t.setDaemon(true);
+    LOG.info("Starting leak detector thread {}.", name);
+    t.start();
+  }
+
+  private void run() {
+    while (true) {
+      try {
+        LeakTracker tracker = (LeakTracker) queue.remove();
+        // Original resource already been GCed, if tracker is not closed yet,
+        // report a leak.
+        if (allLeaks.remove(tracker)) {
+          tracker.reportLeak();
+        }
+      } catch (InterruptedException e) {
+        LOG.warn("Thread interrupted, exiting.", e);
+        break;
+      }
+    }
+
+    LOG.warn("Exiting leak detector {}.", name);
+  }
+
+  public LeakTracker track(Object leakable, Runnable reportLeak) {
+    // A rate filter can be put here to only track a subset of all objects, e.g. 5%, 10%,
+    // if we have proofs that leak tracking impacts performance, or a single LeakDetector
+    // thread can't keep up with the pace of object allocation.
+    // For now, it looks effective enough and let keep it simple.
+    LeakTracker tracker = new LeakTracker(leakable, queue, allLeaks, reportLeak);
+    allLeaks.add(tracker);
+    return tracker;
+  }
+}
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakTracker.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakTracker.java
new file mode 100644
index 000000000000..6103d520ca8a
--- /dev/null
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/LeakTracker.java
@@ -0,0 +1,50 @@
+/*
+ * 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.hadoop.hdds.utils;
+
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
+import java.util.Set;
+
+/**
+ * A token to track resource closure.
+ *
+ * @see LeakDetector
+ */
+public class LeakTracker extends WeakReference<Object> {
+  private final Set<LeakTracker> allLeaks;
+  private final Runnable leakReporter;
+  LeakTracker(Object referent, ReferenceQueue<Object> referenceQueue,
+      Set<LeakTracker> allLeaks, Runnable leakReporter) {
+    super(referent, referenceQueue);
+    this.allLeaks = allLeaks;
+    this.leakReporter = leakReporter;
+  }
+
+  /**
+   * Called by the tracked resource when closing.
+   */
+  public void close() {
+    allLeaks.remove(this);
+  }
+
+  void reportLeak() {
+    leakReporter.run();
+  }
+}
diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/resource/TestLeakDetector.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/resource/TestLeakDetector.java
new file mode 100644
index 000000000000..4a60fcc8a4d5
--- /dev/null
+++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/resource/TestLeakDetector.java
@@ -0,0 +1,69 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hadoop.hdds.resource;
+
+import org.apache.hadoop.hdds.utils.LeakDetector;
+import org.apache.hadoop.hdds.utils.LeakTracker;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test LeakDetector.
+ */
+public class TestLeakDetector {
+  private static final LeakDetector LEAK_DETECTOR = new LeakDetector("test");
+  private AtomicInteger leaks = new AtomicInteger(0);
+
+  @Test
+  public void testLeakDetector() throws Exception {
+    // create and close resource => no leaks.
+    createResource(true);
+    System.gc();
+    Thread.sleep(100);
+    assertEquals(0, leaks.get());
+
+    // create and not close => leaks.
+    createResource(false);
+    System.gc();
+    Thread.sleep(100);
+    assertEquals(1, leaks.get());
+  }
+
+  private void createResource(boolean close) throws Exception {
+    MyResource resource = new MyResource(leaks);
+    if (close) {
+      resource.close();
+    }
+  }
+
+  private static final class MyResource implements AutoCloseable {
+    private final LeakTracker leakTracker;
+
+    private MyResource(final AtomicInteger leaks) {
+      leakTracker = LEAK_DETECTOR.track(this, () -> leaks.incrementAndGet());
+    }
+
+    @Override
+    public void close() throws Exception {
+      leakTracker.close();
+    }
+  }
+}
diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestRDBStoreByteArrayIterator.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestRDBStoreByteArrayIterator.java
index d81771aceb34..525ff27c5450 100644
--- a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestRDBStoreByteArrayIterator.java
+++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/db/TestRDBStoreByteArrayIterator.java
@@ -33,7 +33,6 @@
 import java.util.NoSuchElementException;
 import java.util.function.Consumer;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -307,28 +306,4 @@ public void testNormalPrefixedIterator() throws IOException {
 
     iter.close();
   }
-
-  @Test
-  public void testGetStackTrace() {
-    ManagedRocksIterator iterator = mock(ManagedRocksIterator.class);
-    RocksIterator mock = mock(RocksIterator.class);
-    when(iterator.get()).thenReturn(mock);
-    when(mock.isOwningHandle()).thenReturn(true);
-    ManagedRocksObjectUtils.assertClosed(iterator);
-    verify(iterator, times(1)).getStackTrace();
-
-    iterator = new ManagedRocksIterator(rocksDBIteratorMock);
-
-    // construct the expected trace.
-    StackTraceElement[] traceElements = Thread.currentThread().getStackTrace();
-    StringBuilder sb = new StringBuilder();
-    // first 2 lines will differ.
-    for (int i = 2; i < traceElements.length; i++) {
-      sb.append(traceElements[i]);
-      sb.append("\n");
-    }
-    String expectedTrace = sb.toString();
-    String fromObjectInit = iterator.getStackTrace();
-    assertThat(fromObjectInit).contains(expectedTrace);
-  }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedBloomFilter.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedBloomFilter.java
index 9e7bad651f77..ffee7c1f5519 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedBloomFilter.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedBloomFilter.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.BloomFilter;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed BloomFilter.
  */
 public class ManagedBloomFilter extends BloomFilter {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedColumnFamilyOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedColumnFamilyOptions.java
index 39ee7b0ce7ac..7b1da6a16923 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedColumnFamilyOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedColumnFamilyOptions.java
@@ -18,31 +18,24 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.ColumnFamilyOptions;
 import org.rocksdb.TableFormatConfig;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed ColumnFamilyOptions.
  */
 public class ManagedColumnFamilyOptions extends ColumnFamilyOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
-
   /**
    * Indicate if this ColumnFamilyOptions is intentionally used across RockDB
    * instances.
    */
   private boolean reused = false;
+  private final LeakTracker leakTracker = track(this);
 
   public ManagedColumnFamilyOptions() {
-    super();
   }
 
   public ManagedColumnFamilyOptions(ColumnFamilyOptions columnFamilyOptions) {
@@ -85,9 +78,9 @@ public boolean isReused() {
   }
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 
   /**
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedCompactRangeOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedCompactRangeOptions.java
index 6f7da30cdba7..0e397ed0e9b6 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedCompactRangeOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedCompactRangeOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.CompactRangeOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed CompactRangeOptions.
  */
 public class ManagedCompactRangeOptions extends CompactRangeOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedDBOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedDBOptions.java
index a66a04eae90a..fa01e2e1018b 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedDBOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedDBOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.DBOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed DBOptions.
  */
 public class ManagedDBOptions extends DBOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedEnvOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedEnvOptions.java
index 38d6f95ce708..baad1ad7f4ca 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedEnvOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedEnvOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.EnvOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed EnvOptions.
  */
 public class ManagedEnvOptions extends EnvOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedFlushOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedFlushOptions.java
index 8801f7d241d0..126f5336ba90 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedFlushOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedFlushOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.FlushOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed FlushOptions.
  */
 public class ManagedFlushOptions extends FlushOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedIngestExternalFileOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedIngestExternalFileOptions.java
index 94b34e20d2ca..1783a34587cf 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedIngestExternalFileOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedIngestExternalFileOptions.java
@@ -18,26 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.IngestExternalFileOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed IngestExternalFileOptions.
  */
-public class ManagedIngestExternalFileOptions extends
-    IngestExternalFileOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+public class ManagedIngestExternalFileOptions extends IngestExternalFileOptions {
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedLRUCache.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedLRUCache.java
index 8bf9147c1571..5244863a5a17 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedLRUCache.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedLRUCache.java
@@ -18,29 +18,24 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.LRUCache;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed LRUCache.
  */
 public class ManagedLRUCache extends LRUCache {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   public ManagedLRUCache(long capacity) {
     super(capacity);
   }
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedObject.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedObject.java
index 0ab87f156ec9..1e4068a7a800 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedObject.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedObject.java
@@ -18,11 +18,10 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.RocksObject;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * General template for a managed RocksObject.
@@ -30,13 +29,10 @@
  */
 class ManagedObject<T extends RocksObject> implements AutoCloseable {
   private final T original;
-
-  @Nullable
-  private final StackTraceElement[] elements;
+  private final LeakTracker leakTracker = track(this);
 
   ManagedObject(T original) {
     this.original = original;
-    this.elements = ManagedRocksObjectUtils.getStackTrace();
   }
 
   public T get() {
@@ -46,16 +42,6 @@ public T get() {
   @Override
   public void close() {
     original.close();
+    leakTracker.close();
   }
-
-  @Override
-  protected void finalize() throws Throwable {
-    ManagedRocksObjectUtils.assertClosed(this);
-    super.finalize();
-  }
-
-  public String getStackTrace() {
-    return formatStackTrace(elements);
-  }
-
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedOptions.java
index 4ae96e1b7682..e438068e3a70 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.Options;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed Options.
  */
 public class ManagedOptions extends Options {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedReadOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedReadOptions.java
index a281722d2480..48c2238ec4a0 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedReadOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedReadOptions.java
@@ -18,26 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.ReadOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed {@link ReadOptions}.
  */
 public class ManagedReadOptions extends ReadOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
-
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectMetrics.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectMetrics.java
index 8d9d7fe78d7e..706aa9bb345f 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectMetrics.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectMetrics.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.metrics2.annotation.Metric;
 import org.apache.hadoop.metrics2.annotation.Metrics;
@@ -64,4 +65,14 @@ public void assertNoLeaks() {
   void increaseManagedObject() {
     totalManagedObjects.incr();
   }
+
+  @VisibleForTesting
+  long totalLeakObjects() {
+    return totalLeakObjects.value();
+  }
+
+  @VisibleForTesting
+  long totalManagedObjects() {
+    return totalManagedObjects.value();
+  }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectUtils.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectUtils.java
index b32f1111cbcb..7ae7001ccd39 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectUtils.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedRocksObjectUtils.java
@@ -19,10 +19,11 @@
 package org.apache.hadoop.hdds.utils.db.managed;
 
 import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.utils.LeakDetector;
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.awaitility.Awaitility;
 import org.awaitility.core.ConditionTimeoutException;
 import org.rocksdb.RocksDB;
-import org.rocksdb.RocksObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,38 +42,36 @@ private ManagedRocksObjectUtils() {
 
   public static final Logger LOG =
       LoggerFactory.getLogger(ManagedRocksObjectUtils.class);
+
   private static final Duration POLL_DELAY_DURATION = Duration.ZERO;
   private static final Duration POLL_INTERVAL_DURATION = Duration.ofMillis(100);
 
-  public static void assertClosed(ManagedObject<?> object) {
-    assertClosed(object.get(), object.getStackTrace());
-  }
 
-  static void assertClosed(RocksObject rocksObject, String stackTrace) {
+  private static final LeakDetector LEAK_DETECTOR = new LeakDetector("ManagedRocksObject");
+
+  static LeakTracker track(AutoCloseable object) {
     ManagedRocksObjectMetrics.INSTANCE.increaseManagedObject();
-    if (rocksObject.isOwningHandle()) {
-      reportLeak(rocksObject, stackTrace);
-    }
+    final Class<?> clazz = object.getClass();
+    final StackTraceElement[] stackTrace = getStackTrace();
+    return LEAK_DETECTOR.track(object, () -> reportLeak(clazz, formatStackTrace(stackTrace)));
   }
 
-  static void reportLeak(Object object, String stackTrace) {
+  static void reportLeak(Class<?> clazz, String stackTrace) {
     ManagedRocksObjectMetrics.INSTANCE.increaseLeakObject();
-    String warning = String.format("%s is not closed properly",
-        object.getClass().getSimpleName());
+    String warning = String.format("%s is not closed properly", clazz.getSimpleName());
     if (stackTrace != null && LOG.isDebugEnabled()) {
-      String debugMessage = String
-          .format("%nStackTrace for unclosed instance: %s", stackTrace);
+      String debugMessage = String.format("%nStackTrace for unclosed instance: %s", stackTrace);
       warning = warning.concat(debugMessage);
     }
     LOG.warn(warning);
   }
 
-  static @Nullable StackTraceElement[] getStackTrace() {
+  private static @Nullable StackTraceElement[] getStackTrace() {
     return HddsUtils.getStackTrace(LOG);
   }
 
   static String formatStackTrace(@Nullable StackTraceElement[] elements) {
-    return HddsUtils.formatStackTrace(elements, 3);
+    return HddsUtils.formatStackTrace(elements, 4);
   }
 
   /**
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSlice.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSlice.java
index 2de2031fb9c0..8c366bdaa423 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSlice.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSlice.java
@@ -18,23 +18,19 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.Slice;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed Slice.
  */
 public class ManagedSlice extends Slice {
+  private final LeakTracker leakTracker = track(this);
 
-  @Nullable
-  private final StackTraceElement[] elements;
-
-  public ManagedSlice(byte[] var1) {
-    super(var1);
-    this.elements = ManagedRocksObjectUtils.getStackTrace();
+  public ManagedSlice(byte[] data) {
+    super(data);
   }
 
   @Override
@@ -43,12 +39,10 @@ public synchronized long getNativeHandle() {
   }
 
   @Override
-  protected void finalize() throws Throwable {
-    ManagedRocksObjectMetrics.INSTANCE.increaseManagedObject();
-    if (isOwningHandle()) {
-      ManagedRocksObjectUtils.reportLeak(this, formatStackTrace(elements));
-    }
-    super.finalize();
+  protected void disposeInternal() {
+    super.disposeInternal();
+    // RocksMutableObject.close is final thus can't be decorated.
+    // So, we decorate disposeInternal instead to track closure.
+    leakTracker.close();
   }
-
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReader.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReader.java
index 7ba1001a432e..b49c6e7a9e49 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReader.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReader.java
@@ -29,15 +29,7 @@ public class ManagedSstFileReader extends ManagedObject<SstFileReader> {
     super(original);
   }
 
-  public static ManagedSstFileReader managed(
-      SstFileReader reader) {
+  public static ManagedSstFileReader managed(SstFileReader reader) {
     return new ManagedSstFileReader(reader);
   }
-
-  @Override
-  protected void finalize() throws Throwable {
-    ManagedRocksObjectUtils.assertClosed(this);
-    super.finalize();
-  }
-
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReaderIterator.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReaderIterator.java
index 0916e89571b1..660a7c347aee 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReaderIterator.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileReaderIterator.java
@@ -23,19 +23,12 @@
 /**
  * Managed SstFileReaderIterator.
  */
-public class ManagedSstFileReaderIterator
-    extends ManagedObject<SstFileReaderIterator> {
+public class ManagedSstFileReaderIterator extends ManagedObject<SstFileReaderIterator> {
 
   ManagedSstFileReaderIterator(SstFileReaderIterator original) {
     super(original);
   }
 
-  @Override
-  protected void finalize() throws Throwable {
-    ManagedRocksObjectUtils.assertClosed(this);
-    super.finalize();
-  }
-
   public static ManagedSstFileReaderIterator managed(
       SstFileReaderIterator iterator) {
     return new ManagedSstFileReaderIterator(iterator);
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileWriter.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileWriter.java
index e6fdcbc2ed0b..de7e9d526634 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileWriter.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedSstFileWriter.java
@@ -18,23 +18,18 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.EnvOptions;
 import org.rocksdb.Options;
 import org.rocksdb.SstFileWriter;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed SstFileWriter.
  */
 public class ManagedSstFileWriter extends SstFileWriter {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   public ManagedSstFileWriter(EnvOptions envOptions,
                               Options options) {
@@ -42,8 +37,8 @@ public ManagedSstFileWriter(EnvOptions envOptions,
   }
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedStatistics.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedStatistics.java
index 4cbd6f98287b..75af8b881355 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedStatistics.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedStatistics.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.Statistics;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed Statistics.
  */
 public class ManagedStatistics extends Statistics {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteBatch.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteBatch.java
index 51c9bcc0df02..b1411b09a49a 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteBatch.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteBatch.java
@@ -18,24 +18,18 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.WriteBatch;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed WriteBatch.
  */
 public class ManagedWriteBatch extends WriteBatch {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   public ManagedWriteBatch() {
-    super();
   }
 
   public ManagedWriteBatch(byte[] data) {
@@ -43,8 +37,8 @@ public ManagedWriteBatch(byte[] data) {
   }
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteOptions.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteOptions.java
index 7ba05a0ee6d9..5d32a290b5e2 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteOptions.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/ManagedWriteOptions.java
@@ -18,25 +18,20 @@
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
+import org.apache.hadoop.hdds.utils.LeakTracker;
 import org.rocksdb.WriteOptions;
 
-import javax.annotation.Nullable;
-
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.assertClosed;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.formatStackTrace;
-import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.getStackTrace;
+import static org.apache.hadoop.hdds.utils.db.managed.ManagedRocksObjectUtils.track;
 
 /**
  * Managed {@link WriteOptions}.
  */
 public class ManagedWriteOptions extends WriteOptions {
-
-  @Nullable
-  private final StackTraceElement[] elements = getStackTrace();
+  private final LeakTracker leakTracker = track(this);
 
   @Override
-  protected void finalize() throws Throwable {
-    assertClosed(this, formatStackTrace(elements));
-    super.finalize();
+  public void close() {
+    super.close();
+    leakTracker.close();
   }
 }
diff --git a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/package-info.java b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/package-info.java
index c60dd4c82ec9..ee214f8150ed 100644
--- a/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/package-info.java
+++ b/hadoop-hdds/managed-rocksdb/src/main/java/org/apache/hadoop/hdds/utils/db/managed/package-info.java
@@ -18,18 +18,16 @@
  */
 
 /**
- * RocksDB is deprecating the RocksObject's finalizer that cleans up native
- * resources. In fact, the finalizer is removed in the new version of RocksDB
- * as per https://github.com/facebook/rocksdb/commit/99d86252b. That poses a
+ * RocksDB has deprecated the RocksObject's finalizer that cleans up native
+ * resources, see https://github.com/facebook/rocksdb/commit/99d86252b. That poses a
  * requirement for RocksDb's applications to explicitly close RocksObject
  * instances themselves to avoid leaking native resources. The general approach
  * is to close RocksObjects with try-with-resource statement.
  * Yet, this is not always an easy option in Ozone we need a mechanism to
  * manage and detect leaks.
  *
- * This package contains wrappers and utilities to catch RocksObject
- * instantiates in Ozone, intercept their finalizers and assert if the created
- * instances are closed properly before being GCed.
+ * This package contains RocksObject decorators and utilities to catch track RocksObject's
+ * lifecycle to ensure they're properly closed before being GCed.
  */
 package org.apache.hadoop.hdds.utils.db.managed;
 
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/utils/db/managed/TestRocksObjectLeakDetector.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/utils/db/managed/TestRocksObjectLeakDetector.java
new file mode 100644
index 000000000000..87fbe23ac761
--- /dev/null
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/utils/db/managed/TestRocksObjectLeakDetector.java
@@ -0,0 +1,114 @@
+/*
+ * 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.hadoop.hdds.utils.db.managed;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.ozone.test.tag.Unhealthy;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.UUID;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Test managed rocks object leak detection.
+ * This test creates garbage that will fail other tests and is intended for manual run only.
+ * It is also flaky because of depending on background processes and environment (other background tasks
+ * can create extra managed rocks objects and thus fails the counter assertions).
+ */
+@Unhealthy
+public class TestRocksObjectLeakDetector {
+
+  private static MiniOzoneCluster cluster;
+
+  @BeforeAll
+  static void setUp() throws IOException, InterruptedException,
+      TimeoutException {
+    OzoneConfiguration conf = new OzoneConfiguration();
+    String clusterId = UUID.randomUUID().toString();
+    String scmId = UUID.randomUUID().toString();
+    String omServiceId = "omServiceId1";
+    cluster = MiniOzoneCluster.newBuilder(conf)
+        .setClusterId(clusterId)
+        .setScmId(scmId)
+        .setOMServiceId(omServiceId)
+        .setNumOfOzoneManagers(1)
+        .build();
+    cluster.waitForClusterToBeReady();
+  }
+
+  @AfterAll
+  static void cleanUp() {
+    if (cluster != null) {
+      assertThrows(AssertionError.class, () -> cluster.shutdown());
+    }
+  }
+
+  @Test
+  public void testLeakDetector() throws Exception {
+    testLeakDetector(ManagedBloomFilter::new);
+    testLeakDetector(ManagedColumnFamilyOptions::new);
+    testLeakDetector(ManagedEnvOptions::new);
+    testLeakDetector(ManagedFlushOptions::new);
+    testLeakDetector(ManagedIngestExternalFileOptions::new);
+    testLeakDetector(() -> new ManagedLRUCache(1L));
+    testLeakDetector(ManagedOptions::new);
+    testLeakDetector(ManagedReadOptions::new);
+    testLeakDetector(() -> new ManagedSlice(RandomUtils.nextBytes(10)));
+    testLeakDetector(ManagedStatistics::new);
+    testLeakDetector(ManagedWriteBatch::new);
+    testLeakDetector(ManagedWriteOptions::new);
+  }
+
+  private <T extends AutoCloseable> void testLeakDetector(Supplier<T> supplier) throws Exception {
+    // base metrics
+    long managedObjects = ManagedRocksObjectMetrics.INSTANCE.totalManagedObjects();
+    long leakObjects = ManagedRocksObjectMetrics.INSTANCE.totalLeakObjects();
+
+    // Allocate and close.
+    allocate(supplier, true);
+    System.gc();
+    // it could take a while for leak detection to run in the background.
+    Thread.sleep(500);
+    assertEquals(managedObjects + 1, ManagedRocksObjectMetrics.INSTANCE.totalManagedObjects());
+    assertEquals(leakObjects, ManagedRocksObjectMetrics.INSTANCE.totalLeakObjects());
+
+    // Allocate and not close.
+    allocate(supplier, false);
+    System.gc();
+    Thread.sleep(500);
+    assertEquals(managedObjects + 2, ManagedRocksObjectMetrics.INSTANCE.totalManagedObjects());
+    assertEquals(leakObjects + 1, ManagedRocksObjectMetrics.INSTANCE.totalLeakObjects());
+  }
+
+  private <T extends AutoCloseable> void allocate(Supplier<T> supplier, boolean close) throws Exception {
+    T object = supplier.get();
+    if (close) {
+      object.close();
+    }
+  }
+}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
index e2e737356964..50c89dd3e688 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
@@ -448,15 +448,14 @@ public String getClusterId() {
   public void shutdown() {
     try {
       LOG.info("Shutting down the Mini Ozone Cluster");
+      CodecTestUtil.gc();
       IOUtils.closeQuietly(clients);
       final File baseDir = new File(getBaseDir());
       stop();
       FileUtils.deleteDirectory(baseDir);
       ContainerCache.getInstance(conf).shutdownCache();
       DefaultMetricsSystem.shutdown();
-
       ManagedRocksObjectMetrics.INSTANCE.assertNoLeaks();
-      CodecTestUtil.gc();
     } catch (Exception e) {
       LOG.error("Exception while shutting down the cluster.", e);
     }