improve: allow flaky tests run with specify parameters

bookkeeper-common/src/test/java/org/apache/bookkeeper/common/testing/annotations/FlakyTest.java

@@ -19,14 +19,44 @@
 package org.apache.bookkeeper.common.testing.annotations;
 
 import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
 
 /**
- * Intended for marking a test case as flaky.
+ * Annotation to mark a test as flaky.
+ *
+ * <p>Flaky tests are tests that produce inconsistent results,
+ * occasionally failing or passing without changes to the code under test.
+ * This could be due to external factors such as timing issues, resource contention,
+ * dependency on non-deterministic data, or integration with external systems.
+ *
+ * <p>Tests marked with this annotation are excluded from execution
+ * in CI pipelines and Maven commands by default, to ensure a reliable and
+ * deterministic build process. However, they can still be executed manually
+ * or in specific environments for debugging and resolution purposes.
+ *
+ * <p>Usage:
+ * <pre>
+ * {@code
+ * @FlakyTest
+ * public void testSomething() {
+ *     // Test logic here
+ * }
+ * }
+ * </pre>
+ *
+ * <p>It is recommended to investigate and address the root causes of flaky tests
+ * rather than relying on this annotation long-term.
  */
 @Documented
-@Retention(RetentionPolicy.SOURCE)
+@Target({ ElementType.TYPE, ElementType.METHOD })
+@Retention(RetentionPolicy.RUNTIME)
+@Tag("flaky")
+@Test
 public @interface FlakyTest {
 
     /**

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieStorageThresholdTest.java

@@ -86,12 +86,12 @@ public void setUp() throws Exception {
 
     LedgerHandle[] prepareData(int numEntryLogs) throws Exception {
         // since an entry log file can hold at most 100 entries
-        // first ledger write 2 entries, which is less than low water mark
+        // first ledger write 2 entries, which is less than low watermark
         int num1 = 2;
-        // third ledger write more than high water mark entries
+        // third ledger write more than high watermark entries
         int num3 = (int) (NUM_ENTRIES * 0.7f);
-        // second ledger write remaining entries, which is higher than low water
-        // mark and less than high water mark
+        // second ledger write remaining entries, which is higher than low watermark
+        // and less than high watermark
         int num2 = NUM_ENTRIES - num3 - num1;
 
         LedgerHandle[] lhs = new LedgerHandle[3];
@@ -148,8 +148,8 @@ public void testStorageThresholdCompaction() throws Exception {
         File ledgerDir2 = tmpDirs.createNew("ledger", "test2");
         File journalDir = tmpDirs.createNew("journal", "test");
         String[] ledgerDirNames = new String[]{
-            ledgerDir1.getPath(),
-            ledgerDir2.getPath()
+                ledgerDir1.getPath(),
+                ledgerDir2.getPath()
         };
         conf.setLedgerDirNames(ledgerDirNames);
         conf.setJournalDirName(journalDir.getPath());
@@ -224,7 +224,7 @@ public void diskFull(File disk) {
         // there are no writableLedgerDirs
         for (File ledgerDir : bookie.getLedgerDirsManager().getAllLedgerDirs()) {
             assertFalse("Found entry log file ([0,1,2].log. They should have been compacted" + ledgerDir,
-                TestUtils.hasLogFiles(ledgerDir.getParentFile(), true, 0, 1, 2));
+                    TestUtils.hasLogFiles(ledgerDir.getParentFile(), true, 0, 1, 2));
         }
 
         try {

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperDiskSpaceWeightedLedgerPlacementTest.java

@@ -179,7 +179,7 @@ public void testDiskSpaceWeightedBookieSelection() throws Exception {
 
     /**
      * Test to show that weight based selection honors the disk weight of bookies and also adapts
-     * when the bookies's weight changes.
+     * when the bookies' weight changes.
      */
     @FlakyTest("https://github.com/apache/bookkeeper/issues/503")
     public void testDiskSpaceWeightedBookieSelectionWithChangingWeights() throws Exception {

pom.xml

@@ -918,6 +918,9 @@
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-surefire-plugin</artifactId>
           <version>${maven-surefire-plugin.version}</version>
+          <configuration>
+            <excludedGroups>flaky</excludedGroups>
+          </configuration>
         </plugin>
         <plugin>
           <groupId>org.apache.maven.plugins</groupId>

stream/distributedlog/core/src/test/java/org/apache/distributedlog/TestDistributedLogBase.java

@@ -61,6 +61,12 @@
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Rule;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
+import org.junit.rules.TestName;
 import org.junit.rules.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -72,6 +78,9 @@
 public class TestDistributedLogBase {
     static final Logger LOG = LoggerFactory.getLogger(TestDistributedLogBase.class);
 
+    @Rule
+    public final TestName runtime = new TestName();
+
     @Rule
     public Timeout globalTimeout = Timeout.seconds(120);
 
@@ -105,7 +114,20 @@ public class TestDistributedLogBase {
     protected static int zkPort;
     protected static final List<File> TMP_DIRS = new ArrayList<File>();
 
+    protected String testName;
+
+    @Before
+    public void setTestNameJunit4() {
+        testName = runtime.getMethodName();
+    }
+
+    @BeforeEach
+    void setTestNameJunit5(TestInfo testInfo) {
+        testName = testInfo.getDisplayName();
+    }
+
     @BeforeClass
+    @BeforeAll
     public static void setupCluster() throws Exception {
         setupCluster(numBookies);
     }
@@ -134,6 +156,7 @@ public void uncaughtException(Thread t, Throwable e) {
     }
 
     @AfterClass
+    @AfterAll
     public static void teardownCluster() throws Exception {
         bkutil.teardown();
         zks.stop();
@@ -143,6 +166,7 @@ public static void teardownCluster() throws Exception {
     }
 
     @Before
+    @BeforeEach
     public void setup() throws Exception {
         try {
             zkc = LocalDLMEmulator.connectZooKeeper("127.0.0.1", zkPort);
@@ -153,6 +177,7 @@ public void setup() throws Exception {
     }
 
     @After
+    @AfterEach
     public void teardown() throws Exception {
         if (null != zkc) {
             zkc.close();

stream/distributedlog/core/src/test/java/org/apache/distributedlog/bk/TestLedgerAllocator.java

@@ -18,16 +18,17 @@
 package org.apache.distributedlog.bk;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 
 import java.net.URI;
 import java.util.Enumeration;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.LedgerEntry;
@@ -53,17 +54,18 @@
 import org.apache.zookeeper.Op;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.Stat;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestName;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * TestLedgerAllocator.
  */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class TestLedgerAllocator extends TestDistributedLogBase {
 
     private static final Logger logger = LoggerFactory.getLogger(TestLedgerAllocator.class);
@@ -81,9 +83,6 @@ public void onAbort(Throwable t) {
         }
     };
 
-    @Rule
-    public TestName runtime = new TestName();
-
     private ZooKeeperClient zkc;
     private BookKeeperClient bkc;
     private DistributedLogConfiguration dlConf = new DistributedLogConfiguration();
@@ -92,7 +91,7 @@ private URI createURI(String path) {
         return URI.create("distributedlog://" + zkServers + path);
     }
 
-    @Before
+    @BeforeAll
     public void setup() throws Exception {
         zkc = TestZooKeeperClientBuilder.newBuilder()
                 .uri(createURI("/"))
@@ -102,7 +101,7 @@ public void setup() throws Exception {
                 .dlConfig(dlConf).ledgersPath(ledgersPath).zkc(zkc).build();
     }
 
-    @After
+    @AfterAll
     public void teardown() throws Exception {
         bkc.close();
         zkc.close();
@@ -164,7 +163,8 @@ public void testAllocation() throws Exception {
         Utils.close(allocator);
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testBadVersionOnTwoAllocators() throws Exception {
         String allocationPath = "/allocation-bad-version";
         zkc.get().create(allocationPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
@@ -173,9 +173,11 @@ public void testBadVersionOnTwoAllocators() throws Exception {
         Versioned<byte[]> allocationData = new Versioned<byte[]>(data, new LongVersion(stat.getVersion()));
 
         SimpleLedgerAllocator allocator1 =
-                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf), zkc, bkc);
+                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf),
+                        zkc, bkc);
         SimpleLedgerAllocator allocator2 =
-                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf), zkc, bkc);
+                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf),
+                        zkc, bkc);
         allocator1.allocate();
         // wait until allocated
         ZKTransaction txn1 = newTxn();
@@ -206,7 +208,8 @@ public void testBadVersionOnTwoAllocators() throws Exception {
         assertEquals(1, i);
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testAllocatorWithoutEnoughBookies() throws Exception {
         String allocationPath = "/allocator-without-enough-bookies";
 
@@ -230,17 +233,18 @@ public void testAllocatorWithoutEnoughBookies() throws Exception {
         assertEquals(0, data.length);
     }
 
-    @Test(timeout = 60000)
-    public void testSuccessAllocatorShouldDeleteUnusedledger() throws Exception {
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
+    public void testSuccessAllocatorShouldDeleteUnusedLedger() throws Exception {
         String allocationPath = "/allocation-delete-unused-ledger";
         zkc.get().create(allocationPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
         Stat stat = new Stat();
         byte[] data = zkc.get().getData(allocationPath, false, stat);
 
         Versioned<byte[]> allocationData = new Versioned<byte[]>(data, new LongVersion(stat.getVersion()));
 
-        SimpleLedgerAllocator allocator1 =
-                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf), zkc, bkc);
+        SimpleLedgerAllocator allocator1 = new SimpleLedgerAllocator(allocationPath, allocationData,
+                newQuorumConfigProvider(dlConf), zkc, bkc);
         allocator1.allocate();
         // wait until allocated
         ZKTransaction txn1 = newTxn();
@@ -250,8 +254,8 @@ public void testSuccessAllocatorShouldDeleteUnusedledger() throws Exception {
         stat = new Stat();
         data = zkc.get().getData(allocationPath, false, stat);
         allocationData = new Versioned<byte[]>(data, new LongVersion(stat.getVersion()));
-        SimpleLedgerAllocator allocator2 =
-                new SimpleLedgerAllocator(allocationPath, allocationData, newQuorumConfigProvider(dlConf), zkc, bkc);
+        SimpleLedgerAllocator allocator2 = new SimpleLedgerAllocator(allocationPath, allocationData,
+                newQuorumConfigProvider(dlConf), zkc, bkc);
         allocator2.allocate();
         // wait until allocated
         ZKTransaction txn2 = newTxn();
@@ -296,7 +300,8 @@ public void testSuccessAllocatorShouldDeleteUnusedledger() throws Exception {
         assertEquals(1, i);
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testCloseAllocatorDuringObtaining() throws Exception {
         String allocationPath = "/allocation2";
         SimpleLedgerAllocator allocator = createAllocator(allocationPath);
@@ -329,7 +334,8 @@ public void testCloseAllocatorAfterConfirm() throws Exception {
                 dlConf.getBKDigestPW().getBytes(UTF_8));
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testCloseAllocatorAfterAbort() throws Exception {
         String allocationPath = "/allocation3";
         SimpleLedgerAllocator allocator = createAllocator(allocationPath);
@@ -352,9 +358,10 @@ public void testCloseAllocatorAfterAbort() throws Exception {
                 dlConf.getBKDigestPW().getBytes(UTF_8));
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testConcurrentAllocation() throws Exception {
-        String allocationPath = "/" + runtime.getMethodName();
+        String allocationPath = "/" + testName;
         SimpleLedgerAllocator allocator = createAllocator(allocationPath);
         allocator.allocate();
         ZKTransaction txn1 = newTxn();
@@ -365,15 +372,17 @@ public void testConcurrentAllocation() throws Exception {
         assertTrue(obtainFuture2.isCompletedExceptionally());
         try {
             Utils.ioResult(obtainFuture2);
-            fail("Should fail the concurrent obtain since there is already a transaction obtaining the ledger handle");
+            fail("Should fail the concurrent obtain since there is "
+                    + "already a transaction obtaining the ledger handle");
         } catch (SimpleLedgerAllocator.ConcurrentObtainException cbe) {
             // expected
         }
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testObtainMultipleLedgers() throws Exception {
-        String allocationPath = "/" + runtime.getMethodName();
+        String allocationPath = "/" + testName;
         SimpleLedgerAllocator allocator = createAllocator(allocationPath);
         int numLedgers = 10;
         Set<LedgerHandle> allocatedLedgers = new HashSet<LedgerHandle>();
@@ -387,9 +396,10 @@ public void testObtainMultipleLedgers() throws Exception {
         assertEquals(numLedgers, allocatedLedgers.size());
     }
 
-    @Test(timeout = 60000)
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    @Test
     public void testAllocationWithMetadata() throws Exception {
-        String allocationPath = "/" + runtime.getMethodName();
+        String allocationPath = "/" + testName;
 
         String application = "testApplicationMetadata";
         String component = "testComponentMetadata";