From 3e3c99343fdcc822a23c0066342053aa5844bf9d Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 16 Sep 2026 17:32:34 +0300 Subject: [PATCH 1/4] [fix][build] Restore Netty leak detection for Gradle tests Enable ExtendedNettyLeakDetector with the branch-4.2 paranoid defaults and restore the testLeakDetectionLevel and testExitJvmOnLeak options. Honor NETTY_LEAK_DETECTION=off and keep test profiling free of leak-detection overhead. Document local and CI usage. Validated with quickCheck, assemble, a scoped buildtools test, configuration-cache reuse, and an isolated intentional ByteBuf leak probe covering default/report/fail_on_leak/off modes, level overrides, profiling, and JVM exit on leak. Assisted-by: Codex --- CONTRIBUTING.md | 21 +++++++++++++++ .../kotlin/pulsar.java-conventions.gradle.kts | 27 ++++++++++++++++--- tests/integration/build.gradle.kts | 2 ++ 3 files changed, 47 insertions(+), 3 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index eb09e79df9aec..bb42876d2a7b9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -131,6 +131,27 @@ 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. 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 diff --git a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts index 83d43ff78fef1..5ec724a8bb905 100644 --- a/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts +++ b/build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts @@ -205,6 +205,9 @@ val javaToolchains = extensions.getByType() // 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().configureEach { testJavaVersion.orNull?.let { version -> @@ -248,6 +251,27 @@ tasks.withType().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().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 diff --git a/tests/integration/build.gradle.kts b/tests/integration/build.gradle.kts index 290a61d49ce2a..c11b964e0e75b 100644 --- a/tests/integration/build.gradle.kts +++ b/tests/integration/build.gradle.kts @@ -259,6 +259,8 @@ tasks.register("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") From 64682b175ed4573cf3d0fc6963e99bf6c2af1e34 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 16 Sep 2026 17:32:44 +0300 Subject: [PATCH 2/4] [fix][ci] Restore Netty leak reports for integration and system tests Pass the selected leak-detection mode and dump directory to integration and system jobs, collect their leak reports, and preserve dumps as artifacts. Read container leak archives from the Gradle build directory instead of the obsolete Maven target directory. Validated workflow wiring and shell syntax, plus report and fail_on_leak exit statuses using both actual JVM leak dumps and container log archives. Assisted-by: Codex --- .github/workflows/pulsar-ci.yaml | 14 ++++++++++++++ pulsar-build/pulsar_ci_tool.sh | 4 ++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 9d16d3b0440c1..1ceb22c9eefa9 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.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 diff --git a/pulsar-build/pulsar_ci_tool.sh b/pulsar-build/pulsar_ci_tool.sh index 16a760f37e70c..d4edca1985460 100755 --- a/pulsar-build/pulsar_ci_tool.sh +++ b/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" From dc6aa90bee3576d077123936c49bff08bca133d5 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 16 Sep 2026 17:46:51 +0300 Subject: [PATCH 3/4] [fix][test] Collect Netty leak reports after container shutdown Collect Pulsar container logs after graceful service shutdown and before container removal so ExtendedNettyLeakDetector shutdown-hook reports are retained. Preserve cleanup if stopping or collecting logs fails. Add live Docker regressions for standalone and supervisor-managed JVMs, and run them in the standalone suite. Both reproduce missing shutdown reports before the fix and pass afterward. Add eight CI collector cases for host-only, container-only, combined, and clean runs in report and fail_on_leak modes. Validation: Docker test image build, scoped integration tests with configuration-cache reuse, Python collector regression tests, unit/integration/system workflow wiring checks, and quickCheck. Assisted-by: Codex --- .github/workflows/pulsar-ci.yaml | 3 + CONTRIBUTING.md | 3 +- pulsar-build/tests/test_netty_leak_reports.py | 76 ++++++++++ .../containers/NettyLeakDetectionTest.java | 130 ++++++++++++++++++ .../containers/PulsarContainer.java | 51 ++++--- .../src/test/resources/pulsar-standalone.xml | 1 + 6 files changed, 241 insertions(+), 23 deletions(-) create mode 100644 pulsar-build/tests/test_netty_leak_reports.py create mode 100644 tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/NettyLeakDetectionTest.java diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 1ceb22c9eefa9..7281bb5c9bdf6 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.github/workflows/pulsar-ci.yaml @@ -194,6 +194,9 @@ jobs: ./gradlew assemble rat spotlessCheck checkstyleMain checkstyleTest --no-configuration-cache + - name: Test Netty leak report collection + run: python3 pulsar-build/tests/test_netty_leak_reports.py + - name: Check binary licenses run: ./gradlew checkBinaryLicense --no-configuration-cache diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index bb42876d2a7b9..75d2bb3c41db4 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -139,7 +139,8 @@ to include test names in leak reports and write `netty_leak_*.txt` files. Set `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. For local tests, use `-PtestExitJvmOnLeak=true` to fail the test JVM on a detected leak: +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 \ diff --git a/pulsar-build/tests/test_netty_leak_reports.py b/pulsar-build/tests/test_netty_leak_reports.py new file mode 100644 index 0000000000000..a7dbbfd71084c --- /dev/null +++ b/pulsar-build/tests/test_netty_leak_reports.py @@ -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. +# + +"""Exercise the collector shared by unit, integration and system test jobs.""" + +import io +import os +from pathlib import Path +import subprocess +import tarfile +import tempfile +import unittest + + +COLLECTOR = Path(__file__).resolve().parents[1] / "pulsar_ci_tool.sh" + + +class NettyLeakReportsTest(unittest.TestCase): + def test_host_and_container_reports(self): + for mode in ("report", "fail_on_leak"): + for source in ("none", "host", "container", "both"): + with self.subTest(mode=mode, source=source), tempfile.TemporaryDirectory() as directory: + work = Path(directory) + (work / "pulsar-build").mkdir() + dumps = work / "dumps" + if source in ("host", "both"): + dumps.mkdir() + (dumps / "netty_leak_host.txt").write_text("host-jvm-leak\n") + if source in ("container", "both"): + logs = work / "tests/integration/build/container-logs/probe" + logs.mkdir(parents=True) + with tarfile.open(logs / "var-log-pulsar.tar.gz", "w:gz") as archive: + content = b"container-jvm-leak\n" + entry = tarfile.TarInfo("pulsar/netty_leak_container.txt") + entry.size = len(content) + archive.addfile(entry, io.BytesIO(content)) + env = dict(os.environ, NETTY_LEAK_DETECTION=mode, NETTY_LEAK_DUMP_DIR=str(dumps)) + result = subprocess.run( + ["bash", str(COLLECTOR), "report_netty_leaks"], + cwd=work, env=env, capture_output=True, text=True, timeout=30, + ) + expected_exit = int(mode == "fail_on_leak" and source != "none") + self.assertEqual(result.returncode, expected_exit, result.stdout + result.stderr) + marker = "netty_leaks_not_found" if source == "none" else "netty_leaks_found" + self.assertTrue((work / "pulsar-build" / marker).exists()) + if source == "none": + self.assertIn("No netty leaks found.", result.stdout) + continue + report = (dumps / "leak_report.txt").read_text() + annotation = "::error::" if mode == "fail_on_leak" else "::warning::" + self.assertIn(annotation + "Netty leaks found.", report) + if source in ("host", "both"): + self.assertIn("host-jvm-leak", report) + if source in ("container", "both"): + self.assertIn("container-jvm-leak", report) + self.assertTrue((dumps / "container-logs/probe/netty_leak_container.txt").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/NettyLeakDetectionTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/NettyLeakDetectionTest.java new file mode 100644 index 0000000000000..ece8aa9c620b0 --- /dev/null +++ b/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(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); + } + } +} diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java index 457222677d67a..d7ed36e37a9b7 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/PulsarContainer.java +++ b/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 diff --git a/tests/integration/src/test/resources/pulsar-standalone.xml b/tests/integration/src/test/resources/pulsar-standalone.xml index d8892c0746181..0a3c975e58be3 100644 --- a/tests/integration/src/test/resources/pulsar-standalone.xml +++ b/tests/integration/src/test/resources/pulsar-standalone.xml @@ -22,6 +22,7 @@ + From 10c3cf1bc29e6d27d6a6d075ed3c9e5778e00cec Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Wed, 16 Sep 2026 17:51:55 +0300 Subject: [PATCH 4/4] [test] Remove standalone Python leak collector checks Remove the Python regression script and its CI step to keep leak-detection support focused on the existing Gradle and Docker test infrastructure. Assisted-by: Codex --- .github/workflows/pulsar-ci.yaml | 3 - pulsar-build/tests/test_netty_leak_reports.py | 76 ------------------- 2 files changed, 79 deletions(-) delete mode 100644 pulsar-build/tests/test_netty_leak_reports.py diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 7281bb5c9bdf6..1ceb22c9eefa9 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.github/workflows/pulsar-ci.yaml @@ -194,9 +194,6 @@ jobs: ./gradlew assemble rat spotlessCheck checkstyleMain checkstyleTest --no-configuration-cache - - name: Test Netty leak report collection - run: python3 pulsar-build/tests/test_netty_leak_reports.py - - name: Check binary licenses run: ./gradlew checkBinaryLicense --no-configuration-cache diff --git a/pulsar-build/tests/test_netty_leak_reports.py b/pulsar-build/tests/test_netty_leak_reports.py deleted file mode 100644 index a7dbbfd71084c..0000000000000 --- a/pulsar-build/tests/test_netty_leak_reports.py +++ /dev/null @@ -1,76 +0,0 @@ -# -# 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. -# - -"""Exercise the collector shared by unit, integration and system test jobs.""" - -import io -import os -from pathlib import Path -import subprocess -import tarfile -import tempfile -import unittest - - -COLLECTOR = Path(__file__).resolve().parents[1] / "pulsar_ci_tool.sh" - - -class NettyLeakReportsTest(unittest.TestCase): - def test_host_and_container_reports(self): - for mode in ("report", "fail_on_leak"): - for source in ("none", "host", "container", "both"): - with self.subTest(mode=mode, source=source), tempfile.TemporaryDirectory() as directory: - work = Path(directory) - (work / "pulsar-build").mkdir() - dumps = work / "dumps" - if source in ("host", "both"): - dumps.mkdir() - (dumps / "netty_leak_host.txt").write_text("host-jvm-leak\n") - if source in ("container", "both"): - logs = work / "tests/integration/build/container-logs/probe" - logs.mkdir(parents=True) - with tarfile.open(logs / "var-log-pulsar.tar.gz", "w:gz") as archive: - content = b"container-jvm-leak\n" - entry = tarfile.TarInfo("pulsar/netty_leak_container.txt") - entry.size = len(content) - archive.addfile(entry, io.BytesIO(content)) - env = dict(os.environ, NETTY_LEAK_DETECTION=mode, NETTY_LEAK_DUMP_DIR=str(dumps)) - result = subprocess.run( - ["bash", str(COLLECTOR), "report_netty_leaks"], - cwd=work, env=env, capture_output=True, text=True, timeout=30, - ) - expected_exit = int(mode == "fail_on_leak" and source != "none") - self.assertEqual(result.returncode, expected_exit, result.stdout + result.stderr) - marker = "netty_leaks_not_found" if source == "none" else "netty_leaks_found" - self.assertTrue((work / "pulsar-build" / marker).exists()) - if source == "none": - self.assertIn("No netty leaks found.", result.stdout) - continue - report = (dumps / "leak_report.txt").read_text() - annotation = "::error::" if mode == "fail_on_leak" else "::warning::" - self.assertIn(annotation + "Netty leaks found.", report) - if source in ("host", "both"): - self.assertIn("host-jvm-leak", report) - if source in ("container", "both"): - self.assertIn("container-jvm-leak", report) - self.assertTrue((dumps / "container-logs/probe/netty_leak_container.txt").exists()) - - -if __name__ == "__main__": - unittest.main()