Make MetadataCacheTest reliable. (#10877)

Avoid issue due to async update by using Awaitility Co-authored-by: Surinder Singh <surinders@splunk.com> (cherry picked from commit 4cf1008)
apache · Aug 25, 2021 · 9ea5180 · 9ea5180
1 parent 587548e
commit 9ea5180
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 13 deletions.
diff --git a/pulsar-metadata/pom.xml b/pulsar-metadata/pom.xml
@@ -72,6 +72,12 @@
       <groupId>com.github.ben-manes.caffeine</groupId>
       <artifactId>caffeine</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -95,4 +101,4 @@
       </plugin>
     </plugins>
   </build>
-</project>
+</project>
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataCacheTest.java
@@ -52,6 +52,7 @@
 import org.apache.pulsar.metadata.api.MetadataStoreException.NotFoundException;
 import org.apache.pulsar.metadata.api.MetadataStoreFactory;
 import org.apache.pulsar.metadata.api.Stat;
+import org.awaitility.Awaitility;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
@@ -140,25 +141,28 @@ private void multiStoreAddDelete(List<MetadataCache<MyClass>> caches, int addOn,
         MyClass value1 = new MyClass(testName, 1);
 
         addCache.create(key1, value1).join();
+
         // all time for changes to propagate to other caches
-        Thread.sleep(100);
-        for (MetadataCache<MyClass> cache: caches) {
-            if (cache == addCache) {
+        Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+            for (MetadataCache<MyClass> cache : caches) {
+                if (cache == addCache) {
+                    assertEquals(cache.getIfCached(key1), Optional.of(value1));
+                }
+                assertEquals(cache.get(key1).join(), Optional.of(value1));
                 assertEquals(cache.getIfCached(key1), Optional.of(value1));
             }
-            assertEquals(cache.get(key1).join(), Optional.of(value1));
-            assertEquals(cache.getIfCached(key1), Optional.of(value1));
-        }
+        });
 
         delCache.delete(key1).join();
 
         // all time for changes to propagate to other caches
-        Thread.sleep(100);
-        // The entry should get removed from all caches
-        for (MetadataCache<MyClass> cache: caches) {
-            assertEquals(cache.getIfCached(key1), Optional.empty());
-            assertEquals(cache.get(key1).join(), Optional.empty());
-        }
+        Awaitility.await().ignoreExceptions().untilAsserted(() -> {
+            // The entry should get removed from all caches
+            for (MetadataCache<MyClass> cache : caches) {
+                assertEquals(cache.getIfCached(key1), Optional.empty());
+                assertEquals(cache.get(key1).join(), Optional.empty());
+            }
+        });
     }
 
     @Test(dataProvider = "impl")