[Tests] Add TestNG listeners for resource cleanup, thread leak detection and "fail fast"

buildtools/pom.xml

@@ -37,28 +37,70 @@
   <properties>
     <maven.compiler.source>1.8</maven.compiler.source>
     <maven.compiler.target>1.8</maven.compiler.target>
+    <surefire.version>3.0.0-M3</surefire.version>
+    <log4j2.version>2.14.0</log4j2.version>
+    <slf4j.version>1.7.25</slf4j.version>
+    <testng.version>7.3.0</testng.version>
+    <commons-lang3.version>3.11</commons-lang3.version>
+    <maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
   </properties>
 
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-bom</artifactId>
+        <version>${log4j2.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
   <dependencies>
     <dependency>
       <groupId>org.testng</groupId>
       <artifactId>testng</artifactId>
-      <version>7.3.0</version>
+      <version>${testng.version}</version>
     </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-api</artifactId>
-      <version>2.14.0</version>
     </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-core</artifactId>
-      <version>2.14.0</version>
     </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>
-      <version>2.14.0</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-1.2-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>jcl-over-slf4j</artifactId>
+      <version>${slf4j.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>${commons-lang3.version}</version>
+      <exclusions>
+        <exclusion>
+          <groupId>commons-logging</groupId>
+          <artifactId>*</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <!-- for testing FastThreadLocalStateCleaner -->
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-common</artifactId>
+      <version>4.1.60.Final</version>
+      <scope>test</scope>
     </dependency>
   </dependencies>
 
@@ -79,6 +121,45 @@
           </mapping>
         </configuration>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>${maven-shade-plugin.version}</version>
+        <configuration>
+          <createDependencyReducedPom>true</createDependencyReducedPom>
+          <promoteTransitiveDependencies>true</promoteTransitiveDependencies>
+          <minimizeJar>false</minimizeJar>
+          <artifactSet>
+            <includes>
+              <include>org.apache.commons:commons-lang3</include>
+            </includes>
+          </artifactSet>
+          <relocations>
+            <relocation>
+              <pattern>org.apache.commons.lang3</pattern>
+              <shadedPattern>org.apache.pulsar.buildtools.shaded.org.apache.commons.lang3</shadedPattern>
+            </relocation>
+          </relocations>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.rat</groupId>
+        <artifactId>apache-rat-plugin</artifactId>
+        <configuration>
+          <excludes>
+            <!-- This is generated during maven build -->
+            <exclude>dependency-reduced-pom.xml</exclude>
+          </excludes>
+        </configuration>
+      </plugin>
     </plugins>
     <extensions>
       <extension>

buildtools/src/main/java/org/apache/pulsar/tests/AnnotationListener.java

@@ -24,6 +24,7 @@
 
 import org.testng.IAnnotationTransformer;
 import org.testng.annotations.ITestAnnotation;
+import org.testng.internal.annotations.DisabledRetryAnalyzer;
 
 public class AnnotationListener implements IAnnotationTransformer {
 
@@ -38,7 +39,9 @@ public void transform(ITestAnnotation annotation,
                           Class testClass,
                           Constructor testConstructor,
                           Method testMethod) {
-        annotation.setRetryAnalyzer(RetryAnalyzer.class);
+        if (annotation.getRetryAnalyzerClass() == DisabledRetryAnalyzer.class) {
+            annotation.setRetryAnalyzer(RetryAnalyzer.class);
+        }
 
         // Enforce default test timeout
         if (annotation.getTimeOut() == 0) {

buildtools/src/main/java/org/apache/pulsar/tests/BetweenTestClassesListenerAdapter.java

@@ -0,0 +1,60 @@
+/**
+ * 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;
+
+import org.testng.IClassListener;
+import org.testng.ITestClass;
+import org.testng.ITestContext;
+import org.testng.ITestListener;
+
+/**
+ * TestNG listener adapter for detecting when execution finishes in previous
+ * test class and starts in a new class.
+ */
+abstract class BetweenTestClassesListenerAdapter implements IClassListener, ITestListener {
+    Class<?> lastTestClass;
+
+    @Override
+    public void onBeforeClass(ITestClass testClass) {
+        checkIfTestClassChanged(testClass.getRealClass());
+    }
+
+    private void checkIfTestClassChanged(Class<?> testClazz) {
+        if (lastTestClass != testClazz) {
+            onBetweenTestClasses(lastTestClass, testClazz);
+            lastTestClass = testClazz;
+        }
+    }
+
+    @Override
+    public void onFinish(ITestContext context) {
+        if (lastTestClass != null) {
+            onBetweenTestClasses(lastTestClass, null);
+            lastTestClass = null;
+        }
+    }
+
+    /**
+     * Call back hook for adding logic when test execution moves from test class to another.
+     *
+     * @param endedTestClass the test class which has finished execution. null if the started test class is the first
+     * @param startedTestClass the test class which has started execution. null if the ended test class is the last
+     */
+    protected abstract void onBetweenTestClasses(Class<?> endedTestClass, Class<?> startedTestClass);
+}

buildtools/src/main/java/org/apache/pulsar/tests/FailFastNotifier.java

@@ -0,0 +1,85 @@
+/**
+ * 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;
+
+import org.testng.IInvokedMethod;
+import org.testng.IInvokedMethodListener;
+import org.testng.ITestResult;
+import org.testng.SkipException;
+
+/**
+ * Notifies TestNG core skipping remaining tests after first failure has appeared.
+ *
+ * Enabled when -DtestFailFast=true
+ *
+ * This is a workaround for https://issues.apache.org/jira/browse/SUREFIRE-1762 since
+ * the bug makes the built-in fast-fast feature `-Dsurefire.skipAfterFailureCount=1` unusable.
+ * Maven Surefire version 3.0.0-M5 contains the fix, but that version is unusable because of problems
+ * with test output, https://issues.apache.org/jira/browse/SUREFIRE-1827.
+ * It makes the Pulsar integration tests slow and to fail.
+ *
+ * This implementation is based on org.apache.maven.surefire.testng.utils.FailFastNotifier
+ * implementation that is part of the Maven Surefire plugin.
+ *
+ */
+public class FailFastNotifier
+        implements IInvokedMethodListener {
+    private static final boolean FAIL_FAST_ENABLED = Boolean.valueOf(
+            System.getProperty("testFailFast", "true"));
+
+    static class FailFastEventsSingleton {
+        private static final FailFastEventsSingleton INSTANCE = new FailFastEventsSingleton();
+
+        private volatile boolean skipAfterFailure;
+
+        private FailFastEventsSingleton() {
+        }
+
+        public static FailFastEventsSingleton getInstance() {
+            return INSTANCE;
+        }
+
+        public boolean isSkipAfterFailure() {
+            return skipAfterFailure;
+        }
+
+        public void setSkipOnNextTest() {
+            this.skipAfterFailure = true;
+        }
+    }
+
+    static class FailFastSkipException extends SkipException {
+        FailFastSkipException(String skipMessage) {
+            super(skipMessage);
+            reduceStackTrace();
+        }
+    }
+
+    @Override
+    public void beforeInvocation(IInvokedMethod iInvokedMethod, ITestResult iTestResult) {
+        if (FAIL_FAST_ENABLED && FailFastEventsSingleton.getInstance().isSkipAfterFailure()) {
+            throw new FailFastSkipException("Skipped after failure since testFailFast system property is set.");
+        }
+    }
+
+    @Override
+    public void afterInvocation(IInvokedMethod iInvokedMethod, ITestResult iTestResult) {
+
+    }
+}

buildtools/src/main/java/org/apache/pulsar/tests/FastThreadLocalCleanupListener.java

@@ -0,0 +1,61 @@
+/**
+ * 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;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Cleanup Thread Local state attach to Netty's FastThreadLocal.
+ */
+public class FastThreadLocalCleanupListener extends BetweenTestClassesListenerAdapter {
+    private static final Logger LOG = LoggerFactory.getLogger(FastThreadLocalCleanupListener.class);
+    private static final boolean FAST_THREAD_LOCAL_CLEANUP_ENABLED =
+            Boolean.valueOf(System.getProperty("testFastThreadLocalCleanup", "true"));
+    private static final String FAST_THREAD_LOCAL_CLEANUP_PACKAGE =
+            System.getProperty("testFastThreadLocalCleanupPackage", "org.apache.pulsar");
+    private static final FastThreadLocalStateCleaner CLEANER = new FastThreadLocalStateCleaner(object -> {
+        if ("*".equals(FAST_THREAD_LOCAL_CLEANUP_PACKAGE)) {
+            return true;
+        }
+        Class<?> clazz = object.getClass();
+        if (clazz.isArray()) {
+            clazz = clazz.getComponentType();
+        }
+        Package pkg = clazz.getPackage();
+        if (pkg != null && pkg.getName() != null) {
+            return pkg.getName()
+                    .startsWith(FAST_THREAD_LOCAL_CLEANUP_PACKAGE);
+        } else {
+            return false;
+        }
+    });
+
+    @Override
+    protected void onBetweenTestClasses(Class<?> endedTestClass, Class<?> startedTestClass) {
+        if (FAST_THREAD_LOCAL_CLEANUP_ENABLED && FastThreadLocalStateCleaner.isEnabled()) {
+            LOG.info("Cleaning up FastThreadLocal thread local state.");
+            CLEANER.cleanupAllFastThreadLocals((thread, value) -> {
+                LOG.info("Cleaning FastThreadLocal state for thread {}, instance of class {}, value is {}", thread,
+                        value.getClass().getName(), value);
+            });
+        }
+    }
+
+}