diff --git a/README.md b/README.md index 03cf404ec..dfa577e09 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,22 @@ This setting does not enable diagnostic `@debug` result sets. See [`documentation/EventProcedureJobFlowLogging.md`](documentation/EventProcedureJobFlowLogging.md) for operational guidance and stored-procedure coverage behavior. +### Scheduled cleanup coordination + +The scheduled event-metric and LAB100 cleanup procedures use separate SQL +Server application locks to prevent duplicate execution by multiple service +pods. A pod that cannot acquire its procedure's exclusive session lock returns +immediately and logs `Skipped because it's already running`; the +next cron occurrence retries it. A successful execution logs `Stored proc +execution completed: `. A `-1` procedure result raises a scheduler +error rather than a completion log. + +The lock resources are `RTR:scheduled:event-metric-cleanup` and +`RTR:scheduled:lab100-cleanup`. They use zero timeout, so contention does not +block the scheduler, and the resources are independent, so the two cleanup +jobs can run concurrently. This is duplicate-execution coordination only, not +a distributed work queue or post-processing-cache coordinator. + ### Windows Note (Testcontainers EOF Error) If you see `unexpected EOF` while running containerized tests on Windows, this is usually caused by Testcontainers trying to copy the entire project directory into a container and hitting a locked file. diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/actrelationship/ActRelationshipProcessor.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/actrelationship/ActRelationshipProcessor.java index 1d33dc5d3..8b356cd0a 100644 --- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/actrelationship/ActRelationshipProcessor.java +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/actrelationship/ActRelationshipProcessor.java @@ -49,7 +49,7 @@ public void process(String message, long batchId) { // extract uid and relationship type from message sourceActUid = getSourceActUid(message, operation); String typeCd = getTypeCd(message, operation); - + // call the relevant handler based on the relationship type if (isVaccinationRelationship(typeCd)) { investigationService.processVaccination(message, false, sourceActUid); diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/repository/PostProcRepository.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/repository/PostProcRepository.java index f41cd7e36..a681e9ceb 100644 --- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/repository/PostProcRepository.java +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/repository/PostProcRepository.java @@ -127,9 +127,14 @@ void executeStoredProcForBackfill( @Query(value = "EXEC sp_nrt_backfill_event :statusCd", nativeQuery = true) List executeBackfillEvent(@Param("statusCd") String statusCd); - @Procedure("sp_event_metric_cleanup_postprocessing") - void executeEventMetricCleanup(); + @Query( + value = + "DECLARE @status int; EXEC @status = dbo.sp_event_metric_cleanup_postprocessing; SELECT @status", + nativeQuery = true) + Integer executeEventMetricCleanup(); - @Procedure("sp_lab100_cleanup") - void executeLab100Cleanup(); + @Query( + value = "DECLARE @status int; EXEC @status = dbo.sp_lab100_cleanup; SELECT @status", + nativeQuery = true) + Integer executeLab100Cleanup(); } diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java index b0e8dc932..f210df96e 100644 --- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java @@ -76,6 +76,7 @@ import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; import lombok.NonNull; @@ -1569,15 +1570,22 @@ private void completeLog(String sp) { @Scheduled(cron = "${service.schedule.event-metric-cleanup}") protected void eventMetricCleanup() { - logger.info("Running event metric cleanup..."); - postProcRepository.executeEventMetricCleanup(); - logger.info(SP_EXECUTION_COMPLETED, "sp_event_metric_cleanup_postprocessing"); + processScheduledProcedure( + "sp_event_metric_cleanup_postprocessing", postProcRepository::executeEventMetricCleanup); } @Scheduled(cron = "${service.schedule.lab100-cleanup}") protected void lab100Cleanup() { - logger.info("Running lab100 cleanup..."); - postProcRepository.executeLab100Cleanup(); - logger.info(SP_EXECUTION_COMPLETED, "sp_lab100_cleanup"); + processScheduledProcedure("sp_lab100_cleanup", postProcRepository::executeLab100Cleanup); + } + + private void processScheduledProcedure(String name, Supplier scheduledProcedure) { + logger.info("Running {}...", name); + switch (ScheduledExecutionStatus.fromReturnCode(scheduledProcedure.get())) { + case COMPLETED -> logger.info(SP_EXECUTION_COMPLETED, name); + case SKIPPED -> logger.info("Skipped {} because it's already running", name); + case FAILED -> throw new DataProcessingException(name + " reported a cleanup failure"); + default -> throw new DataProcessingException(name + " encountered an unknown status"); + } } } diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatus.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatus.java new file mode 100644 index 000000000..a56a0a987 --- /dev/null +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatus.java @@ -0,0 +1,27 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import gov.cdc.nbs.report.pipeline.util.DataProcessingException; + +enum ScheduledExecutionStatus { + COMPLETED(1), + SKIPPED(-2), + FAILED(-1); + + private final int returnCode; + + ScheduledExecutionStatus(int returnCode) { + this.returnCode = returnCode; + } + + static ScheduledExecutionStatus fromReturnCode(Integer returnCode) { + if (returnCode == null) { + throw new DataProcessingException("Cleanup procedure did not return an execution status"); + } + for (ScheduledExecutionStatus status : values()) { + if (status.returnCode == returnCode) { + return status; + } + } + throw new DataProcessingException("Unexpected cleanup procedure return code: " + returnCode); + } +} diff --git a/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/345-sp_event_metric_cleanup_postprocessing-001.sql b/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/345-sp_event_metric_cleanup_postprocessing-001.sql index 5112335d1..74aed2651 100644 --- a/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/345-sp_event_metric_cleanup_postprocessing-001.sql +++ b/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/345-sp_event_metric_cleanup_postprocessing-001.sql @@ -19,10 +19,24 @@ BEGIN DECLARE @Proc_Step_Name VARCHAR(200)= ''; DECLARE @Dataflow_Name VARCHAR(200) = 'Event Metric Cleanup POST-Processing'; DECLARE @Package_Name VARCHAR(200) = 'sp_event_metric_cleanup_postprocessing'; + DECLARE @lock_resource NVARCHAR(255) = N'RTR:scheduled:event-metric-cleanup'; + DECLARE @lock_result INT; + DECLARE @lock_acquired BIT = 0; BEGIN TRY - + EXEC @lock_result = sys.sp_getapplock + @Resource = @lock_resource, + @LockMode = N'Exclusive', + @LockOwner = N'Session', + @LockTimeout = 0; + + IF @lock_result < 0 + BEGIN + RETURN -2; + END; + + SET @lock_acquired = 1; SET @Proc_Step_Name = 'SP_Start'; INSERT INTO dbo.job_flow_log ( batch_id @@ -206,7 +220,13 @@ BEGIN INSERT INTO [dbo].[job_flow_log] (batch_id, [Dataflow_Name], [package_Name], [Status_Type], [step_number], [step_name], [row_count]) VALUES (@batch_id, @Dataflow_Name, @Package_Name, 'COMPLETE', 999, @Proc_Step_name, @RowCount_no); - + + EXEC sys.sp_releaseapplock + @Resource = @lock_resource, + @LockOwner = N'Session'; + + RETURN 1; + ------------------------------------------------------------------------------------------- END TRY @@ -214,6 +234,13 @@ BEGIN IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION; + IF @lock_acquired = 1 + BEGIN + EXEC sys.sp_releaseapplock + @Resource = @lock_resource, + @LockOwner = N'Session'; + END; + -- Construct the error message string with all details: DECLARE @FullErrorMessage VARCHAR(8000) = 'Error Number: ' + CAST(ERROR_NUMBER() AS VARCHAR(10)) + CHAR(13) + CHAR(10) + -- Carriage return and line feed for new lines diff --git a/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/371-sp_lab100_cleanup.sql b/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/371-sp_lab100_cleanup.sql index ec3d02c17..a45693f91 100644 --- a/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/371-sp_lab100_cleanup.sql +++ b/reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/371-sp_lab100_cleanup.sql @@ -22,8 +22,23 @@ AS DECLARE @Proc_Step_Name VARCHAR(200)= ''; DECLARE @Dataflow_Name VARCHAR(200) = 'Lab 100 Cleanup'; DECLARE @Package_Name VARCHAR(200) = 'sp_lab100_cleanup'; + DECLARE @lock_resource NVARCHAR(255) = N'RTR:scheduled:lab100-cleanup'; + DECLARE @lock_result INT; + DECLARE @lock_acquired BIT = 0; BEGIN try + EXEC @lock_result = sys.sp_getapplock + @Resource = @lock_resource, + @LockMode = N'Exclusive', + @LockOwner = N'Session', + @LockTimeout = 0; + + IF @lock_result < 0 + BEGIN + RETURN -2; + END; + + SET @lock_acquired = 1; SET @Proc_Step_Name = 'SP_Start'; INSERT INTO dbo.job_flow_log @@ -139,6 +154,12 @@ AS COMMIT TRANSACTION; + + EXEC sys.sp_releaseapplock + @Resource = @lock_resource, + @LockOwner = N'Session'; + + RETURN 1; --------------------------------------------------------------------------------------------------------- END try @@ -146,6 +167,13 @@ AS IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION; + IF @lock_acquired = 1 + BEGIN + EXEC sys.sp_releaseapplock + @Resource = @lock_resource, + @LockOwner = N'Session'; + END; + -- Construct the error message string with all details: DECLARE @FullErrorMessage VARCHAR(8000) = 'Error Number: ' + Cast(Error_number() AS VARCHAR(10)) diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/CleanupStatusRepositoryTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/CleanupStatusRepositoryTest.java new file mode 100644 index 000000000..b683c400e --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/CleanupStatusRepositoryTest.java @@ -0,0 +1,22 @@ +package gov.cdc.nbs.report.pipeline.integration.unit; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; + +class CleanupStatusRepositoryTest extends UnitTest { + + @Autowired private PostProcRepository postProcRepository; + + @Test + void returnsExplicitSuccessForEventMetricCleanup() { + assertEquals(1, postProcRepository.executeEventMetricCleanup()); + } + + @Test + void returnsExplicitSuccessForLab100Cleanup() { + assertEquals(1, postProcRepository.executeLab100Cleanup()); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockConcurrencyTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockConcurrencyTest.java new file mode 100644 index 000000000..f06bbd75e --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockConcurrencyTest.java @@ -0,0 +1,134 @@ +package gov.cdc.nbs.report.pipeline.integration.unit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTimeout; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Duration; +import java.util.function.Supplier; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; + +@Execution(ExecutionMode.SAME_THREAD) +class ScheduledCleanupAppLockConcurrencyTest extends UnitTest { + + private static final String EVENT_METRIC_LOCK = "RTR:scheduled:event-metric-cleanup"; + private static final String EVENT_METRIC_PROCEDURE = "sp_event_metric_cleanup_postprocessing"; + private static final String LAB100_LOCK = "RTR:scheduled:lab100-cleanup"; + private static final String LAB100_PROCEDURE = "sp_lab100_cleanup"; + private static final Duration SKIP_TIMEOUT = Duration.ofSeconds(2); + + @Autowired private PostProcRepository postProcRepository; + + @Value("${spring.datasource.admin.url}") + private String adminJdbcUrl; + + @Value("${spring.datasource.admin.username}") + private String adminJdbcUser; + + @Value("${spring.datasource.admin.password}") + private String adminJdbcPassword; + + @Test + void eventMetricCleanupSkipsWithoutWritingNormalJobFlowLogsWhenLocked() throws SQLException { + assertSkipsWhenLocked( + EVENT_METRIC_LOCK, EVENT_METRIC_PROCEDURE, postProcRepository::executeEventMetricCleanup); + } + + @Test + void lab100CleanupSkipsWithoutWritingNormalJobFlowLogsWhenLocked() throws SQLException { + assertSkipsWhenLocked(LAB100_LOCK, LAB100_PROCEDURE, postProcRepository::executeLab100Cleanup); + } + + @Test + void cleanupLocksDoNotBlockTheOtherCleanupProcedure() throws SQLException { + try (Connection eventMetricLockOwner = adminConnection()) { + acquireLock(eventMetricLockOwner, EVENT_METRIC_LOCK); + try { + assertEquals(1, postProcRepository.executeLab100Cleanup()); + } finally { + releaseLock(eventMetricLockOwner, EVENT_METRIC_LOCK); + } + } + + try (Connection lab100LockOwner = adminConnection()) { + acquireLock(lab100LockOwner, LAB100_LOCK); + try { + assertEquals(1, postProcRepository.executeEventMetricCleanup()); + } finally { + releaseLock(lab100LockOwner, LAB100_LOCK); + } + } + } + + private void assertSkipsWhenLocked( + String lockResource, String procedureName, Supplier scheduledProcedure) + throws SQLException { + try (Connection lockOwner = adminConnection()) { + acquireLock(lockOwner, lockResource); + try { + long jobFlowLogCount = normalJobFlowLogCount(lockOwner, procedureName); + + assertTimeout(SKIP_TIMEOUT, () -> assertEquals(-2, scheduledProcedure.get())); + + assertEquals(jobFlowLogCount, normalJobFlowLogCount(lockOwner, procedureName)); + } finally { + releaseLock(lockOwner, lockResource); + } + } + } + + private long normalJobFlowLogCount(Connection connection, String procedureName) + throws SQLException { + try (Statement statement = connection.createStatement(); + ResultSet result = + statement.executeQuery( + "SELECT COUNT(*) FROM dbo.job_flow_log WHERE package_name = '" + + procedureName + + "' AND status_type IN ('START', 'COMPLETE')")) { + result.next(); + return result.getLong(1); + } + } + + private void acquireLock(Connection connection, String lockResource) throws SQLException { + assertTrue(lockResult(connection, lockResource) >= 0); + } + + private int lockResult(Connection connection, String lockResource) throws SQLException { + try (Statement statement = connection.createStatement(); + ResultSet result = + statement.executeQuery( + "DECLARE @lock_result int; EXEC @lock_result = sys.sp_getapplock " + + "@Resource = N'" + + lockResource + + "', @LockMode = N'Exclusive', @LockOwner = N'Session', @LockTimeout = 0; " + + "SELECT @lock_result")) { + result.next(); + return result.getInt(1); + } + } + + private void releaseLock(Connection connection, String lockResource) throws SQLException { + try (Statement statement = connection.createStatement()) { + statement.execute( + "EXEC sys.sp_releaseapplock @Resource = N'" + + lockResource + + "', @LockOwner = N'Session'"); + } + } + + private Connection adminConnection() throws SQLException { + String rdbJdbcUrl = adminJdbcUrl.replaceAll("databaseName=[^;]+", "databaseName=RDB_MODERN"); + return DriverManager.getConnection(rdbJdbcUrl, adminJdbcUser, adminJdbcPassword); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockReleaseTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockReleaseTest.java new file mode 100644 index 000000000..b2192b605 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockReleaseTest.java @@ -0,0 +1,114 @@ +package gov.cdc.nbs.report.pipeline.integration.unit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; + +class ScheduledCleanupAppLockReleaseTest extends UnitTest { + + private static final String EVENT_METRIC_LOCK = "RTR:scheduled:event-metric-cleanup"; + private static final String LAB100_LOCK = "RTR:scheduled:lab100-cleanup"; + private static final String LAB100_FAILURE_TRIGGER = "trg_test_lab100_cleanup_failure"; + + @Autowired private PostProcRepository postProcRepository; + + @Value("${spring.datasource.admin.url}") + private String adminJdbcUrl; + + @Value("${spring.datasource.admin.username}") + private String adminJdbcUser; + + @Value("${spring.datasource.admin.password}") + private String adminJdbcPassword; + + @Test + void releasesEventMetricLockAfterAProcedureFailure() throws SQLException { + try (Connection connection = adminConnection()) { + String originalValue = readMetricsLookbackDays(connection); + try { + updateMetricsLookbackDays(connection, "not-a-number"); + + assertEquals(-1, postProcRepository.executeEventMetricCleanup()); + assertTrue(canAcquireLock(EVENT_METRIC_LOCK)); + } finally { + updateMetricsLookbackDays(connection, originalValue); + } + } + } + + @Test + void releasesLab100LockAfterAProcedureFailure() throws SQLException { + try (Connection connection = adminConnection()) { + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TRIGGER dbo." + + LAB100_FAILURE_TRIGGER + + " ON dbo.LAB100 AFTER UPDATE AS BEGIN THROW 51000, 'test failure', 1; END"); + assertEquals(-1, postProcRepository.executeLab100Cleanup()); + assertTrue(canAcquireLock(LAB100_LOCK)); + } finally { + try (Statement statement = connection.createStatement()) { + statement.execute("DROP TRIGGER IF EXISTS dbo." + LAB100_FAILURE_TRIGGER); + } + } + } + } + + private String readMetricsLookbackDays(Connection connection) throws SQLException { + try (Statement statement = connection.createStatement(); + ResultSet result = + statement.executeQuery( + "SELECT config_value FROM dbo.nrt_odse_NBS_configuration " + + "WHERE config_key = 'METRICS_GOBACKBY_DAYS'")) { + result.next(); + return result.getString(1); + } + } + + private void updateMetricsLookbackDays(Connection connection, String value) throws SQLException { + try (Statement statement = connection.createStatement()) { + statement.executeUpdate( + "UPDATE dbo.nrt_odse_NBS_configuration SET config_value = '" + + value.replace("'", "''") + + "' WHERE config_key = 'METRICS_GOBACKBY_DAYS'"); + } + } + + private boolean canAcquireLock(String lockResource) throws SQLException { + try (Connection connection = adminConnection(); + Statement statement = connection.createStatement()) { + int lockResult; + try (ResultSet result = + statement.executeQuery( + "DECLARE @lock_result int; EXEC @lock_result = sys.sp_getapplock " + + "@Resource = N'" + + lockResource + + "', @LockMode = N'Exclusive', @LockOwner = N'Session', @LockTimeout = 0; " + + "SELECT @lock_result")) { + result.next(); + lockResult = result.getInt(1); + } + if (lockResult >= 0) { + statement.execute( + "EXEC sys.sp_releaseapplock @Resource = N'" + + lockResource + + "', @LockOwner = N'Session'"); + } + return lockResult >= 0; + } + } + + private Connection adminConnection() throws SQLException { + String rdbJdbcUrl = adminJdbcUrl.replaceAll("databaseName=[^;]+", "databaseName=RDB_MODERN"); + return DriverManager.getConnection(rdbJdbcUrl, adminJdbcUser, adminJdbcPassword); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/EventMetricCleanupTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/EventMetricCleanupTest.java index 6a0c5c333..18b20ba4b 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/EventMetricCleanupTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/EventMetricCleanupTest.java @@ -2,17 +2,20 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import ch.qos.logback.classic.Logger; import ch.qos.logback.classic.spi.ILoggingEvent; import ch.qos.logback.core.read.ListAppender; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import gov.cdc.nbs.report.pipeline.util.DataProcessingException; import gov.cdc.nbs.report.pipeline.util.kafka.RetryTopicResolver; import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; @@ -57,6 +60,7 @@ void setUp() { service.initMetrics(); datamartProcessor.initMetrics(); service.setServiceEnable(true); + when(postProcRepository.executeEventMetricCleanup()).thenReturn(1); Logger logger = (Logger) LoggerFactory.getLogger(PostProcessingService.class); listAppender.start(); @@ -81,12 +85,36 @@ void eventMetricCleanup_invokesRepository() { void eventMetricCleanup_logsCompletion() { service.eventMetricCleanup(); - boolean completionLogged = + assertTrue( listAppender.list.stream() .anyMatch( - e -> e.getFormattedMessage().contains("sp_event_metric_cleanup_postprocessing")); + e -> + e.getFormattedMessage() + .equals( + "Stored proc execution completed:" + + " sp_event_metric_cleanup_postprocessing"))); + } + + @Test + void eventMetricCleanup_logsAlreadyRunningSkipWithoutCompletion() { + when(postProcRepository.executeEventMetricCleanup()).thenReturn(-2); + + service.eventMetricCleanup(); + assertTrue( - completionLogged, "Expected completion log for sp_event_metric_cleanup_postprocessing"); + listAppender.list.stream() + .anyMatch( + e -> + e.getFormattedMessage() + .contains( + "Skipped sp_event_metric_cleanup_postprocessing because it's already running"))); + assertTrue( + listAppender.list.stream() + .noneMatch( + e -> + e.getFormattedMessage() + .contains("Stored proc execution completed: sp_event_metric_cleanup"))); + verify(postProcRepository).executeEventMetricCleanup(); } @Test @@ -101,13 +129,21 @@ void eventMetricCleanup_isScheduledWithCorrectCronProperty() throws NoSuchMethod "cron must reference the event-metric-cleanup property"); } + @Test + void eventMetricCleanup_throwsWhenProcedureReportsFailure() { + when(postProcRepository.executeEventMetricCleanup()).thenReturn(-1); + + assertThrows(DataProcessingException.class, () -> service.eventMetricCleanup()); + verify(postProcRepository).executeEventMetricCleanup(); + } + @Test void eventMetricCleanup_propagatesRepositoryException() { doThrow(new RuntimeException("proc failed")) .when(postProcRepository) .executeEventMetricCleanup(); - org.junit.jupiter.api.Assertions.assertThrows( - RuntimeException.class, () -> service.eventMetricCleanup()); + assertThrows(RuntimeException.class, () -> service.eventMetricCleanup()); + verify(postProcRepository).executeEventMetricCleanup(); } } diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/Lab100CleanupTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/Lab100CleanupTest.java index 43faf94be..6fbcefc18 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/Lab100CleanupTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/Lab100CleanupTest.java @@ -2,17 +2,20 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import ch.qos.logback.classic.Logger; import ch.qos.logback.classic.spi.ILoggingEvent; import ch.qos.logback.core.read.ListAppender; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import gov.cdc.nbs.report.pipeline.util.DataProcessingException; import gov.cdc.nbs.report.pipeline.util.kafka.RetryTopicResolver; import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; @@ -57,6 +60,7 @@ void setUp() { service.initMetrics(); datamartProcessor.initMetrics(); service.setServiceEnable(true); + when(postProcRepository.executeLab100Cleanup()).thenReturn(1); Logger logger = (Logger) LoggerFactory.getLogger(PostProcessingService.class); listAppender.start(); @@ -78,13 +82,36 @@ void lab100Cleanup_invokesRepository() { } @Test - void lab100CleanupCleanup_logsCompletion() { + void lab100Cleanup_logsCompletion() { service.lab100Cleanup(); - boolean completionLogged = + assertTrue( listAppender.list.stream() - .anyMatch(e -> e.getFormattedMessage().contains("sp_lab100_cleanup")); - assertTrue(completionLogged, "Expected completion log for sp_lab100_cleanup"); + .anyMatch( + e -> + e.getFormattedMessage() + .equals("Stored proc execution completed: sp_lab100_cleanup"))); + } + + @Test + void lab100Cleanup_logsAlreadyRunningSkipWithoutCompletion() { + when(postProcRepository.executeLab100Cleanup()).thenReturn(-2); + + service.lab100Cleanup(); + + assertTrue( + listAppender.list.stream() + .anyMatch( + e -> + e.getFormattedMessage() + .contains("Skipped sp_lab100_cleanup because it's already running"))); + assertTrue( + listAppender.list.stream() + .noneMatch( + e -> + e.getFormattedMessage() + .contains("Stored proc execution completed: sp_lab100_cleanup"))); + verify(postProcRepository).executeLab100Cleanup(); } @Test @@ -99,11 +126,19 @@ void lab100Cleanup_isScheduledWithCorrectCronProperty() throws NoSuchMethodExcep "cron must reference the lab100-cleanup property"); } + @Test + void lab100Cleanup_throwsWhenProcedureReportsFailure() { + when(postProcRepository.executeLab100Cleanup()).thenReturn(-1); + + assertThrows(DataProcessingException.class, () -> service.lab100Cleanup()); + verify(postProcRepository).executeLab100Cleanup(); + } + @Test void lab100Cleanup_propagatesRepositoryException() { doThrow(new RuntimeException("proc failed")).when(postProcRepository).executeLab100Cleanup(); - org.junit.jupiter.api.Assertions.assertThrows( - RuntimeException.class, () -> service.lab100Cleanup()); + assertThrows(RuntimeException.class, () -> service.lab100Cleanup()); + verify(postProcRepository).executeLab100Cleanup(); } } diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatusTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatusTest.java new file mode 100644 index 000000000..afd900767 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatusTest.java @@ -0,0 +1,25 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import gov.cdc.nbs.report.pipeline.util.DataProcessingException; +import org.junit.jupiter.api.Test; + +class ScheduledExecutionStatusTest { + + @Test + void mapsExplicitCompletionSkipAndFailureReturnCodes() { + assertEquals(ScheduledExecutionStatus.COMPLETED, ScheduledExecutionStatus.fromReturnCode(1)); + assertEquals(ScheduledExecutionStatus.SKIPPED, ScheduledExecutionStatus.fromReturnCode(-2)); + assertEquals(ScheduledExecutionStatus.FAILED, ScheduledExecutionStatus.fromReturnCode(-1)); + } + + @Test + void rejectsZeroAndMissingOrUnexpectedReturnCodes() { + assertThrows(DataProcessingException.class, () -> ScheduledExecutionStatus.fromReturnCode(0)); + assertThrows( + DataProcessingException.class, () -> ScheduledExecutionStatus.fromReturnCode(null)); + assertThrows(DataProcessingException.class, () -> ScheduledExecutionStatus.fromReturnCode(2)); + } +}