[improve][test] Move most flaky tests to flaky group

build/run_unit_group.sh

@@ -139,13 +139,21 @@ function print_testng_failures() {
 function test_group_broker_flaky() {
   echo "::endgroup::"
   echo "::group::Running quarantined tests"
-  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='quarantine' -DexcludedGroups='' -DfailIfNoTests=false \
+  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='quarantine' -DexcludedGroups='flaky' -DfailIfNoTests=false \
     -DtestForkCount=2 ||
     print_testng_failures pulsar-broker/target/surefire-reports/testng-failed.xml "Quarantined test failure in" "Quarantined test failures"
   echo "::endgroup::"
   echo "::group::Running flaky tests"
-  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='flaky' -DtestForkCount=2
+  mvn_test --no-fail-fast -pl pulsar-broker -Dgroups='flaky' -DexcludedGroups='quarantine' -DtestForkCount=2
   echo "::endgroup::"
+  local modules_with_flaky_tests=$(git grep -l '@Test.*"flaky"' | grep '/src/test/java/' | \
+    awk -F '/src/test/java/' '{ print $1 }' | grep -v -E 'pulsar-broker' | sort | uniq | \
+    perl -0777 -p -e 's/\n(\S)/,$1/g')
+  if [ -n "${modules_with_flaky_tests}" ]; then
+    echo "::group::Running flaky tests in modules '${modules_with_flaky_tests}'"
+    mvn_test --no-fail-fast -pl "${modules_with_flaky_tests}" -Dgroups='flaky' -DexcludedGroups='quarantine' -DfailIfNoTests=false
+    echo "::endgroup::"
+  fi
 }
 
 function test_group_proxy() {
@@ -179,7 +187,7 @@ function test_group_other() {
     perl -0777 -p -e 's/\n(\S)/,$1/g')
   if [ -n "${modules_with_quarantined_tests}" ]; then
     echo "::group::Running quarantined tests outside of pulsar-broker & pulsar-proxy (if any)"
-    mvn_test --no-fail-fast -pl "${modules_with_quarantined_tests}" test -Dgroups='quarantine' -DexcludedGroups='' \
+    mvn_test --no-fail-fast -pl "${modules_with_quarantined_tests}" test -Dgroups='quarantine' -DexcludedGroups='flaky' \
       -DfailIfNoTests=false || \
         echo "::warning::There were test failures in the 'quarantine' test group."
     echo "::endgroup::"

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

@@ -32,6 +32,10 @@ public class AnnotationListener implements IAnnotationTransformer {
     private static final long DEFAULT_TEST_TIMEOUT_MILLIS = TimeUnit.MINUTES.toMillis(5);
     private static final String OTHER_GROUP = "other";
 
+    private static final String FLAKY_GROUP = "flaky";
+
+    private static final String QUARANTINE_GROUP = "quarantine";
+
     public AnnotationListener() {
         System.out.println("Created annotation listener");
     }
@@ -51,9 +55,27 @@ public void transform(ITestAnnotation annotation,
             annotation.setTimeOut(DEFAULT_TEST_TIMEOUT_MILLIS);
         }
 
+        replaceGroupsIfFlakyOrQuarantineGroupIsIncluded(annotation);
         addToOtherGroupIfNoGroupsSpecified(annotation);
     }
 
+    // A test method will inherit the test groups from the class level and this solution ensures that a test method
+    // added to the flaky or quarantine group will not be executed as part of other groups.
+    private void replaceGroupsIfFlakyOrQuarantineGroupIsIncluded(ITestAnnotation annotation) {
+        if (annotation.getGroups() != null && annotation.getGroups().length > 1) {
+            for (String group : annotation.getGroups()) {
+                if (group.equals(QUARANTINE_GROUP)) {
+                    annotation.setGroups(new String[]{QUARANTINE_GROUP});
+                    return;
+                }
+                if (group.equals(FLAKY_GROUP)) {
+                    annotation.setGroups(new String[]{FLAKY_GROUP});
+                    return;
+                }
+            }
+        }
+    }
+
     private void addToOtherGroupIfNoGroupsSpecified(ITestOrConfiguration annotation) {
         // Add test to "other" group if there's no specified group
         if (annotation.getGroups() == null || annotation.getGroups().length == 0) {

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java

@@ -2446,7 +2446,7 @@ public void testRetentionSize() throws Exception {
         });
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testTimestampOnWorkingLedger() throws Exception {
         ManagedLedgerConfig conf = new ManagedLedgerConfig();
         conf.setMaxEntriesPerLedger(1);
@@ -3525,7 +3525,7 @@ public void testLedgerReachMaximumRolloverTime() throws Exception {
                 .until(() -> firstLedgerId != ml.addEntry("test".getBytes()).getLedgerId());
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testLedgerNotRolloverWithoutOpenState() throws Exception {
         ManagedLedgerConfig config = new ManagedLedgerConfig();
         config.setMaxEntriesPerLedger(2);

pom.xml

@@ -92,7 +92,7 @@ flexible messaging model and an intuitive client API.</description>
     <include>**/Test*.java,**/*Test.java,**/*Tests.java,**/*TestCase.java</include>
     <exclude/>
     <groups/>
-    <excludedGroups>quarantine</excludedGroups>
+    <excludedGroups>quarantine,flaky</excludedGroups>
 
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiMultiBrokersTest.java

@@ -132,7 +132,7 @@ public void testTopicLookup(TopicDomain topicDomain, boolean isPartition) throws
         Assert.assertEquals(lookupResultSet.size(), 1);
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testForceDeletePartitionedTopicWithSub() throws Exception {
         final int numPartitions = 10;
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAuthZTest.java

@@ -58,7 +58,7 @@ public class TopicAuthZTest extends MockedPulsarStandalone {
             .claim("sub", TENANT_ADMIN_SUBJECT).signWith(SECRET_KEY).compact();
 
     @SneakyThrows
-    @BeforeClass
+    @BeforeClass(alwaysRun = true)
     public void before() {
         configureTokenAuthentication();
         configureDefaultAuthorization();
@@ -78,7 +78,7 @@ public void before() {
 
 
     @SneakyThrows
-    @AfterClass
+    @AfterClass(alwaysRun = true)
     public void after() {
         if (superUserAdmin != null) {
             superUserAdmin.close();
@@ -988,7 +988,7 @@ public void testExamineMessage() {
         deleteTopic(topic, false);
     }
 
-    @Test(dataProvider = "partitioned")
+    @Test(dataProvider = "partitioned", groups = "flaky")
     @SneakyThrows
     public void testExpireMessage(boolean partitioned) {
         final String random = UUID.randomUUID().toString();

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

@@ -266,7 +266,7 @@ void testPersistentMessageFinderWhenLastMessageDelete() throws Exception {
         ledger.addEntry(createMessageWrittenToLedger("msg2"));
         ledger.addEntry(createMessageWrittenToLedger("msg3"));
         Position lastPosition = ledger.addEntry(createMessageWrittenToLedger("last-message"));
-        
+
         long endTimestamp = System.currentTimeMillis() + 1000;
 
         Result result = new Result();
@@ -383,7 +383,7 @@ public static Set<BrokerEntryMetadataInterceptor> getBrokerEntryMetadataIntercep
      *
      * @throws Exception
      */
-    @Test
+    @Test(groups = "flaky")
     void testMessageExpiryWithTimestampNonRecoverableException() throws Exception {
 
         final String ledgerAndCursorName = "testPersistentMessageExpiryWithNonRecoverableLedgers";
@@ -440,7 +440,7 @@ void testMessageExpiryWithTimestampNonRecoverableException() throws Exception {
 
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testIncorrectClientClock() throws Exception {
         final String ledgerAndCursorName = "testIncorrectClientClock";
         int maxTTLSeconds = 1;

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerConsumerInternalTest.java

@@ -120,7 +120,7 @@ private void removeServiceProducerMaintainedByServerCnx(ServiceProducer serviceP
         });
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testExclusiveConsumerWillAlwaysRetryEvenIfReceivedConsumerBusyError() throws Exception {
         final String topicName = BrokerTestUtil.newUniqueName("persistent://my-property/my-ns/tp_");
         final String subscriptionName = "subscription1";

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java

@@ -98,7 +98,7 @@ public class TransactionEndToEndTest extends TransactionTestBase {
     protected static final int NUM_PARTITIONS = 16;
     private static final int waitTimeForCanReceiveMsgInSec = 5;
     private static final int waitTimeForCannotReceiveMsgInSec = 5;
-    @BeforeClass
+    @BeforeClass(alwaysRun = true)
     protected void setup() throws Exception {
         conf.setAcknowledgmentAtBatchIndexLevelEnabled(true);
         setUpBase(1, NUM_PARTITIONS, TOPIC_OUTPUT, TOPIC_PARTITION);
@@ -1626,7 +1626,7 @@ public void testSendTxnAckBatchMessageToDLQ() throws Exception {
         admin.topics().delete(topic, true);
     }
 
-    @Test
+    @Test(groups = "flaky")
     public void testDelayedTransactionMessages() throws Exception {
         String topic = NAMESPACE1 + "/testDelayedTransactionMessages";
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndWithoutBatchIndexAckTest.java

@@ -30,7 +30,7 @@
 @Test(groups = "broker-impl")
 public class TransactionEndToEndWithoutBatchIndexAckTest extends TransactionEndToEndTest {
 
-    @BeforeClass
+    @BeforeClass(alwaysRun = true)
     protected void setup() throws Exception {
         conf.setAcknowledgmentAtBatchIndexLevelEnabled(false);
         setUpBase(1, NUM_PARTITIONS, TOPIC_OUTPUT, TOPIC_PARTITION);