fix concurrency bug

Signed-off-by: Amit Galitzky <amgalitz@amazon.com>

Author: amitgalitz

CHANGELOG.md

@@ -10,6 +10,8 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 
 
 ### Bug Fixes
+- Fixing concurrency bug on writer ([#1508](https://github.com/opensearch-project/anomaly-detection/pull/1508))
+
 ### Infrastructure
 ### Documentation
 ### Maintenance

src/main/java/org/opensearch/ad/task/ADTaskManager.java

@@ -1928,7 +1928,7 @@ protected <T> void resetLatestConfigTaskState(
     protected String triageState(Boolean hasResult, String error, Long rcfTotalUpdates) {
         if (hasResult != null && hasResult) {
             return TaskState.RUNNING.name();
-        } else if (rcfTotalUpdates != null && rcfTotalUpdates < TimeSeriesSettings.NUM_MIN_SAMPLES) {
+        } else if (rcfTotalUpdates == null || rcfTotalUpdates < TimeSeriesSettings.NUM_MIN_SAMPLES) {
             return TaskState.INIT.name();
         } else {
             return TaskState.RUNNING.name();

src/test/java/org/opensearch/ad/ratelimit/ResultWriteWorkerTests.java

@@ -222,8 +222,9 @@ public void testConcurrentModificationException() throws InterruptedException {
         }
     }
 
-   private boolean runConcurrencyTest() throws InterruptedException {
-        final int numberOfThreads = 20; final CountDownLatch startLatch = new CountDownLatch(1);
+    private boolean runConcurrencyTest() throws InterruptedException {
+        final int numberOfThreads = 20;
+        final CountDownLatch startLatch = new CountDownLatch(1);
         final CountDownLatch endLatch = new CountDownLatch(numberOfThreads);
         final AtomicBoolean exceptionOccurred = new AtomicBoolean(false);
 
@@ -254,30 +255,30 @@ private boolean runConcurrencyTest() throws InterruptedException {
                     for (int j = 0; j < 30; j++) {
                         try {
                             String detectorId = "detector-" + (j % 3);
-                            resultWriteQueue.put(
+                            resultWriteQueue
+                                .put(
                                     new ADResultWriteRequest(
-                                            System.currentTimeMillis() + j,
-                                            detectorId,
-                                            RequestPriority.MEDIUM,
-                                            detectResult,
-                                            null,
-                                            null
+                                        System.currentTimeMillis() + j,
+                                        detectorId,
+                                        RequestPriority.MEDIUM,
+                                        detectResult,
+                                        null,
+                                        null
                                     )
-                            );
+                                );
                         } catch (ConcurrentModificationException e) {
                             exceptionOccurred.set(true);
                             return;
                         }
                     }
-                } catch (Exception ignored) {
-                } finally {
+                } catch (Exception ignored) {} finally {
                     endLatch.countDown();
                 }
             });
             thread.start();
         }
         startLatch.countDown();
         assertTrue("Test should complete", endLatch.await(30, TimeUnit.SECONDS));
-       return exceptionOccurred.get();
+        return exceptionOccurred.get();
     }
 }