[Misc] AsynchronousEventStoreTest#mailentity is flickering

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-api/src/main/java/org/xwiki/eventstream/internal/AbstractAsynchronousEventStore.java

@@ -435,7 +435,9 @@ private <O, I> void complete(EventStoreTask<O, I> task, O output)
             }
         }
 
-        // Notify Future listeners
+        // Notify Future listeners.
+        // We do so before the call to event listeners because callers do not need to wait for them before continuing,
+        // and instead should continue as soon as the output value is available.
         task.future.complete(output);
 
         // Notify event listeners

xwiki-platform-core/xwiki-platform-eventstream/xwiki-platform-eventstream-api/src/test/java/org/xwiki/eventstream/internal/AsynchronousEventStoreTest.java

@@ -25,18 +25,21 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 import org.xwiki.component.phase.InitializationException;
 import org.xwiki.eventstream.EntityEvent;
 import org.xwiki.eventstream.Event;
 import org.xwiki.eventstream.EventQuery;
 import org.xwiki.eventstream.EventSearchResult;
 import org.xwiki.eventstream.EventStatus;
 import org.xwiki.eventstream.EventStreamException;
+import org.xwiki.eventstream.events.AbstractEventStreamEvent;
 import org.xwiki.eventstream.events.EventStreamAddedEvent;
 import org.xwiki.eventstream.events.MailEntityAddedEvent;
 import org.xwiki.eventstream.events.MailEntityDeleteEvent;
@@ -58,7 +61,7 @@
 
 /**
  * Validate {@link AbstractAsynchronousEventStore}.
- * 
+ *
  * @version $Id$
  */
 @ComponentTest
@@ -394,7 +397,7 @@ void eventstatus() throws InterruptedException, ExecutionException, EventStreamE
     }
 
     @Test
-    void mailentity() throws InterruptedException, ExecutionException, EventStreamException
+    void mailentity() throws InterruptedException, ExecutionException
     {
         DefaultEvent event1 = event("id1");
         DefaultEvent event2 = event("id2");
@@ -403,13 +406,23 @@ void mailentity() throws InterruptedException, ExecutionException, EventStreamEx
         DefaultEntityEvent status21 = entityEvent(event2, "entity1");
         DefaultEntityEvent status24 = entityEvent(event2, "entity4");
 
+        CompletableFuture<Object> waitForObservationOnStatus24 =
+            createCompletableFutureForStatus(status24, MailEntityAddedEvent.class);
+        CompletableFuture<Object> waitForObservationOnStatus11 =
+            createCompletableFutureForStatus(status11, MailEntityDeleteEvent.class);
+
         this.store.saveEvent(event1);
         this.store.saveEvent(event2);
         this.store.saveMailEntityEvent(status11);
         this.store.saveMailEntityEvent(status13);
         this.store.saveMailEntityEvent(status21);
         this.store.saveMailEntityEvent(status24).get();
 
+        // Wait for the observation to be called on the last store call. The observation component is intentionally 
+        // called after the completable future returned by saveMailEntityEvent and saveEvent. Therefore, we need to 
+        // mock and wait explicitly for the test. 
+        waitForObservationOnStatus24.get();
+
         assertSame(status11, this.store.events.get(event1.getId()).mailstatuses.get(status11.getEntityId()));
         assertSame(status13, this.store.events.get(event1.getId()).mailstatuses.get(status13.getEntityId()));
         assertSame(status21, this.store.events.get(event2.getId()).mailstatuses.get(status21.getEntityId()));
@@ -422,11 +435,12 @@ void mailentity() throws InterruptedException, ExecutionException, EventStreamEx
         verify(this.observation).notify(any(MailEntityAddedEvent.class), eq(status21));
         verify(this.observation).notify(any(MailEntityAddedEvent.class), eq(status24));
 
-
         this.store.deleteMailEntityEvent(status11).get();
-        // Delete Mail Entity event needs a bit more checks because it's Optional so it might take a bit more time than
-        // other events: we give those few more milliseconds to avoid flickers
-        Thread.sleep(5);
+
+        // Wait for the observation to be called after deleteMailEntityEvent. The observation component is intentionally 
+        // called after the completable future returned by deleteMailEntityEventt. Therefore, we need to mock and wait
+        // explicitly for the test.
+        waitForObservationOnStatus11.get();
 
         assertNull(this.store.events.get(event1.getId()).mailstatuses.get(status11.getEntityId()));
         assertSame(status13, this.store.events.get(event1.getId()).mailstatuses.get(status13.getEntityId()));
@@ -453,4 +467,24 @@ void prefilter() throws InterruptedException, ExecutionException, EventStreamExc
         assertTrue(this.store.getEvent(event1.getId()).get().isPrefiltered());
         assertFalse(this.store.getEvent(event2.getId()).get().isPrefiltered());
     }
+
+    /**
+     * Observe for a call to notify on {@link #observation} for a given event and type. Complete the returned
+     * {@link CompletableFuture} as soon as notify is called. This allows for tests to wait for notify to be called even
+     * if not anticipated in the tested code.
+     *
+     * @param event the event to wait for
+     * @param type the type of the event
+     * @return a completable future to wait for before calling assertions
+     */
+    private CompletableFuture<Object> createCompletableFutureForStatus(DefaultEntityEvent event,
+        Class<? extends AbstractEventStreamEvent> type)
+    {
+        CompletableFuture<Object> waitForObservationOnStatus24 = new CompletableFuture<>();
+        Mockito.doAnswer(invocationOnMock -> {
+            waitForObservationOnStatus24.complete(true);
+            return null;
+        }).when(this.observation).notify(any(type), eq(event));
+        return waitForObservationOnStatus24;
+    }
 }