[fix][test] Restore NETTY_LEAK_DETECTION for Gradle and CI tests

.github/workflows/pulsar-ci.yaml

@@ -493,6 +493,8 @@ jobs:
       JOB_NAME: CI - Integration - ${{ matrix.name }}
       PULSAR_TEST_IMAGE_NAME: apachepulsar/java-test-image:latest
       CI_JDK_MAJOR_VERSION: ${{ needs.preconditions.outputs.jdk_major_version }}
+      NETTY_LEAK_DETECTION: "${{ needs.preconditions.outputs.netty_leak_detection }}"
+      NETTY_LEAK_DUMP_DIR: ${{ github.workspace }}/build/netty-leak-dumps
     strategy:
       fail-fast: false
       matrix:
@@ -617,6 +619,10 @@ jobs:
         if: ${{ always() }}
         uses: ./.github/actions/copy-test-reports
 
+      - name: Report detected Netty leaks
+        if: ${{ always() && env.NETTY_LEAK_DETECTION != 'off' }}
+        run: $GITHUB_WORKSPACE/pulsar-build/pulsar_ci_tool.sh report_netty_leaks
+
       - name: Upload test reports
         uses: actions/upload-artifact@v7
         if: ${{ !success() }}
@@ -638,6 +644,7 @@ jobs:
             **/hs_err_*.log
             **/core.*
             build/threaddumps/
+            ${{ env.NETTY_LEAK_DUMP_DIR }}/*
           retention-days: 7
           if-no-files-found: ignore
 
@@ -754,6 +761,8 @@ jobs:
       JOB_NAME: CI - System - ${{ matrix.name }}
       PULSAR_TEST_IMAGE_NAME: apachepulsar/pulsar-test-latest-version:latest
       CI_JDK_MAJOR_VERSION: ${{ needs.preconditions.outputs.jdk_major_version }}
+      NETTY_LEAK_DETECTION: "${{ needs.preconditions.outputs.netty_leak_detection }}"
+      NETTY_LEAK_DUMP_DIR: ${{ github.workspace }}/build/netty-leak-dumps
     strategy:
       fail-fast: false
       matrix:
@@ -833,6 +842,10 @@ jobs:
         if: ${{ always() }}
         uses: ./.github/actions/copy-test-reports
 
+      - name: Report detected Netty leaks
+        if: ${{ always() && env.NETTY_LEAK_DETECTION != 'off' }}
+        run: $GITHUB_WORKSPACE/pulsar-build/pulsar_ci_tool.sh report_netty_leaks
+
       - name: Upload test reports
         uses: actions/upload-artifact@v7
         if: ${{ !success() }}
@@ -854,6 +867,7 @@ jobs:
             **/hs_err_*.log
             **/core.*
             build/threaddumps/
+            ${{ env.NETTY_LEAK_DUMP_DIR }}/*
           retention-days: 7
           if-no-files-found: ignore
 

CONTRIBUTING.md

@@ -131,6 +131,28 @@ Failed tests are retried once by default (`testRetryCount=1`; `0` when running i
 running tests locally, prefer **`-PtestRetryCount=0`** to catch failures (including flakiness) early
 instead of having retries mask them.
 
+### Netty buffer leak detection
+
+Tests enable Netty's `paranoid` leak detection by default, using `ExtendedNettyLeakDetector`
+to include test names in leak reports and write `netty_leak_*.txt` files. Set
+`NETTY_LEAK_DUMP_DIR` to choose the output directory (the default is the JVM's temporary directory).
+
+`NETTY_LEAK_DETECTION=report` (the default) reports leaks without failing tests.
+In CI, `NETTY_LEAK_DETECTION=fail_on_leak` makes the leak-reporting step fail the job when dumps
+are found. Unit, integration, and system test jobs collect dumps from both the test JVMs and
+Pulsar Docker containers, including reports generated during container shutdown. For local tests, use `-PtestExitJvmOnLeak=true` to fail the test JVM on a detected leak:
+
+```bash
+NETTY_LEAK_DUMP_DIR=/tmp/pulsar-netty-leaks ./gradlew :pulsar-client-original:test \
+  --tests "ConsumerBuilderImplTest" -PtestExitJvmOnLeak=true -PtestRetryCount=0
+```
+
+Set `NETTY_LEAK_DETECTION=off` to disable detection, or use
+`-PtestLeakDetectionLevel=simple|advanced|paranoid|disabled` to change its level.
+`-PtestExitJvmOnLeakDelayMillis=1000` controls the delay before exiting on a leak.
+Detection is disabled automatically for `-PtestAsyncProfiler` and `profilingIntegrationTest`
+to avoid distorting profiles.
+
 ### Micro benchmarks (JMH)
 
 For a **micro**-level question — what a single method, data structure or codec costs — write a

build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts

@@ -205,6 +205,9 @@ val javaToolchains = extensions.getByType<JavaToolchainService>()
 // Effective Java major version used to run tests: the -PtestJavaVersion override when set,
 // otherwise the JVM running Gradle.
 val testJavaMajorVersion = testJavaVersion.orNull ?: JavaVersion.current().majorVersion.toInt()
+val asyncProfilerEnabled = providers.gradleProperty("testAsyncProfiler")
+    .map { it.isBlank() || it.toBoolean() }
+    .getOrElse(false)
 
 tasks.withType<Test>().configureEach {
     testJavaVersion.orNull?.let { version ->
@@ -248,6 +251,27 @@ tasks.withType<Test>().configureEach {
     val defaultTestRetryCount = if (ideaActive) "0" else "1"
     systemProperty("testRetryCount", providers.gradleProperty("testRetryCount").getOrElse(defaultTestRetryCount))
     systemProperty("testFailFast", failFastValue.toString())
+    // Restore the test leak detector defaults from the Maven build. CI's report_netty_leaks
+    // step handles report vs. fail_on_leak after collecting the dumps from all test JVMs.
+    val nettyLeakDetectionEnabled =
+        providers.environmentVariable("NETTY_LEAK_DETECTION").getOrElse("report") != "off" && !asyncProfilerEnabled
+    if (nettyLeakDetectionEnabled) {
+        systemProperty("io.netty.customResourceLeakDetector", "org.apache.pulsar.tests.ExtendedNettyLeakDetector")
+        systemProperty("org.apache.pulsar.tests.ExtendedNettyLeakDetector.exitJvmOnLeak",
+            providers.gradleProperty("testExitJvmOnLeak").getOrElse("false"))
+        systemProperty("org.apache.pulsar.tests.ExtendedNettyLeakDetector.exitJvmDelayMillis",
+            providers.gradleProperty("testExitJvmOnLeakDelayMillis").getOrElse("1000"))
+        systemProperty("io.netty.leakDetection.level",
+            providers.gradleProperty("testLeakDetectionLevel").getOrElse("paranoid"))
+        // Track every allocation with less overhead by recording only acquire/release operations.
+        systemProperty("io.netty.leakDetection.targetRecords", "16")
+        systemProperty("io.netty.leakDetection.acquireAndReleaseOnly", "true")
+        systemProperty("io.netty.leakDetection.samplingInterval", "32")
+        // Process weak references promptly when the test listener triggers leak detection.
+        jvmArgs("-XX:+UnlockExperimentalVMOptions", "-XX:ReferencesPerThread=0", "-XX:+ParallelRefProcEnabled")
+    } else {
+        systemProperty("io.netty.leakDetection.level", "disabled")
+    }
     jvmArgs(
         "-XX:+HeapDumpOnOutOfMemoryError",
         "-XX:HeapDumpPath=${providers.gradleProperty("testHeapDumpPath").getOrElse("/tmp")}",
@@ -297,9 +321,6 @@ tasks.withType<Test>().configureEach {
 // the names of the `testAsyncProfiler` Maven profile that the 4.x branches use. This is a second
 // `configureEach` block so that it overrides the settings above, and so that the environment
 // variable and JDK lookups it does stay out of the configuration cache inputs when profiling is off.
-val asyncProfilerEnabled = providers.gradleProperty("testAsyncProfiler")
-    .map { it.isBlank() || it.toBoolean() }
-    .getOrElse(false)
 if (asyncProfilerEnabled) {
     // Locate the agent library: an explicit -Ptest.asyncprofiler.libpath wins, then the
     // LIBASYNCPROFILER_PATH environment variable (the variable microbench/README.md already uses for

pulsar-build/pulsar_ci_tool.sh

@@ -182,12 +182,12 @@ ci_report_netty_leaks() {
   fi
 
   # check if there are any netty_leak_*.txt files in the container logs
-  local container_logs_dir="tests/integration/target/container-logs"
+  local container_logs_dir="tests/integration/build/container-logs"
   if [ -d "$container_logs_dir" ]; then
     local container_netty_leak_dump_dir="$NETTY_LEAK_DUMP_DIR/container-logs"
     mkdir -p "$container_netty_leak_dump_dir"
     while read -r file; do
-      # example file name "tests/integration/target/container-logs/ltnizrzm-standalone/var-log-pulsar.tar.gz"
+      # example file name "tests/integration/build/container-logs/ltnizrzm-standalone/var-log-pulsar.tar.gz"
       # take ltnizrzm-standalone part
       container_name=$(basename "$(dirname "$file")")
       target_dir="$container_netty_leak_dump_dir/$container_name"

tests/integration/build.gradle.kts

@@ -259,6 +259,8 @@ tasks.register<Test>("profilingIntegrationTest") {
         "${dockerOrganization}/java-test-image:${dockerTag}-asyncprofiler")
     // Leak detection is paranoid by default and would distort the allocation profile.
     environment("NETTY_LEAK_DETECTION", "off")
+    systemProperties.remove("io.netty.customResourceLeakDetector")
+    systemProperty("io.netty.leakDetection.level", "disabled")
 
     // A retried test would profile the cluster twice into the same run.
     systemProperty("testRetryCount", "0")

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/NettyLeakDetectionTest.java

@@ -0,0 +1,130 @@
+/*
+ * 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.tests.integration.containers;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.assertj.core.api.Assertions.assertThat;
+import io.netty.buffer.ByteBufAllocator;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.zip.GZIPInputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
+import org.apache.commons.io.FileUtils;
+import org.apache.pulsar.tests.ExtendedNettyLeakDetector;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.MountableFile;
+import org.testng.SkipException;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class NettyLeakDetectionTest {
+    @DataProvider
+    public Object[][] shutdownModes() {
+        return new Object[][] {{false}, {true}};
+    }
+
+    @Test(dataProvider = "shutdownModes")
+    public void collectsLeaksReportedDuringShutdown(boolean supervised) throws Exception {
+        if (!ExtendedNettyLeakDetector.isExtendedNettyLeakDetectorEnabled()
+                || !"paranoid".equals(System.getProperty("io.netty.leakDetection.level"))) {
+            throw new SkipException("Requires the default paranoid test leak detector");
+        }
+        var container = new LeakProbeContainer(supervised);
+        Path logs = Path.of(System.getProperty("buildDirectory", "build"),
+                "container-logs", container.getContainerName());
+        try (container) {
+            container.start();
+            assertThat(container.execCmd("sh", "-c", "ls /var/log/pulsar/netty_leak_*.txt 2>/dev/null || true")
+                    .getStdout()).as("No leak report before JVM shutdown").isEmpty();
+            container.stop();
+            Path archive = logs.resolve("var-log-pulsar.tar.gz");
+            boolean foundLeak = false;
+            try (var tar = new TarArchiveInputStream(new GZIPInputStream(Files.newInputStream(archive)))) {
+                TarArchiveEntry entry;
+                while ((entry = tar.getNextEntry()) != null) {
+                    if (entry.isFile() && entry.getName().contains("netty_leak_")) {
+                        String report = new String(tar.readAllBytes(), UTF_8);
+                        assertThat(report).contains("Traced leak detected ByteBuf", "container-shutdown-leak");
+                        foundLeak = true;
+                    }
+                }
+            }
+            assertThat(foundLeak).as("Shutdown leak included in the collected container logs").isTrue();
+        } finally {
+            // This test deliberately leaks in a separate JVM. Do not report its expected leak in CI.
+            FileUtils.deleteDirectory(logs.toFile());
+        }
+    }
+
+    private static class LeakProbeContainer extends PulsarContainer<LeakProbeContainer> {
+        LeakProbeContainer(boolean supervised) {
+            super("leak-test-" + UUID.randomUUID(), "probe", "probe",
+                    supervised ? "/usr/bin/supervisord" : "bin/pulsar", INVALID_PORT, INVALID_PORT);
+            String className = LeakProbe.class.getName();
+            String resource = className.replace('.', '/') + ".class";
+            withCopyFileToContainer(MountableFile.forClasspathResource(resource), "/tmp/" + resource);
+            String script = "#!/bin/sh\nexec java $PULSAR_EXTRA_OPTS -cp '/pulsar/lib/*:/tmp' '"
+                    + className + "'\n";
+            if (supervised) {
+                withCopyToContainer(Transferable.of(script, 0755), "/tmp/leak-probe.sh");
+                withCopyToContainer(Transferable.of("""
+                        [program:leak-probe]
+                        command=/tmp/leak-probe.sh
+                        autostart=true
+                        autorestart=false
+                        stopwaitsecs=15
+                        """), "/etc/supervisord/conf.d/leak-probe.conf");
+                withCommand("-c", "/etc/supervisord.conf");
+            } else {
+                // Use the standalone shutdown path with a small JVM instead of starting a broker.
+                withCopyToContainer(Transferable.of(script, 0755), "/pulsar/bin/pulsar");
+                withCommand();
+            }
+            waitingFor(Wait.forSuccessfulCommand("test -f /tmp/leak-probe-ready")
+                    .withStartupTimeout(Duration.ofSeconds(60)));
+        }
+
+        @Override
+        protected void passNettyLeakDetectionSystemProperties() {
+            super.passNettyLeakDetectionSystemProperties();
+            // Keep the deliberate leak alive until shutdown even when local tests fail on leaks.
+            appendToEnv("PULSAR_EXTRA_OPTS",
+                    "-D" + ExtendedNettyLeakDetector.EXIT_JVM_ON_LEAK_SYSTEM_PROPERTY_NAME + "=false");
+        }
+    }
+
+    public static class LeakProbe {
+        public static void main(String[] args) throws Exception {
+            ExtendedNettyLeakDetector.setInitialHint("container-shutdown-leak");
+            leakBuffer();
+            Files.writeString(Path.of("/tmp/leak-probe-ready"), "ready");
+            // Only the detector's shutdown hook will force collection and report the leaked buffer.
+            new CountDownLatch(1).await();
+        }
+
+        private static void leakBuffer() {
+            ByteBufAllocator.DEFAULT.directBuffer(16).writeLong(42);
+        }
+    }
+}

tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java

@@ -172,11 +172,6 @@ public static void configureLeaveContainerRunning(
     protected void beforeStop() {
         super.beforeStop();
         if (null != getContainerId()) {
-            DockerUtils.dumpContainerDirToTargetCompressed(
-                getDockerClient(),
-                getContainerId(),
-                "/var/log/pulsar"
-            );
             try {
                 // stop the "tail -f ..." commands started in afterStart method
                 // so that shutdown output doesn't clutter logs
@@ -199,26 +194,38 @@ public void stop() {
 
     @Override
     protected void doStop() {
-        if (getContainerId() != null) {
-            if (serviceEntryPoint.equals("bin/pulsar")) {
-                // attempt graceful shutdown using "docker stop"
-                dockerClient.stopContainerCmd(getContainerId())
-                        .withTimeout(15)
-                        .exec();
-            } else {
-                // use "supervisorctl stop all" for graceful shutdown
-                try {
-                    ContainerExecResult result = execCmd("/usr/bin/supervisorctl", "stop", "all");
-                    log.info().attr("exitCode", result.getExitCode())
-                            .attr("stdout", result.getStdout())
-                            .attr("stderr", result.getStderr())
-                            .log("Stopped supervisor services");
-                } catch (Exception e) {
-                    log.error().exception(e).log("Cannot run 'supervisorctl stop all'");
+        try {
+            if (getContainerId() != null) {
+                if (serviceEntryPoint.equals("bin/pulsar")) {
+                    // attempt graceful shutdown using "docker stop"
+                    dockerClient.stopContainerCmd(getContainerId())
+                            .withTimeout(15)
+                            .exec();
+                } else {
+                    // use "supervisorctl stop all" for graceful shutdown
+                    try {
+                        ContainerExecResult result = execCmd("/usr/bin/supervisorctl", "stop", "all");
+                        log.info().attr("exitCode", result.getExitCode())
+                                .attr("stdout", result.getStdout())
+                                .attr("stderr", result.getStderr())
+                                .log("Stopped supervisor services");
+                    } catch (Exception e) {
+                        log.error().exception(e).log("Cannot run 'supervisorctl stop all'");
+                    }
+                }
+            }
+        } finally {
+            try {
+                if (getContainerId() != null) {
+                    // The leak detector's JVM shutdown hook can produce additional reports. Copy them
+                    // after stopping the services, while the container still exists.
+                    DockerUtils.dumpContainerDirToTargetCompressed(
+                            getDockerClient(), getContainerId(), "/var/log/pulsar");
                 }
+            } finally {
+                super.doStop();
             }
         }
-        super.doStop();
     }
 
     @Override

tests/integration/src/test/resources/pulsar-standalone.xml

@@ -22,6 +22,7 @@
 <suite name="Pulsar Standalone Tests" verbose="2" annotations="JDK">
     <test name="pulsar-standalone-suite" preserve-order="true" >
         <classes>
+            <class name="org.apache.pulsar.tests.integration.containers.NettyLeakDetectionTest" />
             <class name="org.apache.pulsar.tests.integration.standalone.SmokeTest" />
         </classes>
     </test>