Handle race condition

bitrepository-alarm-service/src/main/java/org/bitrepository/alarm/store/AlarmDatabaseIngestor.java

@@ -29,6 +29,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -192,14 +193,16 @@ private Object[] extractArgumentsFromEvent(Alarm alarm) {
         return res.toArray();
     }
 
+    private static final String SQLSTATE_UNIQUE_VIOLATION = "23505";
+
     /**
      * Retrieve the guid for a given component. If the component does not exist within the component table,
      * then it is created.
      *
      * @param componentId The name of the alarm producing component.
      * @return The guid of the component with the given name.
      */
-    private synchronized long retrieveComponentGuid(String componentId) {
+    private long retrieveComponentGuid(String componentId) {
         String sqlRetrieve = "SELECT " + COMPONENT_GUID + " FROM " + COMPONENT_TABLE
                 + " WHERE " + COMPONENT_ID + " = ?";
 
@@ -208,12 +211,25 @@ private synchronized long retrieveComponentGuid(String componentId) {
         if (guid == null) {
             log.debug("Inserting component '{}' into the component table", componentId);
             String sqlInsert = "INSERT INTO " + COMPONENT_TABLE + " ( " + COMPONENT_ID + " ) VALUES ( ? )";
-            DatabaseUtils.executeStatement(dbConnector, sqlInsert, componentId);
+            try {
+                DatabaseUtils.executeStatement(dbConnector, sqlInsert, componentId);
+            } catch (IllegalStateException e) {
+                if (!isUniqueConstraintViolation(e)) {
+                    throw e;
+                }
+                log.debug("Component '{}' was concurrently inserted by another thread, using its guid instead",
+                        componentId);
+            }
 
             guid = DatabaseUtils.selectLongValue(dbConnector, sqlRetrieve, componentId);
         }
 
         assert guid != null;
         return guid;
     }
+
+    private boolean isUniqueConstraintViolation(IllegalStateException e) {
+        return e.getCause() instanceof SQLException sqlException
+                && SQLSTATE_UNIQUE_VIOLATION.equals(sqlException.getSQLState());
+    }
 }

bitrepository-alarm-service/src/test/java/org/bitrepository/alarm/store/AlarmDatabaseTest.java

@@ -2,7 +2,7 @@
  * #%L
  * Bitrepository Audit Trail Service
  * %%
- * Copyright (C) 2010 - 2012 The State and University Library, The Royal Library and The State Archives, Denmark
+ * Copyright (C) 2010 - 2059 The Royal Danish Library and The State Archives, Denmark
  * %%
  * This program is free software: you can redistribute it and/or modify
  * it under the terms of the GNU Lesser General Public License as
@@ -44,6 +44,13 @@
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static org.bitrepository.common.utils.AllureTestUtils.addDescription;
 import static org.bitrepository.common.utils.AllureTestUtils.addStep;
@@ -291,6 +298,70 @@ void alarmDatabaseCorrectTimestampTest() {
 
     }
 
+    @Test
+    @Tag(TestGroups.REGRESSIONTEST)
+    @Tag(TestGroups.DATABASETEST)
+    void concurrentAlarmsFromNewComponentAreAllStoredTest() throws InterruptedException {
+        addDescription("Testing that concurrent alarms from a not-before-seen component are all stored. "
+                + "Regression test for a race condition where two concurrent inserts of the same new component "
+                + "into the component table could cause one of them to fail with a unique constraint violation, "
+                + "losing an alarm.");
+
+        AlarmDAOFactory alarmDAOFactory = new AlarmDAOFactory();
+        AlarmServiceDAO database = alarmDAOFactory.getAlarmServiceDAOInstance(
+                settings.getReferenceSettings().getAlarmServiceSettings().getAlarmServiceDatabase());
+
+        String raceComponent = "RACE-COMPONENT-" + Instant.now().toEpochMilli();
+        int concurrentAlarms = 20;
+
+        addStep("Fire " + concurrentAlarms + " alarms from the same not-before-seen component at the same time.",
+                "None of the insertions should fail.");
+        ExecutorService executor = Executors.newFixedThreadPool(concurrentAlarms);
+        CountDownLatch readyLatch = new CountDownLatch(concurrentAlarms);
+        CountDownLatch startLatch = new CountDownLatch(1);
+        List<Future<?>> futures = new ArrayList<>();
+        for (int i = 0; i < concurrentAlarms; i++) {
+            int index = i;
+            futures.add(executor.submit(() -> {
+                Alarm alarm = new Alarm();
+                alarm.setAlarmCode(AlarmCode.COMPONENT_FAILURE);
+                alarm.setAlarmRaiser(raceComponent);
+                alarm.setAlarmText("Concurrent alarm #" + index);
+                alarm.setOrigDateTime(CalendarUtils.getNow());
+                readyLatch.countDown();
+                startLatch.await();
+                database.addAlarm(alarm);
+                return null;
+            }));
+        }
+        readyLatch.await();
+        startLatch.countDown();
+
+        addStep("Wait for all insertions to complete.", "None should throw an exception.");
+        List<Exception> failures = new ArrayList<>();
+        for (Future<?> future : futures) {
+            try {
+                future.get(30, TimeUnit.SECONDS);
+            } catch (ExecutionException | TimeoutException e) {
+                failures.add(e);
+            }
+        }
+        executor.shutdown();
+        Assertions.assertTrue(failures.isEmpty(), "Concurrent alarm ingestion threw: " + failures);
+
+        addStep("Verify every alarm was actually stored, and the component was only inserted once.",
+                "Should find all alarms and exactly one component row.");
+        List<Alarm> storedAlarms = database.extractAlarms(raceComponent, null, (Instant) null,
+                (Instant) null, null, null, null, false);
+        Assertions.assertEquals(concurrentAlarms, storedAlarms.size());
+
+        DBConnector connector = new DBConnector(settings.getReferenceSettings().getAlarmServiceSettings().getAlarmServiceDatabase());
+        Long componentCount = DatabaseUtils.selectLongValue(connector,
+                "SELECT COUNT(*) FROM " + AlarmDatabaseConstants.COMPONENT_TABLE + " WHERE "
+                        + AlarmDatabaseConstants.COMPONENT_ID + " = ?", raceComponent);
+        Assertions.assertEquals(1L, componentCount);
+    }
+
     private List<Alarm> makeAlarms() {
         List<Alarm> res = new ArrayList<>();