[FSSDK-9054] fix(odp): flush odp events instantly upon calling close (#511)

* corrected flush all test and added shutdown event instead of shutdown boolean.

* Added check

---------

Co-authored-by: NomanShoaib <m.nomanshoaib09@gmail.com>

Author: mnoman09

core-api/src/main/java/com/optimizely/ab/odp/ODPEventManager.java

@@ -35,6 +35,7 @@ public class ODPEventManager {
     private static final int DEFAULT_FLUSH_INTERVAL = 1000;
     private static final int MAX_RETRIES = 3;
     private static final String EVENT_URL_PATH = "/v3/events";
+    private static final Object SHUTDOWN_SIGNAL = new Object();
 
     private final int queueSize;
     private final int batchSize;
@@ -188,8 +189,6 @@ public void stop() {
 
     private class EventDispatcherThread extends Thread {
 
-        private volatile boolean shouldStop = false;
-
         private final List<ODPEvent> currentBatch = new ArrayList<>();
 
         private long nextFlushTime = new Date().getTime();
@@ -198,7 +197,7 @@ private class EventDispatcherThread extends Thread {
         public void run() {
             while (true) {
                 try {
-                    Object nextEvent;
+                    Object nextEvent = null;
 
                     // If batch has events, set the timeout to remaining time for flush interval,
                     //      otherwise wait for the new event indefinitely
@@ -213,9 +212,6 @@ public void run() {
                         if (!currentBatch.isEmpty()) {
                             flush();
                         }
-                        if (shouldStop) {
-                            break;
-                        }
                         continue;
                     }
 
@@ -228,7 +224,11 @@ public void run() {
                         // Batch starting, create a new flush time
                         nextFlushTime = new Date().getTime() + flushInterval;
                     }
-
+                    if (nextEvent == SHUTDOWN_SIGNAL) {
+                        flush();
+                        logger.info("Received shutdown signal.");
+                        break;
+                    }
                     currentBatch.add((ODPEvent) nextEvent);
 
                     if (currentBatch.size() >= batchSize) {
@@ -268,7 +268,9 @@ private void flush() {
         }
 
         public void signalStop() {
-            shouldStop = true;
+            if (!eventQueue.offer(SHUTDOWN_SIGNAL)) {
+                logger.error("Failed to Process Shutdown odp Event. Event Queue is not accepting any more events");
+            }
         }
     }
 

core-api/src/test/java/com/optimizely/ab/odp/ODPEventManagerTest.java

@@ -182,18 +182,19 @@ public void retryFailedEvents() throws InterruptedException {
 
     @Test
     public void shouldFlushAllScheduledEventsBeforeStopping() throws InterruptedException {
+        int flushInterval = 20000;
         Mockito.reset(mockApiManager);
         Mockito.when(mockApiManager.sendEvents(any(), any(), any())).thenReturn(202);
         ODPConfig odpConfig = new ODPConfig("key", "http://www.odp-host.com", null);
-        ODPEventManager eventManager = new ODPEventManager(mockApiManager);
+        ODPEventManager eventManager = new ODPEventManager(mockApiManager, null, flushInterval);
         eventManager.updateSettings(odpConfig);
         eventManager.start();
-        for (int i = 0; i < 25; i++) {
+        for (int i = 0; i < 8; i++) {
             eventManager.sendEvent(getEvent(i));
         }
         eventManager.stop();
         Thread.sleep(1500);
-        Mockito.verify(mockApiManager, times(3)).sendEvents(eq("key"), eq("http://www.odp-host.com/v3/events"), any());
+        Mockito.verify(mockApiManager, times(1)).sendEvents(eq("key"), eq("http://www.odp-host.com/v3/events"), any());
         logbackVerifier.expectMessage(Level.DEBUG, "Exiting ODP Event Dispatcher Thread.");
     }