improve: allow flaky tests run with specify parameters (#4543)

Fix #3249 

### Motivation

This PR addresses the issue where tests annotated with @FlakyTest were not being executed by JUnit unless they were also annotated with @test.

To resolve this, we have updated our Maven Surefire plugin configuration to exclude tests tagged with flaky during regular CI runs, allowing these tests to be run separately.

This adjustment ensures that all tests, including those marked as flaky, are executed as intended unless explicitly excluded.(In CI still don't run it) 

Moreover, I've planned the addition of a daily CI job and non-required CI to specifically run tests tagged as flaky, ensuring continuous monitoring and quicker identification of intermittent issues without affecting the main test pipeline. This setup also enhances the ability to run tests directly from the IDE, making it more convenient for developers to execute and debug individual tests as needed.

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 {

bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/BookieAutoRecoveryTest.java

@@ -375,7 +375,7 @@ public void testNoSuchLedgerExists() throws Exception {
     }
 
     /**
-     * Test that if a empty ledger loses the bookie not in the quorum for entry 0, it will
+     * Test that if an empty ledger loses the bookie not in the quorum for entry 0, it will
      * still be openable when it loses enough bookies to lose a whole quorum.
      */
     @Test

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";