(Flaky-tests) Reduce flakiness of tests

.github/workflows/ci-unit.yaml

@@ -64,7 +64,7 @@ jobs:
         run: mvn install -DfailIfNoTests=false '-Dtest=!KafkaProducerSimpleConsumerTest,!PrimitiveSchemaTest,!BlobStoreManagedLedgerOffloaderTest' -pl '!pulsar-broker,!pulsar-proxy,!pulsar-broker-auth-sasl,!pulsar-io/kafka-connect-adaptor,!tests/pulsar-storm-test'
 
       - name: package surefire artifacts
-        if: failure()
+        if: always()
         run: |
           rm -rf artifacts
           mkdir artifacts
@@ -73,7 +73,7 @@ jobs:
 
       - uses: actions/upload-artifact@master
         name: upload surefire-artifacts
-        if: failure()
+        if: always()
         with:
           name: surefire-artifacts
           path: artifacts.zip

pom.xml

@@ -214,6 +214,7 @@ flexible messaging model and an intuitive client API.</description>
     <javassist.version>3.25.0-GA</javassist.version>
     <failsafe.version>2.3.1</failsafe.version>
     <skyscreamer.version>1.5.0</skyscreamer.version>
+    <awaitility.version>4.0.2</awaitility.version>
 
     <!-- Plugin dependencies -->
     <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
@@ -251,6 +252,13 @@ flexible messaging model and an intuitive client API.</description>
         <version>${testng.version}</version>
       </dependency>
 
+      <dependency>
+        <groupId>org.awaitility</groupId>
+        <artifactId>awaitility</artifactId>
+        <version>${awaitility.version}</version>
+        <scope>test</scope>
+      </dependency>
+
       <dependency>
         <groupId>org.mockito</groupId>
         <artifactId>mockito-core</artifactId>

pulsar-broker/pom.xml

@@ -34,6 +34,11 @@
   <name>Pulsar Broker</name>
 
   <dependencies>
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>commons-codec</groupId>
       <artifactId>commons-codec</artifactId>

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -140,7 +140,7 @@ public class ServerCnx extends PulsarHandler {
     AuthenticationState originalAuthState;
     private boolean pendingAuthChallengeResponse = false;
 
-    // Max number of pending requests per connections. If multiple producers are sharing the same connection the flow
+    // Max number of pending requests per connections. If multiple producers are sharing the same connection, the flow
     // control done by a single producer might not be enough to prevent write spikes on the broker.
     private final int maxPendingSendRequests;
     private final int resumeReadsThreshold;
@@ -704,6 +704,7 @@ protected void handleAuthResponse(CommandAuthResponse authResponse) {
 
     @Override
     protected void handleSubscribe(final CommandSubscribe subscribe) {
+        log.info("handleSubscribe is getting called on ServerCnx");
         checkArgument(state == State.Connected);
         final long requestId = subscribe.getRequestId();
         final long consumerId = subscribe.getConsumerId();

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

@@ -206,7 +206,7 @@ public void clusters() throws Exception {
         admin.clusters().createCluster("usw",
                 new ClusterData("http://broker.messaging.use.example.com:8080"));
         // "test" cluster is part of config-default cluster and it's znode gets created when PulsarService creates
-        // failure-domain znode of this default cluster
+        // failure-domain znode of this default cluster.
         assertEquals(admin.clusters().getClusters(), Lists.newArrayList("test", "usw"));
 
         assertEquals(admin.clusters().getCluster("test"),

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

@@ -436,7 +436,7 @@ private void unloadTopic(String topicName, boolean isPersistentTopic) throws Exc
      * @param namespaceName
      * @throws Exception
      */
-    @Test(dataProvider = "namespaceNames", timeOut = 10000)
+    @Test(dataProvider = "namespaceNames", timeOut = 30000)
     public void testResetCursorOnPosition(String namespaceName) throws Exception {
         final String topicName = "persistent://prop-xyz/use/" + namespaceName + "/resetPosition";
         final int totalProducedMessages = 50;
@@ -459,7 +459,7 @@ public void testResetCursorOnPosition(String namespaceName) throws Exception {
         MessageIdImpl resetMessageId = null;
         int resetPositionId = 10;
         for (int i = 0; i < 20; i++) {
-            message = consumer.receive(1, TimeUnit.SECONDS);
+            message = consumer.receive(5, TimeUnit.SECONDS);
             consumer.acknowledge(message);
             if (i == resetPositionId) {
                 resetMessageId = (MessageIdImpl) message.getMessageId();
@@ -959,7 +959,7 @@ public void testCreateNamespaceWithNoClusters() throws PulsarAdminException {
                 Collections.singletonList(localCluster));
     }
 
-    @Test(timeOut = 30000)
+    @Test(timeOut = 90000)
     public void testConsumerStatsLastTimestamp() throws PulsarClientException, PulsarAdminException, InterruptedException {
         long timestamp = System.currentTimeMillis();
         final String topicName = "consumer-stats-" + timestamp;

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v1/V1_AdminApiTest2.java

@@ -449,7 +449,7 @@ public void testResetCursorOnPosition(String namespaceName) throws Exception {
         MessageIdImpl resetMessageId = null;
         int resetPositionId = 10;
         for (int i = 0; i < 20; i++) {
-            message = consumer.receive(1, TimeUnit.SECONDS);
+            message = consumer.receive(5, TimeUnit.SECONDS);
             consumer.acknowledge(message);
             if (i == resetPositionId) {
                 resetMessageId = (MessageIdImpl) message.getMessageId();

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

@@ -463,7 +463,7 @@ public void testSimpleBatchProducerConsumer1kMessages(BatcherBuilder builder) th
 
         Message<byte[]> lastunackedMsg = null;
         for (int i = 0; i < numMsgs; i++) {
-            Message<byte[]> msg = consumer.receive(1, TimeUnit.SECONDS);
+            Message<byte[]> msg = consumer.receive(5, TimeUnit.SECONDS);
             assertNotNull(msg);
             lastunackedMsg = msg;
         }
@@ -708,7 +708,7 @@ public void testConcurrentBatchMessageAck(BatcherBuilder builder) throws Excepti
         for (int i = 0; i < numMsgs; i++) {
             executor.submit(() -> {
                 try {
-                    Message<byte[]> msg = myConsumer.receive(1, TimeUnit.SECONDS);
+                    Message<byte[]> msg = myConsumer.receive(5, TimeUnit.SECONDS);
                     myConsumer.acknowledge(msg);
                 } catch (Exception e) {
                     failed.set(false);

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

@@ -96,7 +96,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
         }
         Message<byte[]> msg = null;
         for (int i = 0; i < 10; i++) {
-            msg = consumer.receive(1, TimeUnit.SECONDS);
+            msg = consumer.receive(5, TimeUnit.SECONDS);
             consumer.acknowledge(msg);
         }
 
@@ -130,7 +130,7 @@ public void testCrashBrokerWithoutCursorLedgerLeak() throws Exception {
             producer.send(message.getBytes());
         }
         for (int i = 0; i < 10; i++) {
-            msg = consumer.receive(1, TimeUnit.SECONDS);
+            msg = consumer.receive(5, TimeUnit.SECONDS);
             consumer.acknowledge(msg);
         }
 
@@ -215,7 +215,7 @@ public void testSkipCorruptDataLedger() throws Exception {
         }
 
         // validate: consumer is able to consume msg and close consumer after reading 1 entry
-        Assert.assertNotNull(consumer.receive(1, TimeUnit.SECONDS));
+        Assert.assertNotNull(consumer.receive(5, TimeUnit.SECONDS)); // Need to replace with await()
         consumer.close();
 
         NavigableMap<Long, LedgerInfo> ledgerInfo = ml.getLedgersInfo();
@@ -249,14 +249,14 @@ public void testSkipCorruptDataLedger() throws Exception {
         Message<byte[]> msg = null;
         // start consuming message
         consumer = client.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe();
-        msg = consumer.receive(1, TimeUnit.SECONDS);
+        msg = consumer.receive(10, TimeUnit.MILLISECONDS);
         Assert.assertNull(msg);
         consumer.close();
 
         // (4) enable dynamic config to skip non-recoverable data-ledgers
         admin.brokers().updateDynamicConfiguration("autoSkipNonRecoverableData", "true");
 
-        retryStrategically((test) -> config.isAutoSkipNonRecoverableData(), 5, 100);
+        retryStrategically((test) -> config.isAutoSkipNonRecoverableData(), 15, 300);
 
         // (5) consumer will be able to consume 20 messages from last non-deleted ledger
         consumer = client.newConsumer().topic(topic1).subscriptionName("my-subscriber-name").subscribe();

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

@@ -246,7 +246,7 @@ public void testBookieIsolation() throws Exception {
      * @throws Exception
      */
     @Test
-    public void testBookieIsilationWithSecondaryGroup() throws Exception {
+    public void testBookieIsolationWithSecondaryGroup() throws Exception {
         final String tenant1 = "tenant1";
         final String cluster = "use";
         final String ns1 = String.format("%s/%s/%s", tenant1, cluster, "ns1");

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

@@ -30,7 +30,7 @@
 /**
  */
 public abstract class BrokerTestBase extends MockedPulsarServiceBaseTest {
-    protected static final int ASYNC_EVENT_COMPLETION_WAIT = 100;
+    protected static final int ASYNC_EVENT_COMPLETION_WAIT = 500;
 
     protected PulsarService getPulsar() {
         return pulsar;

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

@@ -27,6 +27,8 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  */
 @Test

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

@@ -48,13 +48,13 @@
 public class PeerReplicatorTest extends ReplicatorTestBase {
 
     @Override
-    @BeforeClass(timeOut = 300000)
+    @BeforeClass(timeOut = 900000)
     void setup() throws Exception {
         super.setup();
     }
 
     @Override
-    @AfterClass(timeOut = 300000)
+    @AfterClass(timeOut = 900000)
     void shutdown() throws Exception {
         super.shutdown();
     }