diff --git a/bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/LeakDetectionPolicy.java b/bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/LeakDetectionPolicy.java
index f5c99a7e030..90d19a4aeef 100644
--- a/bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/LeakDetectionPolicy.java
+++ b/bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/LeakDetectionPolicy.java
@@ -17,9 +17,12 @@
  */
 package org.apache.bookkeeper.common.allocator;
 
+import lombok.extern.slf4j.Slf4j;
+
 /**
  * Define the policy for the Netty leak detector.
  */
+@Slf4j
 public enum LeakDetectionPolicy {
 
     /**
@@ -43,5 +46,17 @@ public enum LeakDetectionPolicy {
      * stack traces of places where the buffer was used. Introduce very
      * significant overhead.
      */
-    Paranoid,
+    Paranoid;
+
+    public static LeakDetectionPolicy parseLevel(String levelStr) {
+        String trimmedLevelStr = levelStr.trim();
+        for (LeakDetectionPolicy policy : values()) {
+            if (trimmedLevelStr.equalsIgnoreCase(policy.name())) {
+                return policy;
+            }
+        }
+        log.warn("Parse leak detection policy level {} failed. Use the default level: {}", levelStr,
+                LeakDetectionPolicy.Disabled.name());
+        return LeakDetectionPolicy.Disabled;
+    }
 }
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
index 21bde894a36..438dc40983e 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java
@@ -1129,8 +1129,17 @@ public T setAllocatorOutOfMemoryPolicy(OutOfMemoryPolicy oomPolicy) {
      * Return the configured leak detection policy for the allocator.
      */
     public LeakDetectionPolicy getAllocatorLeakDetectionPolicy() {
-        return LeakDetectionPolicy
-                .valueOf(this.getString(ALLOCATOR_LEAK_DETECTION_POLICY, LeakDetectionPolicy.Disabled.toString()));
+        //see: https://lists.apache.org/thread/d3zw8bxhlg0wxfhocyjglq0nbxrww3sg
+        String nettyLevelStr = System.getProperty("io.netty.leakDetectionLevel", LeakDetectionPolicy.Disabled.name());
+        nettyLevelStr = System.getProperty("io.netty.leakDetection.level", nettyLevelStr);
+        String bkLevelStr = getString(ALLOCATOR_LEAK_DETECTION_POLICY, LeakDetectionPolicy.Disabled.name());
+        LeakDetectionPolicy nettyLevel = LeakDetectionPolicy.parseLevel(nettyLevelStr);
+        LeakDetectionPolicy bkLevel = LeakDetectionPolicy.parseLevel(bkLevelStr);
+        if (nettyLevel.ordinal() >= bkLevel.ordinal()) {
+            return nettyLevel;
+        } else {
+            return bkLevel;
+        }
     }
 
     /**
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java
index 54e07bd5992..a6333a47d32 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/conf/AbstractConfigurationTest.java
@@ -22,6 +22,7 @@
 import static org.mockito.Mockito.CALLS_REAL_METHODS;
 import static org.mockito.Mockito.mock;
 
+import org.apache.bookkeeper.common.allocator.LeakDetectionPolicy;
 import org.apache.bookkeeper.meta.AbstractZkLedgerManagerFactory;
 import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
 import org.apache.bookkeeper.meta.LedgerManagerFactory;
@@ -130,4 +131,52 @@ public void testUnknownZkLedgerManagerFactory() throws Exception {
         conf.getMetadataServiceUri();
     }
 
+    @Test
+    public void testAllocatorLeakDetectionPolicy() {
+        String nettyOldLevelKey = "io.netty.leakDetectionLevel";
+        String nettyLevelKey = "io.netty.leakDetection.level";
+
+        String nettyOldLevelStr = System.getProperty(nettyOldLevelKey);
+        String nettyLevelStr = System.getProperty(nettyLevelKey);
+
+        //Remove netty property for test.
+        System.getProperties().remove(nettyOldLevelKey);
+        System.getProperties().remove(nettyLevelKey);
+
+        assertEquals(LeakDetectionPolicy.Disabled, conf.getAllocatorLeakDetectionPolicy());
+
+        System.getProperties().put(nettyOldLevelKey, "zazaza");
+        assertEquals(LeakDetectionPolicy.Disabled, conf.getAllocatorLeakDetectionPolicy());
+
+        conf.setProperty(AbstractConfiguration.ALLOCATOR_LEAK_DETECTION_POLICY, "zazaza");
+        assertEquals(LeakDetectionPolicy.Disabled, conf.getAllocatorLeakDetectionPolicy());
+
+        System.getProperties().put(nettyOldLevelKey, "simple");
+        assertEquals(LeakDetectionPolicy.Simple, conf.getAllocatorLeakDetectionPolicy());
+
+        System.getProperties().put(nettyLevelKey, "disabled");
+        assertEquals(LeakDetectionPolicy.Disabled, conf.getAllocatorLeakDetectionPolicy());
+
+        System.getProperties().put(nettyLevelKey, "advanCed");
+        assertEquals(LeakDetectionPolicy.Advanced, conf.getAllocatorLeakDetectionPolicy());
+
+        conf.setProperty(AbstractConfiguration.ALLOCATOR_LEAK_DETECTION_POLICY, "simPle");
+        assertEquals(LeakDetectionPolicy.Advanced, conf.getAllocatorLeakDetectionPolicy());
+
+        conf.setProperty(AbstractConfiguration.ALLOCATOR_LEAK_DETECTION_POLICY, "advanCed");
+        assertEquals(LeakDetectionPolicy.Advanced, conf.getAllocatorLeakDetectionPolicy());
+
+        conf.setProperty(AbstractConfiguration.ALLOCATOR_LEAK_DETECTION_POLICY, "paranoiD");
+        assertEquals(LeakDetectionPolicy.Paranoid, conf.getAllocatorLeakDetectionPolicy());
+
+        System.getProperties().remove(nettyOldLevelKey);
+        System.getProperties().remove(nettyLevelKey);
+        //Revert the netty properties.
+        if (nettyOldLevelStr != null) {
+            System.getProperties().put(nettyOldLevelKey, nettyOldLevelStr);
+        }
+        if (nettyLevelStr != null) {
+            System.getProperties().put(nettyLevelKey, nettyLevelStr);
+        }
+    }
 }