diff --git a/reporting-pipeline-service/build.gradle b/reporting-pipeline-service/build.gradle index dff223e11..9f1b239dc 100644 --- a/reporting-pipeline-service/build.gradle +++ b/reporting-pipeline-service/build.gradle @@ -142,6 +142,7 @@ dependencies { implementation 'org.springframework.boot:spring-boot-starter-actuator' implementation 'org.springframework.boot:spring-boot-starter-data-jpa' implementation 'org.springframework.boot:spring-boot-starter-web' + implementation 'org.springframework.boot:spring-boot-starter-validation' developmentOnly 'org.springframework.boot:spring-boot-devtools' // Kafka diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/ReportingPipelineServiceApplication.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/ReportingPipelineServiceApplication.java index 770be1b97..4a4058a81 100644 --- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/ReportingPipelineServiceApplication.java +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/ReportingPipelineServiceApplication.java @@ -1,6 +1,7 @@ package gov.cdc.nbs.report.pipeline; import gov.cdc.nbs.report.pipeline.config.EventProcedureLoggingProperties; +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.connector.ConnectorProperties; import gov.cdc.nbs.report.pipeline.lag.LagProperties; import org.springframework.boot.SpringApplication; @@ -15,7 +16,8 @@ @EnableConfigurationProperties({ ConnectorProperties.class, LagProperties.class, - EventProcedureLoggingProperties.class + EventProcedureLoggingProperties.class, + PostProcessingProperties.class }) public class ReportingPipelineServiceApplication { diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/config/PostProcessingProperties.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/config/PostProcessingProperties.java new file mode 100644 index 000000000..7bc8283cf --- /dev/null +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/config/PostProcessingProperties.java @@ -0,0 +1,9 @@ +package gov.cdc.nbs.report.pipeline.config; + +import jakarta.validation.constraints.Min; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +@ConfigurationProperties(prefix = "service.post-processing") +@Validated +public record PostProcessingProperties(@Min(0) int maxBatchSize) {} 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 b786f98df..06189a6e8 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 @@ -42,6 +42,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.BackfillData; @@ -164,6 +165,7 @@ public class PostProcessingService { private final ProcessDatamartData dmProcessor; private final RetryTopicResolver retryTopicResolver; + private final PostProcessingProperties postProcessingProperties; static final String PAYLOAD = "payload"; static final String SP_EXECUTION_COMPLETED = "Stored proc execution completed: {}"; @@ -1481,7 +1483,7 @@ protected void processDatamartIds() { } } - private String listToParameterString(Collection inputList) { + private String listToParameterString(Collection inputList) { return Optional.ofNullable(inputList) .map(list -> list.stream().map(String::valueOf).distinct().collect(Collectors.joining(","))) .orElse(""); @@ -1514,10 +1516,14 @@ private void processTopic( Collection ids, Consumer repositoryMethod, String... names) { - if (!ids.isEmpty()) { - String idsString = listToParameterString(ids); - String spName = names.length > 0 ? names[0] : entity.getStoredProcedure(); - prepareAndLog(keyTopic, idsString, entity.getEntityName(), spName); + String spName = names.length > 0 ? names[0] : entity.getStoredProcedure(); + List> chunks = + UidChunker.chunkDistinct(ids, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String idsString = listToParameterString(chunk); + prepareAndLog( + keyTopic, entity.getEntityName(), spName, index + 1, chunks.size(), chunk.size()); repositoryMethod.accept(idsString); completeLog(spName); } @@ -1525,11 +1531,17 @@ private void processTopic( private void processTopic( String keyTopic, Entity entity, Collection cds, Consumer repositoryMethod) { - String cdString = cds.stream().distinct().collect(Collectors.joining(",")); String spName = entity.getStoredProcedure(); - prepareAndLog(keyTopic, cdString, entity.getEntityName(), spName); - repositoryMethod.accept(cdString); - completeLog(spName); + List> chunks = + UidChunker.chunkDistinct(cds, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String cdString = listToParameterString(chunk); + prepareAndLog( + keyTopic, entity.getEntityName(), spName, index + 1, chunks.size(), chunk.size()); + repositoryMethod.accept(cdString); + completeLog(spName); + } } private List processTopic( @@ -1540,11 +1552,19 @@ private List processTopic( Consumer> checkResult, String... names) { String spName = names.length > 0 ? names[0] : entity.getStoredProcedure(); - String idString = listToParameterString(ids); - prepareAndLog(keyTopic, idString, entity.getEntityName(), spName); - List result = repositoryMethod.apply(idString); - checkResult.accept(result); - completeLog(spName); + List result = new ArrayList<>(); + List> chunks = + UidChunker.chunkDistinct(ids, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String idString = listToParameterString(chunk); + prepareAndLog( + keyTopic, entity.getEntityName(), spName, index + 1, chunks.size(), chunk.size()); + List chunkResult = repositoryMethod.apply(idString); + checkResult.accept(chunkResult); + result.addAll(chunkResult); + completeLog(spName); + } return result; } @@ -1556,28 +1576,47 @@ private void processTopic( BiFunction> repositoryMethod, Consumer> checkResult) { String name = entity.getEntityName(); - name = logger.isInfoEnabled() ? StringUtils.capitalize(name) : name; - String idString = listToParameterString(ids); - logger.info( - "Processing {} for topic: {}. Calling stored proc: {} '{}', '{}'", - name, - keyTopic, - entity.getStoredProcedure(), - idString, - vals); - List result = repositoryMethod.apply(idString, vals); - checkResult.accept(result); - completeLog(entity.getStoredProcedure()); + String displayName = logger.isInfoEnabled() ? StringUtils.capitalize(name) : name; + List> chunks = + UidChunker.chunkDistinct(ids, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String idString = listToParameterString(chunk); + logger.info( + "Processing {} for topic: {}. Calling stored proc: {} (chunk {}/{}, distinct UIDs: {}," + + " max batch size: {}, table: '{}')", + displayName, + keyTopic, + entity.getStoredProcedure(), + index + 1, + chunks.size(), + chunk.size(), + postProcessingProperties.maxBatchSize(), + vals); + List result = repositoryMethod.apply(idString, vals); + checkResult.accept(result); + completeLog(entity.getStoredProcedure()); + } } - private void prepareAndLog(String keyTopic, String idString, String name, String spName) { + private void prepareAndLog( + String keyTopic, + String name, + String spName, + int chunkNumber, + int totalChunks, + int distinctCount) { name = logger.isInfoEnabled() ? StringUtils.capitalize(name) : name; logger.info( - "Processing {} for topic: {}. Calling stored proc: {} '{}'", + "Processing {} for topic: {}. Calling stored proc: {} (chunk {}/{}, distinct values: {}," + + " max batch size: {})", name, keyTopic, spName, - idString); + chunkNumber, + totalChunks, + distinctCount, + postProcessingProperties.maxBatchSize()); } private void completeLog(String sp) { diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartData.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartData.java index 682c09748..2de0f108e 100644 --- a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartData.java +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartData.java @@ -4,6 +4,7 @@ import static gov.cdc.nbs.report.pipeline.util.UtilHelper.errorMessage; import com.google.common.base.Strings; +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.BackfillData; @@ -58,6 +59,8 @@ public class ProcessDatamartData { @Qualifier("ppInvestigationRepository") private final InvestigationRepository invRepository; + private final PostProcessingProperties postProcessingProperties; + private final CustomJsonGeneratorImpl jsonGenerator = new CustomJsonGeneratorImpl(); private final ModelMapper modelMapper = new ModelMapper(); private final Object retryCacheLock = new Object(); @@ -73,11 +76,6 @@ public class ProcessDatamartData { static final String MULTI_ID_DATAMART = "MultiId_Datamart"; static final String SP_EXECUTION_COMPLETED = "Stored proc execution completed: {}"; - static final String PROCESSING_MESSAGE_TOPIC_LOG_MSG = - "Processing {} message topic. Calling stored proc: {} '{}'"; - static final String PROCESSING_MESSAGE_TOPIC_LOG_MSG_2 = - "Processing {} message topic. Calling stored proc: {} '{}', '{}'"; - static final String STATUS_READY = "READY"; static final String STATUS_COMPLETE = "COMPLETE"; static final String STATUS_SUSPENDED = "SUSPENDED"; @@ -376,8 +374,8 @@ private void processDatamart( executeDmProc( COVID_VACCINATION_DATAMART, invRepository::executeStoredProcForCovidVacDatamart, - cases, - pats, + uids, + patUids, this::checkResult); break; case COVID_LAB_DATAMART: @@ -487,13 +485,6 @@ private void processMultiIdDatamart( BatchProcessingState state, Map> dmMulti, List caseLabInvUids) { Map> multi = (dmMulti == null) ? Map.of() : dmMulti; - String invString = listToParameterString(multi.get(INVESTIGATION.getEntityName())); - String obsString = listToParameterString(multi.get(OBSERVATION.getEntityName())); - String notifString = listToParameterString(multi.get(NOTIFICATION.getEntityName())); - String patString = listToParameterString(multi.get(PATIENT.getEntityName())); - String provString = listToParameterString(multi.get(PROVIDER.getEntityName())); - String orgString = listToParameterString(multi.get(ORGANIZATION.getEntityName())); - // INV_SUMM additionally covers investigations whose CASE_LAB_DATAMART was rebuilt this batch. List invSummaryUids = new ArrayList<>( @@ -501,45 +492,69 @@ private void processMultiIdDatamart( invSummaryUids.addAll(caseLabInvUids); String invSummaryInvString = listToParameterString(invSummaryUids); - int totalLengthInvSummary = - invSummaryInvString.length() + notifString.length() + obsString.length(); - int totalLengthMorbReportDM = - obsString.length() - + patString.length() - + provString.length() - + orgString.length() - + invString.length(); + Map> invSummaryInputs = new LinkedHashMap<>(); + invSummaryInputs.put(INVESTIGATION.getEntityName(), invSummaryUids); + invSummaryInputs.put(NOTIFICATION.getEntityName(), multi.get(NOTIFICATION.getEntityName())); + invSummaryInputs.put(OBSERVATION.getEntityName(), multi.get(OBSERVATION.getEntityName())); + + Map> morbidityInputs = new LinkedHashMap<>(); + morbidityInputs.put(OBSERVATION.getEntityName(), multi.get(OBSERVATION.getEntityName())); + morbidityInputs.put(PATIENT.getEntityName(), multi.get(PATIENT.getEntityName())); + morbidityInputs.put(PROVIDER.getEntityName(), multi.get(PROVIDER.getEntityName())); + morbidityInputs.put(ORGANIZATION.getEntityName(), multi.get(ORGANIZATION.getEntityName())); + morbidityInputs.put(INVESTIGATION.getEntityName(), multi.get(INVESTIGATION.getEntityName())); try { - if (totalLengthInvSummary > 0) { - // reusing the same DTO class for Dynamic Marts - logger.info( - "Executing stored proc: sp_inv_summary_datamart_postprocessing '{}', '{}', '{}'", - invSummaryInvString, - notifString, - obsString); - List dmDataList = - procRepository.executeStoredProcForInvSummaryDatamart( - invSummaryInvString, notifString, obsString); - logExecutionCompleted("sp_inv_summary_datamart_postprocessing"); - processDynDatamart(state, dmDataList); + List>> summaryBatches = + UidChunker.chunkDistinct(invSummaryInputs, postProcessingProperties.maxBatchSize()); + if (!summaryBatches.isEmpty()) { + // Reusing the same DTO class for Dynamic Marts. Collect all chunks before dispatching + // dynamic datamarts so their prerequisite ordering remains unchanged. + List dmData = new ArrayList<>(); + for (int index = 0; index < summaryBatches.size(); index++) { + Map> batch = summaryBatches.get(index); + String invString = listToParameterString(batch.get(INVESTIGATION.getEntityName())); + String notifString = listToParameterString(batch.get(NOTIFICATION.getEntityName())); + String obsString = listToParameterString(batch.get(OBSERVATION.getEntityName())); + logDatamartChunk( + "INV_SUMMARY_DATAMART", + "sp_inv_summary_datamart_postprocessing", + index + 1, + summaryBatches.size(), + batch.values().stream().mapToInt(List::size).sum(), + postProcessingProperties.maxBatchSize()); + dmData.addAll( + procRepository.executeStoredProcForInvSummaryDatamart( + invString, notifString, obsString)); + logExecutionCompleted("sp_inv_summary_datamart_postprocessing"); + } + processDynDatamart(state, dmData); incrementIf(ppDmSuccess, !state.isRetry); } else { logger.info("No updates to INV_SUMMARY Datamart"); } - if (totalLengthMorbReportDM > 0) { - logger.info( - "Executing stored proc: sp_morbidity_report_datamart_postprocessing '{}', '{}', '{}'," - + " '{}', '{}'", - obsString, - patString, - provString, - orgString, - invString); - procRepository.executeStoredProcForMorbidityReportDatamart( - obsString, patString, provString, orgString, invString); - logExecutionCompleted("sp_morbidity_report_datamart_postprocessing"); + List>> morbidityBatches = + UidChunker.chunkDistinct(morbidityInputs, postProcessingProperties.maxBatchSize()); + if (!morbidityBatches.isEmpty()) { + for (int index = 0; index < morbidityBatches.size(); index++) { + Map> batch = morbidityBatches.get(index); + String obsString = listToParameterString(batch.get(OBSERVATION.getEntityName())); + String patString = listToParameterString(batch.get(PATIENT.getEntityName())); + String provString = listToParameterString(batch.get(PROVIDER.getEntityName())); + String orgString = listToParameterString(batch.get(ORGANIZATION.getEntityName())); + String invString = listToParameterString(batch.get(INVESTIGATION.getEntityName())); + logDatamartChunk( + "MORBIDITY_REPORT_DATAMART", + "sp_morbidity_report_datamart_postprocessing", + index + 1, + morbidityBatches.size(), + batch.values().stream().mapToInt(List::size).sum(), + postProcessingProperties.maxBatchSize()); + procRepository.executeStoredProcForMorbidityReportDatamart( + obsString, patString, provString, orgString, invString); + logExecutionCompleted("sp_morbidity_report_datamart_postprocessing"); + } incrementIf(ppDmSuccess, !state.isRetry); } else { logger.info("No updates to MORBIDITY_REPORT_DATAMART"); @@ -570,29 +585,37 @@ private void processDynDatamart(BatchProcessingState state, List d List> futures = new ArrayList<>(); datamartPhcIdMap.forEach( (datamart, phcIds) -> { - String phcIdsString = - phcIds.stream().map(String::valueOf).collect(Collectors.joining(",")); - futures.add( - CompletableFuture.runAsync( - () -> { - logger.info( - "Executing stored proc: sp_dyn_datamart_postprocessing '{}', '{}'", - datamart, - phcIdsString); - try { - procRepository.executeStoredProcForDynDatamart(datamart, phcIdsString); - logExecutionCompleted("sp_dyn_datamart_postprocessing"); - incrementIf(ppDmSuccess, !state.isRetry); - } catch (Exception e) { - incrementIf(ppDmFailure, !state.isRetry); - logger.error("Error processing dynamic datamart: {}", datamart, e); - state.registerFailure( + List> chunks = + UidChunker.chunkDistinct(phcIds, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String phcIdsString = listToParameterString(chunk); + int chunkNumber = index + 1; + futures.add( + CompletableFuture.runAsync( + () -> { + logDatamartChunk( datamart, - Collections.singletonMap(INVESTIGATION.getEntityName(), phcIds), - e); - } - }, - dynDmExecutor)); + "sp_dyn_datamart_postprocessing", + chunkNumber, + chunks.size(), + chunk.size(), + postProcessingProperties.maxBatchSize()); + try { + procRepository.executeStoredProcForDynDatamart(datamart, phcIdsString); + logExecutionCompleted("sp_dyn_datamart_postprocessing"); + incrementIf(ppDmSuccess, !state.isRetry); + } catch (Exception e) { + incrementIf(ppDmFailure, !state.isRetry); + logger.error("Error processing dynamic datamart: {}", datamart, e); + state.registerFailure( + datamart, + Collections.singletonMap(INVESTIGATION.getEntityName(), chunk), + e); + } + }, + dynDmExecutor)); + } }); // Wait for all async tasks to complete before returning @@ -601,37 +624,98 @@ private void processDynDatamart(BatchProcessingState state, List d } void processMetricEventDatamart(Map> dmMulti) { - String invString = listToParameterString(dmMulti.get(INVESTIGATION.getEntityName())); - String obsString = listToParameterString(dmMulti.get(OBSERVATION.getEntityName())); - String notifString = listToParameterString(dmMulti.get(NOTIFICATION.getEntityName())); - String ctrString = listToParameterString(dmMulti.get(CONTACT.getEntityName())); - String vaxString = listToParameterString(dmMulti.get(VACCINATION.getEntityName())); - - int totalLengthEventMetric = - invString.length() - + obsString.length() - + notifString.length() - + ctrString.length() - + vaxString.length(); - - if (totalLengthEventMetric > 0) { - Timer.Sample sample = metrics.startSample(); - logger.info( - "Executing stored proc: sp_event_metric_datamart_postprocessing '{}', '{}', '{}', '{}'," - + " '{}'", - invString, - obsString, - notifString, - ctrString, - vaxString); - procRepository.executeStoredProcForEventMetric( - invString, obsString, notifString, ctrString, vaxString); - logExecutionCompleted("sp_event_metric_datamart_postprocessing"); + List invUids = + new ArrayList<>( + dmMulti.getOrDefault(INVESTIGATION.getEntityName(), new ConcurrentLinkedQueue<>())); + List obsUids = + new ArrayList<>( + dmMulti.getOrDefault(OBSERVATION.getEntityName(), new ConcurrentLinkedQueue<>())); + List notifUids = + new ArrayList<>( + dmMulti.getOrDefault(NOTIFICATION.getEntityName(), new ConcurrentLinkedQueue<>())); + List contactUids = + new ArrayList<>( + dmMulti.getOrDefault(CONTACT.getEntityName(), new ConcurrentLinkedQueue<>())); + List vaxUids = + new ArrayList<>( + dmMulti.getOrDefault(VACCINATION.getEntityName(), new ConcurrentLinkedQueue<>())); + + int totalUidCount = + distinctCount(invUids) + + distinctCount(obsUids) + + distinctCount(notifUids) + + distinctCount(contactUids) + + distinctCount(vaxUids); + + if (totalUidCount == 0) { + return; + } + + Timer.Sample sample = metrics.startSample(); + try { + int maxBatchSize = postProcessingProperties.maxBatchSize(); + if (maxBatchSize == 0 || totalUidCount <= maxBatchSize) { + logDatamartChunk( + "EVENT_METRIC_DATAMART", + "sp_event_metric_datamart_postprocessing", + 1, + 1, + totalUidCount, + maxBatchSize); + Map> inputMap = new EnumMap<>(Entity.class); + inputMap.put(Entity.INVESTIGATION, invUids); + inputMap.put(Entity.OBSERVATION, obsUids); + inputMap.put(Entity.NOTIFICATION, notifUids); + inputMap.put(Entity.CONTACT, contactUids); + inputMap.put(Entity.VACCINATION, vaxUids); + processMetricEventChunk(inputMap); + } else { + processMetricEventChunks(INVESTIGATION, invUids); + processMetricEventChunks(OBSERVATION, obsUids); + processMetricEventChunks(NOTIFICATION, notifUids); + processMetricEventChunks(CONTACT, contactUids); + processMetricEventChunks(VACCINATION, vaxUids); + } incrementIf(ppDmSuccess, true); + } finally { metrics.stopSample(sample, processTimer); } } + private int distinctCount(Collection ids) { + return (int) ids.stream().distinct().count(); + } + + private void processMetricEventChunks(Entity entity, Collection ids) { + List> chunks = + UidChunker.chunkDistinct(ids, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + logDatamartChunk( + "EVENT_METRIC_DATAMART", + "sp_event_metric_datamart_postprocessing", + index + 1, + chunks.size(), + chunk.size(), + postProcessingProperties.maxBatchSize()); + Map> inputMap = new EnumMap<>(Entity.class); + inputMap.put(entity, chunk); + processMetricEventChunk(inputMap); + } + } + + private void processMetricEventChunk(Map> idsByEntity) { + String invString = listToParameterString(idsByEntity.get(Entity.INVESTIGATION)); + String obsString = listToParameterString(idsByEntity.get(Entity.OBSERVATION)); + String notifString = listToParameterString(idsByEntity.get(Entity.NOTIFICATION)); + String contactString = listToParameterString(idsByEntity.get(Entity.CONTACT)); + String vaxString = listToParameterString(idsByEntity.get(Entity.VACCINATION)); + + procRepository.executeStoredProcForEventMetric( + invString, obsString, notifString, contactString, vaxString); + logExecutionCompleted("sp_event_metric_datamart_postprocessing"); + } + /** * Scheduled task to reprocess failed datamarts by iterating through retry cache and invoking * {@link #processDmCache(Map, Long)} with snapshots of the cached IDs. @@ -816,36 +900,76 @@ private void executeDmProc( String ids, Consumer> checkResult) { if (!ids.isEmpty()) { - logger.info( - PROCESSING_MESSAGE_TOPIC_LOG_MSG, - dmEntity.getEntityName(), - dmEntity.getStoredProcedure(), - ids); - List result = repositoryMethod.apply(ids); - checkResult.accept(result); - logExecutionCompleted(dmEntity.getStoredProcedure()); + List> chunks = + UidChunker.chunkDistinct(parseUidParameter(ids), postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + List chunk = chunks.get(index); + String chunkIds = listToParameterString(chunk); + logDatamartChunk( + dmEntity.getEntityName(), + dmEntity.getStoredProcedure(), + index + 1, + chunks.size(), + chunk.size(), + postProcessingProperties.maxBatchSize()); + List result = repositoryMethod.apply(chunkIds); + checkResult.accept(result); + logExecutionCompleted(dmEntity.getStoredProcedure()); + } } } + private List parseUidParameter(String ids) { + return Arrays.stream(ids.split(",")).map(String::trim).map(Long::valueOf).toList(); + } + private void executeDmProc( Entity dmEntity, BiFunction> repositoryMethod, - String ids, - String pids, + Collection ids, + Collection pids, Consumer> checkResult) { - if (!ids.isEmpty() && !pids.isEmpty()) { - logger.info( - PROCESSING_MESSAGE_TOPIC_LOG_MSG_2, + Map> inputLists = new LinkedHashMap<>(); + inputLists.put("ids", ids); + inputLists.put("pids", pids); + + List>> chunks = + UidChunker.chunkDistinct(inputLists, postProcessingProperties.maxBatchSize()); + for (int index = 0; index < chunks.size(); index++) { + Map> chunk = chunks.get(index); + String chunkIds = listToParameterString(chunk.get("ids")); + String chunkPids = listToParameterString(chunk.get("pids")); + logDatamartChunk( dmEntity.getEntityName(), dmEntity.getStoredProcedure(), - ids, - pids); - List result = repositoryMethod.apply(ids, pids); + index + 1, + chunks.size(), + chunk.values().stream().mapToInt(List::size).sum(), + postProcessingProperties.maxBatchSize()); + List result = repositoryMethod.apply(chunkIds, chunkPids); checkResult.accept(result); logExecutionCompleted(dmEntity.getStoredProcedure()); } } + private void logDatamartChunk( + String datamart, + String storedProcedure, + int chunkNumber, + int totalChunks, + int distinctCount, + int maxBatchSize) { + logger.info( + "Processing datamart {} with stored proc {} (chunk {}/{}, distinct UIDs: {}, max batch" + + " size: {})", + datamart, + storedProcedure, + chunkNumber, + totalChunks, + distinctCount, + maxBatchSize); + } + private void logExecutionCompleted(String spName) { logger.info(SP_EXECUTION_COMPLETED, spName); } diff --git a/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunker.java b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunker.java new file mode 100644 index 000000000..fae6abea4 --- /dev/null +++ b/reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunker.java @@ -0,0 +1,91 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +final class UidChunker { + + private UidChunker() {} + + static List> chunkDistinct(Collection values, int maxSize) { + Objects.requireNonNull(values, "values"); + validateMaxSize(maxSize); + + List distinctValues = values.stream().distinct().toList(); + if (distinctValues.isEmpty()) { + return List.of(); + } + if (maxSize == 0 || distinctValues.size() <= maxSize) { + return List.of(distinctValues); + } + + List> chunks = new ArrayList<>((distinctValues.size() - 1) / maxSize + 1); + for (int start = 0; start < distinctValues.size(); start += maxSize) { + int end = Math.min(start + maxSize, distinctValues.size()); + chunks.add(List.copyOf(distinctValues.subList(start, end))); + } + return List.copyOf(chunks); + } + + static List>> chunkDistinct( + Map> valuesByKey, int maxSize) { + Objects.requireNonNull(valuesByKey, "valuesByKey"); + validateMaxSize(maxSize); + + Map> distinctValues = new LinkedHashMap<>(); + valuesByKey.forEach( + (key, values) -> { + if (values == null) { + return; + } + List> distinctChunks = chunkDistinct(values, 0); + if (!distinctChunks.isEmpty()) { + distinctValues.put(key, distinctChunks.get(0)); + } + }); + + if (distinctValues.isEmpty()) { + return List.of(); + } + int distinctValueCount = distinctValues.values().stream().mapToInt(List::size).sum(); + if (maxSize == 0 || distinctValueCount <= maxSize) { + return List.of(copyMap(distinctValues)); + } + + List>> chunks = new ArrayList<>(); + Map> currentChunk = new LinkedHashMap<>(); + int currentSize = 0; + for (Map.Entry> entry : distinctValues.entrySet()) { + for (T value : entry.getValue()) { + if (currentSize == maxSize) { + chunks.add(copyMap(currentChunk)); + currentChunk = new LinkedHashMap<>(); + currentSize = 0; + } + currentChunk.computeIfAbsent(entry.getKey(), ignored -> new ArrayList<>()).add(value); + currentSize++; + } + } + if (currentSize > 0) { + chunks.add(copyMap(currentChunk)); + } + return List.copyOf(chunks); + } + + private static void validateMaxSize(int maxSize) { + if (maxSize < 0) { + throw new IllegalArgumentException("maxSize must be zero or greater"); + } + } + + private static Map> copyMap(Map> values) { + Map> copy = new LinkedHashMap<>(); + values.forEach((key, value) -> copy.put(key, List.copyOf(value))); + return Collections.unmodifiableMap(copy); + } +} diff --git a/reporting-pipeline-service/src/main/resources/application.yaml b/reporting-pipeline-service/src/main/resources/application.yaml index 6ff7f13e7..6445a0503 100644 --- a/reporting-pipeline-service/src/main/resources/application.yaml +++ b/reporting-pipeline-service/src/main/resources/application.yaml @@ -151,6 +151,8 @@ management: enabled: true service: + post-processing: + max-batch-size: ${POST_PROCESSING_MAX_BATCH_SIZE:1000} fixed-delay: cached-ids: ${FIXED_DELAY_ID:20000} datamart: ${FIXED_DELAY_DM:60000} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/config/PostProcessingPropertiesTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/config/PostProcessingPropertiesTest.java new file mode 100644 index 000000000..cff898ff3 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/config/PostProcessingPropertiesTest.java @@ -0,0 +1,79 @@ +package gov.cdc.nbs.report.pipeline.config; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.autoconfigure.context.ConfigurationPropertiesAutoConfiguration; +import org.springframework.boot.autoconfigure.validation.ValidationAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +class PostProcessingPropertiesTest { + + private final ApplicationContextRunner contextRunner = + new ApplicationContextRunner() + .withConfiguration( + AutoConfigurations.of( + ConfigurationPropertiesAutoConfiguration.class, + ValidationAutoConfiguration.class)) + .withUserConfiguration(TestConfiguration.class) + .withPropertyValues( + "service.post-processing.max-batch-size=${POST_PROCESSING_MAX_BATCH_SIZE:1000}"); + + @Test + void bindsConfiguredMaxBatchSize() { + contextRunner + .withPropertyValues("service.post-processing.max-batch-size=25") + .run( + context -> { + PostProcessingProperties properties = context.getBean(PostProcessingProperties.class); + assertEquals(25, properties.maxBatchSize()); + }); + } + + @Test + void usesDefaultMaxBatchSize() { + contextRunner.run( + context -> { + PostProcessingProperties properties = context.getBean(PostProcessingProperties.class); + assertEquals(1000, properties.maxBatchSize()); + }); + } + + @Test + void bindsEnvironmentOverride() { + contextRunner + .withSystemProperties("POST_PROCESSING_MAX_BATCH_SIZE=25") + .run( + context -> { + PostProcessingProperties properties = context.getBean(PostProcessingProperties.class); + assertEquals(25, properties.maxBatchSize()); + }); + } + + @Test + void allowsZeroToDisableBatching() { + contextRunner + .withPropertyValues("service.post-processing.max-batch-size=0") + .run( + context -> { + PostProcessingProperties properties = context.getBean(PostProcessingProperties.class); + assertEquals(0, properties.maxBatchSize()); + }); + } + + @Test + void rejectsNegativeMaxBatchSize() { + contextRunner + .withPropertyValues("service.post-processing.max-batch-size=-1") + .run( + context -> { + assertNotNull(context.getStartupFailure()); + }); + } + + @EnableConfigurationProperties(PostProcessingProperties.class) + private static class TestConfiguration {} +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/DatamartProcessingTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/DatamartProcessingTest.java index 46e72e9a0..c3a548a1e 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/DatamartProcessingTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/DatamartProcessingTest.java @@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.DatamartData; @@ -54,6 +55,7 @@ void setUp() { kafkaTemplate, postProcRepositoryMock, investigationRepositoryMock, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); datamartProcessor.initMetrics(); } 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 18b20ba4b..3b71f4e49 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 @@ -13,6 +13,7 @@ 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.config.PostProcessingProperties; 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; @@ -47,6 +48,7 @@ void setUp() { kafkaTemplate, postProcRepository, investigationRepository, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); service = spy( @@ -55,6 +57,7 @@ void setUp() { investigationRepository, datamartProcessor, new RetryTopicResolver(), + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry()))); PostProcessingTestUtils.configureNrtTopics(service); service.initMetrics(); 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 6fbcefc18..8bba2a44f 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 @@ -13,6 +13,7 @@ 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.config.PostProcessingProperties; 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; @@ -47,6 +48,7 @@ void setUp() { kafkaTemplate, postProcRepository, investigationRepository, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); service = spy( @@ -55,6 +57,7 @@ void setUp() { investigationRepository, datamartProcessor, new RetryTopicResolver(), + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry()))); PostProcessingTestUtils.configureNrtTopics(service); service.initMetrics(); diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkRetryTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkRetryTest.java new file mode 100644 index 000000000..1458fa42d --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkRetryTest.java @@ -0,0 +1,85 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; +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.kafka.RetryTopicResolver; +import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.kafka.core.KafkaTemplate; + +class PostProcessingServiceChunkRetryTest { + + @Mock private PostProcRepository postProcRepository; + @Mock private InvestigationRepository investigationRepository; + @Mock private KafkaTemplate kafkaTemplate; + + private PostProcessingService service; + private SimpleMeterRegistry metricsRegistry; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + metricsRegistry = new SimpleMeterRegistry(); + ProcessDatamartData datamartProcessor = + new ProcessDatamartData( + kafkaTemplate, + postProcRepository, + investigationRepository, + new PostProcessingProperties(0), + new CustomMetrics(new SimpleMeterRegistry())); + service = + new PostProcessingService( + postProcRepository, + investigationRepository, + datamartProcessor, + new RetryTopicResolver(), + new PostProcessingProperties(2), + new CustomMetrics(metricsRegistry)); + service.setMaxRetries(2); + PostProcessingTestUtils.configureNrtTopics(service); + service.initMetrics(); + datamartProcessor.initMetrics(); + service.setServiceEnable(true); + + when(postProcRepository.executeStoredProcForPatientIds(anyString())).thenReturn(List.of()); + } + + @Test + void retriesTheWholeEntityAfterALaterChunkFails() { + when(postProcRepository.executeStoredProcForPatientIds("3,4")) + .thenThrow(new RuntimeException("chunk failure")) + .thenReturn(List.of()); + + service.idCache.put( + PostProcessingTestUtils.PATIENT_TOPIC, + new ConcurrentLinkedQueue<>(List.of(1L, 2L, 3L, 4L, 5L))); + + service.processCachedIds(); + + assertTrue( + service.retryCache.values().stream() + .anyMatch(batch -> batch.containsKey(PostProcessingTestUtils.PATIENT_TOPIC))); + verify(postProcRepository, times(2)).executeStoredProcForPatientIds(anyString()); + assertEquals(5.0, metricsRegistry.get("post_msg_failure").counter().count()); + assertEquals(0.0, metricsRegistry.get("post_msg_success").counter().count()); + + service.processRetryCache(); + + assertTrue(service.retryCache.isEmpty()); + verify(postProcRepository, times(5)).executeStoredProcForPatientIds(anyString()); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkingTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkingTest.java new file mode 100644 index 000000000..7b13672fc --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceChunkingTest.java @@ -0,0 +1,82 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; +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.kafka.RetryTopicResolver; +import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.kafka.core.KafkaTemplate; + +class PostProcessingServiceChunkingTest { + + @Mock private PostProcRepository postProcRepository; + @Mock private InvestigationRepository investigationRepository; + @Mock private KafkaTemplate kafkaTemplate; + + private PostProcessingService service; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + ProcessDatamartData datamartProcessor = + new ProcessDatamartData( + kafkaTemplate, + postProcRepository, + investigationRepository, + new PostProcessingProperties(0), + new CustomMetrics(new SimpleMeterRegistry())); + service = + new PostProcessingService( + postProcRepository, + investigationRepository, + datamartProcessor, + new RetryTopicResolver(), + new PostProcessingProperties(2), + new CustomMetrics(new SimpleMeterRegistry())); + PostProcessingTestUtils.configureNrtTopics(service); + service.initMetrics(); + datamartProcessor.initMetrics(); + service.setServiceEnable(true); + } + + @Test + void chunksNumericStoredProcedureInput() { + when(postProcRepository.executeStoredProcForPatientIds(anyString())).thenReturn(List.of()); + service.idCache.put( + PostProcessingTestUtils.PATIENT_TOPIC, + new ConcurrentLinkedQueue<>(List.of(1L, 2L, 3L, 4L, 5L))); + + service.processCachedIds(); + + ArgumentCaptor ids = ArgumentCaptor.forClass(String.class); + verify(postProcRepository, times(3)).executeStoredProcForPatientIds(ids.capture()); + assertEquals(List.of("1,2", "3,4", "5"), ids.getAllValues()); + } + + @Test + void chunksConditionCodeStoredProcedureInput() { + service.cdCache.put( + PostProcessingTestUtils.CONDITION_CODE_TOPIC, + new ConcurrentLinkedQueue<>(List.of("A", "B", "C", "D", "E"))); + + service.processCachedIds(); + + ArgumentCaptor codes = ArgumentCaptor.forClass(String.class); + verify(postProcRepository, times(3)).executeStoredProcForConditionCode(codes.capture()); + assertEquals(List.of("A,B", "C,D", "E"), codes.getAllValues()); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceDmTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceDmTest.java index 54ae1c14b..f1b138f6e 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceDmTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceDmTest.java @@ -8,6 +8,7 @@ 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.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.DatamartData; @@ -52,6 +53,7 @@ void setUp() { kafkaTemplate, postProcRepositoryMock, investigationRepositoryMock, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); postProcessingServiceMock = spy( @@ -60,6 +62,7 @@ void setUp() { investigationRepositoryMock, datamartProcessor, new RetryTopicResolver(), + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry()))); PostProcessingTestUtils.configureNrtTopics(postProcessingServiceMock); postProcessingServiceMock.initMetrics(); diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEntityTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEntityTest.java index fe1684d9e..3a868643c 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEntityTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceEntityTest.java @@ -11,6 +11,7 @@ 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.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.DatamartData; @@ -53,6 +54,7 @@ void setUp() { kafkaTemplate, postProcRepositoryMock, investigationRepositoryMock, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); postProcessingServiceMock = spy( @@ -61,6 +63,7 @@ void setUp() { investigationRepositoryMock, datamartProcessor, new RetryTopicResolver(), + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry()))); PostProcessingTestUtils.configureNrtTopics(postProcessingServiceMock); postProcessingServiceMock.initMetrics(); diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceRetryTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceRetryTest.java index 3813c16f5..02114ba27 100644 --- a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceRetryTest.java +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingServiceRetryTest.java @@ -9,6 +9,7 @@ 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.config.PostProcessingProperties; import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.BackfillData; @@ -53,6 +54,7 @@ void setUp() { kafkaTemplate, postProcRepositoryMock, investigationRepositoryMock, + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry())); postProcessingServiceMock = spy( @@ -61,6 +63,7 @@ void setUp() { investigationRepositoryMock, datamartProcessor, new RetryTopicResolver(), + new PostProcessingProperties(0), new CustomMetrics(new SimpleMeterRegistry()))); postProcessingServiceMock.setMaxRetries(2); PostProcessingTestUtils.configureNrtTopics(postProcessingServiceMock); diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataChunkingTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataChunkingTest.java new file mode 100644 index 000000000..de333ab21 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataChunkingTest.java @@ -0,0 +1,60 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; +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.metrics.CustomMetrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.kafka.core.KafkaTemplate; + +class ProcessDatamartDataChunkingTest { + + @Mock private KafkaTemplate kafkaTemplate; + @Mock private PostProcRepository postProcRepository; + @Mock private InvestigationRepository investigationRepository; + + private ProcessDatamartData datamartProcessor; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + datamartProcessor = + new ProcessDatamartData( + kafkaTemplate, + postProcRepository, + investigationRepository, + new PostProcessingProperties(2), + new CustomMetrics(new SimpleMeterRegistry())); + datamartProcessor.initMetrics(); + when(investigationRepository.executeStoredProcForStdHIVDatamart(anyString())) + .thenReturn(List.of()); + } + + @Test + void chunksSingleListDatamartProcedureInput() { + boolean processed = + datamartProcessor.processDmCache( + Map.of( + Entity.STD_HIV_DATAMART.getEntityName(), + Map.of(Entity.INVESTIGATION.getEntityName(), List.of(1L, 2L, 3L, 4L, 5L))), + null); + + assertTrue(processed); + ArgumentCaptor ids = ArgumentCaptor.forClass(String.class); + verify(investigationRepository, times(3)).executeStoredProcForStdHIVDatamart(ids.capture()); + org.junit.jupiter.api.Assertions.assertEquals(List.of("1,2", "3,4", "5"), ids.getAllValues()); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataEventMetricChunkingTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataEventMetricChunkingTest.java new file mode 100644 index 000000000..6403840d6 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataEventMetricChunkingTest.java @@ -0,0 +1,69 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static gov.cdc.nbs.report.pipeline.postprocessing.service.Entity.INVESTIGATION; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; +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.metrics.CustomMetrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.kafka.core.KafkaTemplate; + +class ProcessDatamartDataEventMetricChunkingTest { + + @Mock private KafkaTemplate kafkaTemplate; + @Mock private PostProcRepository postProcRepository; + @Mock private InvestigationRepository investigationRepository; + + private ProcessDatamartData datamartProcessor; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + datamartProcessor = + new ProcessDatamartData( + kafkaTemplate, + postProcRepository, + investigationRepository, + new PostProcessingProperties(2), + new CustomMetrics(new SimpleMeterRegistry())); + datamartProcessor.initMetrics(); + } + + @Test + void chunksEventMetricInputByEntityCategory() { + datamartProcessor.processMetricEventDatamart( + Map.of( + INVESTIGATION.getEntityName(), new ConcurrentLinkedQueue<>(List.of(1L, 2L, 3L, 4L)))); + + ArgumentCaptor investigationIds = ArgumentCaptor.forClass(String.class); + verify(postProcRepository, times(2)) + .executeStoredProcForEventMetric( + investigationIds.capture(), eq(""), eq(""), eq(""), eq("")); + assertEquals(List.of("1,2", "3,4"), investigationIds.getAllValues()); + } + + @Test + void keepsAllCategoriesInOneCallWhenTotalFitsLimit() { + datamartProcessor.processMetricEventDatamart( + Map.of( + INVESTIGATION.getEntityName(), + new ConcurrentLinkedQueue<>(List.of(1L)), + Entity.OBSERVATION.getEntityName(), + new ConcurrentLinkedQueue<>(List.of(2L)))); + + verify(postProcRepository).executeStoredProcForEventMetric("1", "2", "", "", ""); + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataMultiChunkingTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataMultiChunkingTest.java new file mode 100644 index 000000000..1967f2fd3 --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ProcessDatamartDataMultiChunkingTest.java @@ -0,0 +1,121 @@ +package gov.cdc.nbs.report.pipeline.postprocessing.service; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import gov.cdc.nbs.report.pipeline.config.PostProcessingProperties; +import gov.cdc.nbs.report.pipeline.postprocessing.repository.InvestigationRepository; +import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository; +import gov.cdc.nbs.report.pipeline.postprocessing.repository.model.DatamartData; +import gov.cdc.nbs.report.pipeline.util.metrics.CustomMetrics; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.springframework.kafka.core.KafkaTemplate; + +class ProcessDatamartDataMultiChunkingTest { + + @Mock private KafkaTemplate kafkaTemplate; + @Mock private PostProcRepository postProcRepository; + @Mock private InvestigationRepository investigationRepository; + + private ProcessDatamartData datamartProcessor; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + datamartProcessor = + new ProcessDatamartData( + kafkaTemplate, + postProcRepository, + investigationRepository, + new PostProcessingProperties(3), + new CustomMetrics(new SimpleMeterRegistry())); + datamartProcessor.initMetrics(); + } + + @Test + void chunksCovidVaccinationInputsWithoutDroppingEitherList() { + when(investigationRepository.executeStoredProcForCovidVacDatamart(anyString(), anyString())) + .thenReturn(List.of()); + + boolean processed = + datamartProcessor.processDmCache( + Map.of( + Entity.COVID_VACCINATION_DATAMART.getEntityName(), + Map.of( + Entity.INVESTIGATION.getEntityName(), List.of(1L, 2L, 3L), + Entity.PATIENT.getEntityName(), List.of(10L, 11L))), + null); + + assertTrue(processed); + ArgumentCaptor vaccinationIds = ArgumentCaptor.forClass(String.class); + ArgumentCaptor patientIds = ArgumentCaptor.forClass(String.class); + verify(investigationRepository, times(2)) + .executeStoredProcForCovidVacDatamart(vaccinationIds.capture(), patientIds.capture()); + assertEquals(List.of("1,2,3", ""), vaccinationIds.getAllValues()); + assertEquals(List.of("", "10,11"), patientIds.getAllValues()); + } + + @Test + void chunksMultiIdDatamartsBeforeProcessingDynamicOutputs() { + when(postProcRepository.executeStoredProcForInvSummaryDatamart( + anyString(), anyString(), anyString())) + .thenReturn(List.of()); + datamartProcessor.processDmCache( + Map.of( + ProcessDatamartData.MULTI_ID_DATAMART, + Map.of( + Entity.INVESTIGATION.getEntityName(), List.of(1L, 2L), + Entity.OBSERVATION.getEntityName(), List.of(3L), + Entity.NOTIFICATION.getEntityName(), List.of(4L), + Entity.PATIENT.getEntityName(), List.of(5L), + Entity.PROVIDER.getEntityName(), List.of(6L), + Entity.ORGANIZATION.getEntityName(), List.of(7L))), + null); + + verify(postProcRepository, times(2)) + .executeStoredProcForInvSummaryDatamart(anyString(), anyString(), anyString()); + verify(postProcRepository, times(2)) + .executeStoredProcForMorbidityReportDatamart( + anyString(), anyString(), anyString(), anyString(), anyString()); + } + + @Test + void chunksDynamicDatamartOutputsBeforeSubmittingAsyncCalls() { + List dynamicData = + List.of( + dynamicData(1L), dynamicData(2L), dynamicData(3L), dynamicData(4L), dynamicData(5L)); + when(postProcRepository.executeStoredProcForInvSummaryDatamart( + anyString(), anyString(), anyString())) + .thenReturn(dynamicData); + + datamartProcessor.processDmCache( + Map.of( + ProcessDatamartData.MULTI_ID_DATAMART, + Map.of(Entity.INVESTIGATION.getEntityName(), List.of(100L))), + null); + + ArgumentCaptor ids = ArgumentCaptor.forClass(String.class); + verify(postProcRepository, times(2)) + .executeStoredProcForDynDatamart(org.mockito.ArgumentMatchers.eq("DYN_DM"), ids.capture()); + org.junit.jupiter.api.Assertions.assertEquals( + List.of("1,2,3", "4,5"), ids.getAllValues().stream().sorted().toList()); + } + + private static DatamartData dynamicData(Long uid) { + DatamartData data = new DatamartData(); + data.setDatamart("DYN_DM"); + data.setPublicHealthCaseUid(uid); + return data; + } +} diff --git a/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunkerTest.java b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunkerTest.java new file mode 100644 index 000000000..3822dd01d --- /dev/null +++ b/reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/postprocessing/service/UidChunkerTest.java @@ -0,0 +1,69 @@ +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 java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class UidChunkerTest { + + @Test + void returnsNoChunksForEmptyInput() { + assertEquals(List.of(), UidChunker.chunkDistinct(List.of(), 2)); + } + + @Test + void removesDuplicatesBeforeCreatingOrderedChunks() { + List> chunks = UidChunker.chunkDistinct(List.of(3L, 1L, 3L, 2L, 1L), 2); + + assertEquals(List.of(List.of(3L, 1L), List.of(2L)), chunks); + } + + @Test + void returnsOneChunkWhenInputFitsLimit() { + assertEquals(List.of(List.of(1L, 2L, 3L)), UidChunker.chunkDistinct(List.of(1L, 2L, 3L), 3)); + } + + @Test + void createsFinalPartialChunkWhenInputExceedsLimit() { + assertEquals( + List.of(List.of(1L, 2L), List.of(3L, 4L), List.of(5L)), + UidChunker.chunkDistinct(List.of(1L, 2L, 3L, 4L, 5L), 2)); + } + + @Test + void zeroLimitDisablesChunkingButStillRemovesDuplicates() { + assertEquals( + List.of(List.of("a", "b", "c")), UidChunker.chunkDistinct(List.of("a", "b", "a", "c"), 0)); + } + + @Test + void chunksDistinctValuesAcrossOrderedMapEntries() { + Map> values = new LinkedHashMap<>(); + values.put("investigation", List.of(1L, 2L, 1L)); + values.put("observation", List.of(3L, 4L)); + + assertEquals( + List.of( + Map.of("investigation", List.of(1L, 2L), "observation", List.of(3L)), + Map.of("observation", List.of(4L))), + UidChunker.chunkDistinct(values, 3)); + } + + @Test + void returnsOneDistinctMapChunkWhenMapLimitIsZero() { + Map> values = new LinkedHashMap<>(); + values.put("investigation", List.of(1L, 1L)); + + assertEquals( + List.of(Map.of("investigation", List.of(1L))), UidChunker.chunkDistinct(values, 0)); + } + + @Test + void rejectsNegativeLimit() { + assertThrows(IllegalArgumentException.class, () -> UidChunker.chunkDistinct(List.of(1L), -1)); + } +} diff --git a/reporting-pipeline-service/src/test/resources/application-test.yaml b/reporting-pipeline-service/src/test/resources/application-test.yaml index 98ebebdb2..ff5fa5cee 100644 --- a/reporting-pipeline-service/src/test/resources/application-test.yaml +++ b/reporting-pipeline-service/src/test/resources/application-test.yaml @@ -41,3 +41,7 @@ connector: kafka-connect: enabled: true url: http://localhost:8083 + +service: + post-processing: + max-batch-size: 100