[improve] Improve logic for enabling Netty leak detection (apache#23613)

pulsar-common/src/main/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocator.java

@@ -21,9 +21,13 @@
 import com.google.common.annotations.VisibleForTesting;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.PooledByteBufAllocator;
+import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import lombok.experimental.UtilityClass;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.common.allocator.ByteBufAllocatorBuilder;
@@ -44,6 +48,13 @@ public class PulsarByteBufAllocator {
     public static final String PULSAR_ALLOCATOR_LEAK_DETECTION = "pulsar.allocator.leak_detection";
     public static final String PULSAR_ALLOCATOR_OUT_OF_MEMORY_POLICY = "pulsar.allocator.out_of_memory_policy";
 
+    // the highest level of leak detection policy will be used when it is set by any of the following property names
+    private static final String[] LEAK_DETECTION_PROPERTY_NAMES = {
+            PULSAR_ALLOCATOR_LEAK_DETECTION,
+            "io.netty.leakDetection.level", // io.netty.util.ResourceLeakDetector.PROP_LEVEL
+            "io.netty.leakDetectionLevel" // io.netty.util.ResourceLeakDetector.PROP_LEVEL_OLD
+    };
+
     public static final ByteBufAllocator DEFAULT;
 
     private static final List<Consumer<OutOfMemoryError>> LISTENERS = new CopyOnWriteArrayList<>();
@@ -64,8 +75,7 @@ static ByteBufAllocator createByteBufAllocator() {
         final OutOfMemoryPolicy outOfMemoryPolicy = OutOfMemoryPolicy.valueOf(
                 System.getProperty(PULSAR_ALLOCATOR_OUT_OF_MEMORY_POLICY, "FallbackToHeap"));
 
-        final LeakDetectionPolicy leakDetectionPolicy = LeakDetectionPolicy
-                .valueOf(System.getProperty(PULSAR_ALLOCATOR_LEAK_DETECTION, "Disabled"));
+        final LeakDetectionPolicy leakDetectionPolicy = resolveLeakDetectionPolicyWithHighestLevel(System::getProperty);
         if (log.isDebugEnabled()) {
             log.debug("Is Pooled: {} -- Exit on OOM: {}", isPooled, isExitOnOutOfMemory);
         }
@@ -98,4 +108,19 @@ static ByteBufAllocator createByteBufAllocator() {
         return builder.build();
 
     }
+
+    /**
+     * Resolve the leak detection policy. The value is resolved from the system properties in
+     * the order of LEAK_DETECTION_PROPERTY_NAMES.
+     * @return parsed leak detection policy
+     */
+    @VisibleForTesting
+    static LeakDetectionPolicy resolveLeakDetectionPolicyWithHighestLevel(Function<String, String> propertyResolver) {
+        return Arrays.stream(LEAK_DETECTION_PROPERTY_NAMES)
+                .map(propertyResolver)
+                .filter(Objects::nonNull)
+                .map(LeakDetectionPolicy::parseLevel)
+                .max(Comparator.comparingInt(Enum::ordinal))
+                .orElse(LeakDetectionPolicy.Disabled);
+    }
 }

pulsar-common/src/test/java/org/apache/pulsar/common/allocator/PulsarByteBufAllocatorTest.java

@@ -0,0 +1,76 @@
+/*
+ * 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.pulsar.common.allocator;
+
+import static org.testng.Assert.assertEquals;
+import java.util.Properties;
+import org.apache.bookkeeper.common.allocator.LeakDetectionPolicy;
+import org.testng.annotations.Test;
+
+public class PulsarByteBufAllocatorTest {
+
+    @Test
+    public void testResolveLeakDetectionPolicyWithHighestLevel() {
+        Properties properties = new Properties();
+        properties.setProperty("io.netty.leakDetectionLevel", "paranoid");
+        properties.setProperty("io.netty.leakDetection.level", "advanced");
+        properties.setProperty("pulsar.allocator.leak_detection", "simple");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Paranoid);
+
+        properties.setProperty("io.netty.leakDetectionLevel", "advanced");
+        properties.setProperty("io.netty.leakDetection.level", "simple");
+        properties.setProperty("pulsar.allocator.leak_detection", "paranoid");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Paranoid);
+
+        properties.setProperty("io.netty.leakDetectionLevel", "simple");
+        properties.setProperty("io.netty.leakDetection.level", "paranoid");
+        properties.setProperty("pulsar.allocator.leak_detection", "advanced");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Paranoid);
+
+        properties.setProperty("io.netty.leakDetectionLevel", "disabled");
+        properties.setProperty("io.netty.leakDetection.level", "simple");
+        properties.setProperty("pulsar.allocator.leak_detection", "disabled");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Simple);
+
+        properties.setProperty("io.netty.leakDetectionLevel", "invalid");
+        properties.setProperty("io.netty.leakDetection.level", "invalid");
+        properties.setProperty("pulsar.allocator.leak_detection", "invalid");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Disabled);
+
+        properties.clear();
+        properties.setProperty("pulsar.allocator.leak_detection", "Paranoid");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Paranoid);
+
+        properties.clear();
+        properties.setProperty("io.netty.leakDetectionLevel", "Advanced");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Advanced);
+
+        properties.clear();
+        properties.setProperty("io.netty.leakDetection.level", "Simple");
+        assertEquals(PulsarByteBufAllocator.resolveLeakDetectionPolicyWithHighestLevel(properties::getProperty),
+                LeakDetectionPolicy.Simple);
+    }
+}