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/CONTRIBUTING.md b/CONTRIBUTING.md index eb09e79df9aec..75d2bb3c41db4 100644 --- a/CONTRIBUTING.md +++ b/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 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/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" 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") 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 @@ +