Fix race condition in scheduler engine test

This commit addresses a race condition in the scheduler engine test that
a listener that throws an exception does not cause other listeners to be
skipped. The race here is that we were counting down a latch, and then
throwing an exception yet an assertion that expected the exception to
have been thrown already could execute after the latch was counted down
for the final time but before the exception was thrown and acted upon by
the scheduler engine. This commit addresses this by moving the counting
down of the latch to definitely be after the exception was acted upon by
the scheduler engine.

Author: jasontedor

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngineTests.java

@@ -21,17 +21,19 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static org.hamcrest.Matchers.any;
 import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 public class SchedulerEngineTests extends ESTestCase {
 
-    @AwaitsFix(bugUrl="https://github.com/elastic/elasticsearch/issues/33124")
     public void testListenersThrowingExceptionsDoNotCauseOtherListenersToBeSkipped() throws InterruptedException {
         final Logger mockLogger = mock(Logger.class);
         final SchedulerEngine engine = new SchedulerEngine(Settings.EMPTY, Clock.systemUTC(), mockLogger);
@@ -40,6 +42,7 @@ public void testListenersThrowingExceptionsDoNotCauseOtherListenersToBeSkipped()
             final int numberOfListeners = randomIntBetween(1, 32);
             int numberOfFailingListeners = 0;
             final CountDownLatch latch = new CountDownLatch(numberOfListeners);
+
             for (int i = 0; i < numberOfListeners; i++) {
                 final AtomicBoolean trigger = new AtomicBoolean();
                 final SchedulerEngine.Listener listener;
@@ -55,12 +58,17 @@ public void testListenersThrowingExceptionsDoNotCauseOtherListenersToBeSkipped()
                     numberOfFailingListeners++;
                     listener = event -> {
                         if (trigger.compareAndSet(false, true)) {
-                            latch.countDown();
+                            // we count down the latch after this exception is caught and mock logged in SchedulerEngine#notifyListeners
                             throw new RuntimeException(getTestName());
                         } else {
                             fail("listener invoked twice");
                         }
                     };
+                    doAnswer(invocationOnMock -> {
+                        // this happens after the listener has been notified, threw an exception, and then mock logged the exception
+                        latch.countDown();
+                        return null;
+                    }).when(mockLogger).warn(argThat(any(ParameterizedMessage.class)), argThat(any(RuntimeException.class)));
                 }
                 listeners.add(Tuple.tuple(listener, trigger));
             }
@@ -135,7 +143,7 @@ public void testListenersThrowingExceptionsDoNotCauseNextScheduledTaskToBeSkippe
             listenersLatch.await();
             assertTrue(listeners.stream().map(Tuple::v2).allMatch(count -> count.get() == numberOfSchedules));
             latch.await();
-            assertFailedListenerLogMessage(mockLogger, numberOfListeners * numberOfSchedules);
+            assertFailedListenerLogMessage(mockLogger, numberOfSchedules * numberOfListeners);
             verifyNoMoreInteractions(mockLogger);
         } finally {
             engine.stop();