Introduce retries for flaky tests (apache#1166)

* Introduce retries for flaky tests

* Avoid conflicts in allocating ports across different processes

* Removed duplicated builtools module after merge

* Also do retries on pulsar-discovery-service

buildtools/pom.xml

@@ -33,4 +33,11 @@
   <packaging>jar</packaging>
   <name>Pulsar Build Tools</name>
 
+  <dependencies>
+    <dependency>
+      <groupId>org.testng</groupId>
+      <artifactId>testng</artifactId>
+    </dependency>
+  </dependencies>
+
 </project>

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

@@ -0,0 +1,35 @@
+/**
+ * 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 java.lang.reflect.Constructor;
+import java.lang.reflect.Method;
+
+import org.testng.IAnnotationTransformer;
+import org.testng.annotations.ITestAnnotation;
+
+@SuppressWarnings("rawtypes")
+public class AnnotationListener implements IAnnotationTransformer {
+
+    @Override
+    public void transform(ITestAnnotation annotation, Class testClass, Constructor testConstructor, Method testMethod) {
+        annotation.setRetryAnalyzer(RetryAnalyzer.class);
+    }
+
+}

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

@@ -0,0 +1,36 @@
+/**
+ * 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.IRetryAnalyzer;
+import org.testng.ITestResult;
+
+public class RetryAnalyzer implements IRetryAnalyzer {
+
+    private int count = 0;
+
+    // Only try again once
+    private static final int MAX_RETRIES = 1;
+
+    @Override
+    public boolean retry(ITestResult result) {
+        return count++ < MAX_RETRIES;
+    }
+
+}

managed-ledger/pom.xml

@@ -98,10 +98,29 @@
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>buildtools</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <properties>
+            <property>
+              <name>listener</name>
+              <value>org.apache.pulsar.tests.AnnotationListener</value>
+            </property>
+          </properties>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-jar-plugin</artifactId>

managed-ledger/src/test/java/org/apache/bookkeeper/test/PortManager.java

@@ -18,37 +18,96 @@
  */
 package org.apache.bookkeeper.test;
 
-import java.net.ServerSocket;
+import java.io.FileReader;
+import java.io.FileWriter;
 import java.io.IOException;
+import java.net.ServerSocket;
+import java.nio.CharBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 
 /**
  * Port manager allows a base port to be specified on the commandline. Tests will then use ports, counting up from this
  * base port. This allows multiple instances of the bookkeeper tests to run at once.
  */
 public class PortManager {
-    private static int nextPort = getBasePort();
 
+    private static final String lockFilename = System.getProperty("test.lockFilename",
+            "/tmp/pulsar-test-port-manager.lock");
+    private static final int basePort = Integer.valueOf(System.getProperty("test.basePort", "15000"));
+
+    private static final int maxPort = 32000;
+
+    /**
+     * Return a TCP port that is currently unused.
+     *
+     * Keeps track of assigned ports and avoid race condition between different processes
+     */
     public synchronized static int nextFreePort() {
-        while (true) {
-            ServerSocket ss = null;
+        Path path = Paths.get(lockFilename);
+
+        try {
+            FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+            FileLock lock = fileChannel.lock();
+
             try {
-                int port = nextPort++;
-                ss = new ServerSocket(port);
-                ss.setReuseAddress(true);
-                return port;
-            } catch (IOException ioe) {
-            } finally {
-                if (ss != null) {
-                    try {
-                        ss.close();
-                    } catch (IOException ioe) {
-                    }
+
+                FileReader reader = new FileReader(lockFilename);
+                CharBuffer buffer = CharBuffer.allocate(16);
+                int len = reader.read(buffer);
+                buffer.flip();
+
+                int lastUsedPort = basePort;
+                if (len > 0) {
+                    String lastUsedPortStr = buffer.toString();
+                    lastUsedPort = Integer.parseInt(lastUsedPortStr);
                 }
+
+                int freePort = probeFreePort(lastUsedPort + 1);
+
+                FileWriter writer = new FileWriter(lockFilename);
+                writer.write(Integer.toString(freePort));
+
+                reader.close();
+                writer.close();
+
+                return freePort;
+
+            } finally {
+                lock.release();
+                fileChannel.close();
             }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
         }
     }
 
-    private static int getBasePort() {
-        return Integer.valueOf(System.getProperty("test.basePort", "15000"));
+    private static final int MAX_PORT_CONFLICTS = 10;
+
+    private synchronized static int probeFreePort(int port) {
+        int exceptionCount = 0;
+        while (true) {
+            if (port == maxPort) {
+                // Rollover the port probe
+                port = basePort;
+            }
+
+            try (ServerSocket ss = new ServerSocket(port)) {
+                ss.close();
+                // Give it some time to truly close the connection
+                Thread.sleep(100);
+                return port;
+
+            } catch (Exception e) {
+                port++;
+                exceptionCount++;
+                if (exceptionCount > MAX_PORT_CONFLICTS) {
+                    throw new RuntimeException(e);
+                }
+            }
+        }
     }
 }

pulsar-broker/pom.xml

@@ -224,10 +224,29 @@
       <groupId>com.ea.agentloader</groupId>
       <artifactId>ea-agent-loader</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>buildtools</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
     <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <properties>
+            <property>
+              <name>listener</name>
+              <value>org.apache.pulsar.tests.AnnotationListener</value>
+            </property>
+          </properties>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>aspectj-maven-plugin</artifactId>

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -86,7 +87,13 @@ void setup() throws Exception {
     @Override
     @AfterClass(timeOut = 30000)
     void shutdown() throws Exception {
-        super.shutdown();
+        ForkJoinPool.commonPool().execute(() -> {
+            try {
+                super.shutdown();
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        });
     }
 
     @Test(enabled = true, timeOut = 30000)

pulsar-discovery-service/pom.xml

@@ -135,6 +135,28 @@
       <type>test-jar</type>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>buildtools</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <properties>
+            <property>
+              <name>listener</name>
+              <value>org.apache.pulsar.tests.AnnotationListener</value>
+            </property>
+          </properties>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
 </project>