From 9fddf7e6a3ce13597d21c61d9818e301016b7b03 Mon Sep 17 00:00:00 2001
From: Jason Tedor <jason@tedor.me>
Date: Fri, 24 Aug 2018 07:45:16 -0400
Subject: [PATCH] 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.
---
 .../xpack/core/scheduler/SchedulerEngineTests.java | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngineTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngineTests.java
index 0f98acefe5b..5ab7b805cc1 100644
--- a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngineTests.java
+++ b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/scheduler/SchedulerEngineTests.java
@@ -21,9 +21,12 @@ import java.util.concurrent.CountDownLatch;
 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;
@@ -31,7 +34,6 @@ 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 class SchedulerEngineTests extends ESTestCase {
             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 class SchedulerEngineTests extends ESTestCase {
                     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 class SchedulerEngineTests extends ESTestCase {
             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();