diff --git a/bitrepository-alarm-service/src/main/java/org/bitrepository/alarm/store/AlarmDatabaseIngestor.java b/bitrepository-alarm-service/src/main/java/org/bitrepository/alarm/store/AlarmDatabaseIngestor.java index 4b24e2192..ee6daaecd 100644 --- a/bitrepository-alarm-service/src/main/java/org/bitrepository/alarm/store/AlarmDatabaseIngestor.java +++ b/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,6 +193,8 @@ 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. @@ -199,7 +202,7 @@ private Object[] extractArgumentsFromEvent(Alarm alarm) { * @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,7 +211,15 @@ 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); } @@ -216,4 +227,9 @@ private synchronized long retrieveComponentGuid(String componentId) { assert guid != null; return guid; } + + private boolean isUniqueConstraintViolation(IllegalStateException e) { + return e.getCause() instanceof SQLException sqlException + && SQLSTATE_UNIQUE_VIOLATION.equals(sqlException.getSQLState()); + } } diff --git a/bitrepository-alarm-service/src/test/java/org/bitrepository/alarm/store/AlarmDatabaseTest.java b/bitrepository-alarm-service/src/test/java/org/bitrepository/alarm/store/AlarmDatabaseTest.java index 5a87a6412..ea920187c 100644 --- a/bitrepository-alarm-service/src/test/java/org/bitrepository/alarm/store/AlarmDatabaseTest.java +++ b/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> 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 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 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 makeAlarms() { List res = new ArrayList<>();