BFD-4656: Optimize stream usage and process EOBs in parallel (#3084)

Author: aschey-forpeople

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/beneficiary/BeneficiaryRepository.java

@@ -36,10 +36,10 @@ public List<BeneficiaryIdentity> getValidBeneficiaryIdentities(long beneXrefSk)
     return entityManager
         .createQuery(
             """
-            SELECT identity
-            FROM BeneficiaryIdentity identity
-            WHERE identity.id.xrefSk = :beneXrefSk
-            """,
+                SELECT identity
+                FROM BeneficiaryIdentity identity
+                WHERE identity.id.xrefSk = :beneXrefSk
+                """,
             BeneficiaryIdentity.class)
         .setParameter("beneXrefSk", beneXrefSk)
         .getResultList();
@@ -96,10 +96,10 @@ public Optional<Long> getXrefSkFromBeneSk(long beneSk) {
     return entityManager
         .createQuery(
             """
-              SELECT bene.xrefSk
-              FROM Beneficiary bene
-              WHERE bene.beneSk = :beneSk
-            """,
+                 SELECT bene.xrefSk
+                 FROM Beneficiary bene
+                 WHERE bene.beneSk = :beneSk
+               """,
             Long.class)
         .setParameter("beneSk", beneSk)
         .getResultList()
@@ -118,10 +118,10 @@ public Optional<Long> getXrefSkFromMbi(String mbi) {
     return entityManager
         .createQuery(
             """
-              SELECT bene.xrefSk
-              FROM Beneficiary bene
-              WHERE bene.identifier.mbi = :mbi
-            """,
+                  SELECT bene.xrefSk
+                  FROM Beneficiary bene
+                  WHERE bene.identifier.mbi = :mbi
+                """,
             Long.class)
         .setParameter("mbi", mbi)
         .getResultList()
@@ -168,6 +168,7 @@ public PatientMatchResult searchPatientMatch(PatientMatch patientMatch) {
 
       if (uniqueXrefs.size() == 1) {
         var matchedBene = benes.getFirst();
+        LogUtil.logBeneSk(matchedBene.getBeneSk());
         var finalDetermination =
             new FinalDetermination(combinationIndex, matchedRecords.getFirst());
         return new PatientMatchResult(

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/claim/ClaimAsyncService.java

@@ -5,6 +5,7 @@
 import gov.cms.bfd.server.ng.DbFilterParam;
 import gov.cms.bfd.server.ng.claim.model.*;
 import gov.cms.bfd.server.ng.input.ClaimSearchCriteria;
+import gov.cms.bfd.server.ng.util.LogUtil;
 import jakarta.persistence.EntityManager;
 import jakarta.persistence.PersistenceContext;
 import java.util.*;
@@ -46,7 +47,7 @@ CompletableFuture<Optional<C>> findByIdInClaimType(
             .getResultList()
             .stream()
             .findFirst();
-
+    result.ifPresent(claim -> LogUtil.logBeneSk(claim.getBeneficiary().getBeneSk()));
     return CompletableFuture.completedFuture(result);
   }
 
@@ -73,7 +74,9 @@ protected <T extends ClaimBase> CompletableFuture<List<T>> fetchClaims(
         DbFilterParam.withParams(entityManager.createQuery(jpql, claimClass), filters.params())
             .setParameter("beneSk", criteria.beneSk())
             .getResultList();
-
+    result.stream()
+        .findFirst()
+        .ifPresent(claim -> LogUtil.logBeneSk(claim.getBeneficiary().getBeneSk()));
     return CompletableFuture.completedFuture(result);
   }
 

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/claim/ClaimRepository.java

@@ -5,7 +5,6 @@
 import gov.cms.bfd.server.ng.claim.model.*;
 import gov.cms.bfd.server.ng.input.ClaimSearchCriteria;
 import gov.cms.bfd.server.ng.input.DateTimeRange;
-import gov.cms.bfd.server.ng.util.LogUtil;
 import io.micrometer.core.annotation.Timed;
 import io.micrometer.core.aop.MeterTag;
 import java.util.*;
@@ -23,42 +22,42 @@ public class ClaimRepository {
 
   private static final String CLAIM_PROFESSIONAL_SHARED_SYSTEMS =
       """
-              SELECT c
-              FROM ClaimProfessionalSharedSystems c
-              JOIN FETCH c.beneficiary b
-              LEFT JOIN FETCH c.claimItems cl
-            """;
+        SELECT c
+        FROM ClaimProfessionalSharedSystems c
+        JOIN FETCH c.beneficiary b
+        LEFT JOIN FETCH c.claimItems cl
+      """;
 
   private static final String CLAIM_PROFESSIONAL_NCH =
       """
-              SELECT c
-              FROM ClaimProfessionalNch c
-              JOIN FETCH c.beneficiary b
-              JOIN FETCH c.claimItems cl
-            """;
+        SELECT c
+        FROM ClaimProfessionalNch c
+        JOIN FETCH c.beneficiary b
+        JOIN FETCH c.claimItems cl
+      """;
 
   private static final String CLAIM_INSTITUTIONAL_SHARED_SYSTEMS =
       """
-              SELECT c
-              FROM ClaimInstitutionalSharedSystems c
-              JOIN FETCH c.beneficiary b
-              LEFT JOIN FETCH c.claimItems cl
-            """;
+        SELECT c
+        FROM ClaimInstitutionalSharedSystems c
+        JOIN FETCH c.beneficiary b
+        LEFT JOIN FETCH c.claimItems cl
+      """;
 
   private static final String CLAIM_INSTITUTIONAL_NCH =
       """
-              SELECT c
-              FROM ClaimInstitutionalNch c
-              JOIN FETCH c.beneficiary b
-              JOIN FETCH c.claimItems cl
-            """;
+        SELECT c
+        FROM ClaimInstitutionalNch c
+        JOIN FETCH c.beneficiary b
+        JOIN FETCH c.claimItems cl
+      """;
 
   private static final String CLAIM_RX =
       """
-              SELECT c
-              FROM ClaimRx c
-              JOIN FETCH c.beneficiary b
-            """;
+        SELECT c
+        FROM ClaimRx c
+        JOIN FETCH c.beneficiary b
+      """;
 
   private static final List<ClaimTypeDefinition> ALL_CLAIM_TYPES =
       List.of(
@@ -152,11 +151,7 @@ public Optional<ClaimBase> findById(
     institutionalNchClaims.join().ifPresent(allClaims::add);
     rxClaims.join().ifPresent(allClaims::add);
 
-    var result = allClaims.stream().findFirst();
-
-    result.ifPresent(claim -> LogUtil.logBeneSk(claim.getBeneficiary().getBeneSk()));
-
-    return result;
+    return allClaims.stream().findFirst();
   }
 
   /**

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/coverage/CoverageHandler.java

@@ -7,8 +7,8 @@
 import gov.cms.bfd.server.ng.loadprogress.LoadProgressRepository;
 import gov.cms.bfd.server.ng.util.FhirUtil;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
 import lombok.RequiredArgsConstructor;
 import org.hl7.fhir.r4.model.Bundle;
 import org.hl7.fhir.r4.model.Coverage;
@@ -75,7 +75,7 @@ public Bundle searchByBeneficiary(Long beneSk, DateTimeRange lastUpdated) {
             .searchBeneficiaryWithCoverage(beneSk, lastUpdated)
             .filter(b -> !b.isMergedBeneficiary());
     if (beneficiaryOpt.isEmpty()) {
-      return FhirUtil.bundleOrDefault(List.of(), loadProgressRepository::lastUpdated);
+      return FhirUtil.bundleOrDefault(Stream.of(), loadProgressRepository::lastUpdated);
     }
     var beneficiary = beneficiaryOpt.get();
     var coverages =

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/eob/EobHandler.java

@@ -96,14 +96,18 @@ public Bundle searchByBene(ClaimSearchCriteria criteria, SamhsaFilterMode samhsa
                   return claim.toFhir(securityStatus);
                 });
 
-    return FhirUtil.bundleOrDefault(filteredClaims.toList(), loadProgressRepository::lastUpdated);
+    return FhirUtil.bundleOrDefault(filteredClaims, loadProgressRepository::lastUpdated);
   }
 
   private Stream<? extends ClaimBase> filterSamhsaClaims(
       List<? extends ClaimBase> claims,
       SamhsaFilterMode samhsaFilterMode,
       ClaimSearchCriteria claimSearchCriteria) {
-    var claimStream = claims.stream().sorted(Comparator.comparing(ClaimBase::getClaimUniqueId));
+    // Process claims in parallel
+    // Note: DO NOT call toList() until the very end as materializing the list multiple times could
+    // negatively impact perf.
+    var claimStream =
+        claims.parallelStream().sorted(Comparator.comparing(ClaimBase::getClaimUniqueId));
     var filteredClaimStream =
         switch (samhsaFilterMode) {
           case INCLUDE -> claimStream;

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/input/FhirInputConverter.java

@@ -101,7 +101,7 @@ public static Optional<Integer> toIntOptional(@Nullable NumberParam numberParam)
     }
     try {
       return Optional.of(numberParam.getValue().intValueExact());
-    } catch (ArithmeticException ex) {
+    } catch (ArithmeticException _) {
       throw new InvalidRequestException("Numeric input was not in a valid format");
     }
   }
@@ -114,8 +114,12 @@ public static Optional<Integer> toIntOptional(@Nullable NumberParam numberParam)
    * @return ID
    */
   public static Long toLong(@Nullable ReferenceParam reference, String validResourceType) {
-    if (reference == null || reference.getIdPartAsLong() == null) {
-      throw new InvalidRequestException("Reference is missing");
+    try {
+      if (reference == null || reference.getIdPartAsLong() == null) {
+        throw new InvalidRequestException("Reference is missing");
+      }
+    } catch (NumberFormatException _) {
+      throw new InvalidRequestException("Reference is not a valid number");
     }
     var resourceType = reference.getResourceType();
     if (!StringUtils.isBlank(resourceType) && !resourceType.equals(validResourceType)) {

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/log/LogStreamAuditLogger.java

@@ -27,7 +27,7 @@ public void log(PatientMatchAuditRecord auditRecord) {
         LOGGER
             .atInfo()
             .setMessage(PATIENT_MATCH_REQUESTED)
-            .addKeyValue("logType", "patientMatchAudit")
+            .addKeyValue(LOG_TYPE, "patientMatchAudit")
             .addKeyValue(logKey(AUDIT_PREFIX, MATCHED_BENE_SK), matchedBeneSk.get().toString())
             .addKeyValue(
                 logKey(AUDIT_PREFIX, BENE_SKS_FOUND),

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/patient/PatientHandler.java

@@ -17,7 +17,6 @@
 import gov.cms.bfd.server.ng.util.*;
 import java.time.Instant;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Stream;
@@ -126,7 +125,7 @@ public Bundle searchByBeneficiaryC4DIC(Long beneSk) {
             .searchBeneficiaryWithCoverage(beneSk, new DateTimeRange())
             .filter(b -> !b.isMergedBeneficiary());
     if (beneficiaryOpt.isEmpty()) {
-      return FhirUtil.bundleOrDefault(List.of(), loadProgressRepository::lastUpdated);
+      return FhirUtil.bundleOrDefault(Stream.of(), loadProgressRepository::lastUpdated);
     }
     var beneficiary = beneficiaryOpt.get();
 

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/util/FhirUtil.java

@@ -1,7 +1,6 @@
 package gov.cms.bfd.server.ng.util;
 
 import java.time.ZonedDateTime;
-import java.util.List;
 import java.util.Optional;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
@@ -58,24 +57,13 @@ public static String getHcpcsSystem(String code) {
    * @return bundle
    */
   public static Bundle bundleOrDefault(
-      Stream<Resource> resources, Supplier<ZonedDateTime> batchLastUpdated) {
-    return bundleOrDefault(resources.toList(), batchLastUpdated);
-  }
+      Stream<? extends Resource> resources, Supplier<ZonedDateTime> batchLastUpdated) {
+    var bundle = getBundle(resources);
 
-  /**
-   * Creates a bundle from the resource, returning a default bundle with lastUpdated populated if
-   * empty.
-   *
-   * @param resources resources
-   * @param batchLastUpdated last updated
-   * @return bundle
-   */
-  public static Bundle bundleOrDefault(
-      List<? extends Resource> resources, Supplier<ZonedDateTime> batchLastUpdated) {
-    if (resources.isEmpty()) {
+    if (bundle.getEntry().isEmpty()) {
       return defaultBundle(batchLastUpdated);
     }
-    return getBundle(resources.stream());
+    return bundle;
   }
 
   /**
@@ -89,7 +77,7 @@ public static Bundle bundleOrDefault(
   public static Bundle bundleOrDefault(
       Optional<Resource> resource, Supplier<ZonedDateTime> batchLastUpdated) {
     return resource
-        .map(value -> bundleOrDefault(List.of(value), batchLastUpdated))
+        .map(value -> bundleOrDefault(Stream.of(value), batchLastUpdated))
         .orElseGet(() -> defaultBundle(batchLastUpdated));
   }
 

apps/bfd-server-ng/src/main/java/gov/cms/bfd/server/ng/util/LogUtil.java

@@ -1,5 +1,7 @@
 package gov.cms.bfd.server.ng.util;
 
+import static gov.cms.bfd.server.ng.util.LoggerConstants.LOG_TYPE;
+
 import org.slf4j.LoggerFactory;
 
 /** Utility class used for logging. */
@@ -20,8 +22,9 @@ private LogUtil() {
   public static void logBeneSk(Long beneSk) {
     LOGGER
         .atInfo()
-        .setMessage(LoggerConstants.BENE_SK_REQUESTED)
-        .addKeyValue("bene_sk", beneSk)
+        .setMessage(LoggerConstants.BENE_SK_FOUND)
+        .addKeyValue(LOG_TYPE, "beneFound")
+        .addKeyValue("beneSk", beneSk)
         .log();
   }
 }