Fix Nullpointer Exception in ReferenceResolve

Author: Lucas0T

src/main/java/de/medizininformatikinitiative/torch/config/AppConfig.java

@@ -33,6 +33,7 @@
 import de.medizininformatikinitiative.torch.util.Redaction;
 import de.medizininformatikinitiative.torch.util.ReferenceExtractor;
 import de.medizininformatikinitiative.torch.util.ReferenceHandler;
+import de.medizininformatikinitiative.torch.util.ResourceGroupValidator;
 import de.medizininformatikinitiative.torch.util.ResourceReader;
 import de.medizininformatikinitiative.torch.util.ResultFileManager;
 import de.numcodex.sq2cql.Translator;
@@ -126,8 +127,13 @@ public ProcessedGroupFactory attributeGroupProcessor(CompartmentManager manager)
 
 
     @Bean
-    ReferenceHandler referenceHandler(DataStore dataStore, ProfileMustHaveChecker mustHaveChecker, CompartmentManager compartmentManager, ConsentValidator consentValidator) {
-        return new ReferenceHandler(mustHaveChecker);
+    ReferenceHandler referenceHandler(ResourceGroupValidator resourceGroupValidator) {
+        return new ReferenceHandler(resourceGroupValidator);
+    }
+
+    @Bean
+    ResourceGroupValidator resourceGroupValidator(ProfileMustHaveChecker profileMustHaveChecker) {
+        return new ResourceGroupValidator(profileMustHaveChecker);
     }
 
     @Bean

src/main/java/de/medizininformatikinitiative/torch/exceptions/ResourceTypeMissmatchException.java

@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 @Service
 public class CrtdlProcessingService {
@@ -117,14 +118,15 @@ private Mono<Void> processIntern(AnnotatedCrtdl crtdl, String jobID, Flux<Patien
         // Step 2: Process Patient Batches Using Preprocessed Core Bundle
         logger.debug("Process patient batches with a concurrency of {}", maxConcurrency);
         return preProcessedCoreBundle.flatMapMany(coreSnapshot ->
-                batches
-                        .flatMap(batch -> crtdl.consentKey()
-                                        .map(s -> consentHandler.fetchAndBuildConsentInfo(s, batch))
-                                        .orElse(Mono.just(PatientBatchWithConsent.fromBatch(batch)))
-                                        .onErrorResume(ConsentViolatedException.class, ex -> Mono.empty())
-                                , maxConcurrency) //skip batches without consenting patient
-                        .doOnNext(patientBatch -> patientBatch.addStaticInfo(coreSnapshot))
-                        .flatMap(batch -> processBatch(batch, jobID, groupsToProcess, coreBundle), maxConcurrency)
+                batches.flatMap(batch ->
+                        processBatch(
+                                batch,
+                                jobID,
+                                groupsToProcess,
+                                coreBundle,
+                                crtdl.consentKey(),
+                                coreSnapshot
+                        ), maxConcurrency)
         ).then(
                 // Step 3: Write the Final Core Resource Bundle to File
                 Mono.defer(() -> {
@@ -135,15 +137,34 @@ private Mono<Void> processIntern(AnnotatedCrtdl crtdl, String jobID, Flux<Patien
         );
     }
 
-    private Mono<Void> processBatch(PatientBatchWithConsent batch, String jobID, GroupsToProcess groupsToProcess, ResourceBundle coreBundle) {
-        return directResourceLoader.directLoadPatientCompartment(groupsToProcess.directPatientCompartmentGroups(), batch)
-                .flatMap(patientBatch -> referenceResolver.processSinglePatientBatch(patientBatch, coreBundle, groupsToProcess.allGroups()))
-                .map(patientBatch -> cascadingDelete.handlePatientBatch(patientBatch, groupsToProcess.allGroups()))
-                .map(patientBatch -> batchCopierRedacter.transformBatch(patientBatch, groupsToProcess.allGroups()))
-                .flatMap(patientBatch -> {
-                            batchToCoreWriter.updateCore(patientBatch, coreBundle);
-                            return writeBatch(jobID, patientBatch);
-                        }
+    private Mono<Void> processBatch(
+            PatientBatch batch,
+            String jobID,
+            GroupsToProcess groupsToProcess,
+            ResourceBundle coreBundle,
+            Optional<String> consentKey,
+            CachelessResourceBundle coreSnapshot
+    ) {
+        // Fetch consent (or assume consent if key is empty)
+        Mono<PatientBatchWithConsent> withConsent = consentKey
+                .map(key -> consentHandler.fetchAndBuildConsentInfo(key, batch))
+                .orElse(Mono.just(PatientBatchWithConsent.fromBatch(batch)))
+                .onErrorResume(ConsentViolatedException.class, ex -> {
+                    logger.debug("Skipping batch due to consent violation: {}", ex.getMessage());
+                    return Mono.empty();
+                });
+
+        return withConsent
+                .doOnNext(b -> b.addStaticInfo(coreSnapshot))
+                .flatMap(consented ->
+                        directResourceLoader.directLoadPatientCompartment(groupsToProcess.directPatientCompartmentGroups(), consented)
+                                .flatMap(loaded -> referenceResolver.processSinglePatientBatch(loaded, coreBundle, groupsToProcess.allGroups()))
+                                .map(processed -> cascadingDelete.handlePatientBatch(processed, groupsToProcess.allGroups()))
+                                .map(transformed -> batchCopierRedacter.transformBatch(transformed, groupsToProcess.allGroups()))
+                                .flatMap(finalBatch -> {
+                                    batchToCoreWriter.updateCore(finalBatch, coreBundle);
+                                    return writeBatch(jobID, finalBatch);
+                                })
                 );
     }
 

src/main/java/de/medizininformatikinitiative/torch/service/CrtdlProcessingService.java

@@ -67,9 +67,7 @@ public Mono<ResourceBundle> resolveCoreBundle(ResourceBundle coreBundle, Map<Str
                         .collect(Collectors.toSet()))
                 .expand(currentGroupSet ->
                         processResourceGroups(currentGroupSet, null, coreBundle, false, groupMap)
-                                .onErrorResume(e -> {
-                                    return Mono.empty(); // Skip this resource group on error
-                                }))
+                                .onErrorResume(e -> Mono.empty()))
                 .then(Mono.just(coreBundle));
     }
 
@@ -144,26 +142,20 @@ public Mono<Set<ResourceGroup>> processResourceGroups(
         return bundleLoader.fetchUnknownResources(referencesGroupedByResourceGroup, patientBundle, coreBundle, applyConsent)
                 .thenMany(
                         Flux.fromIterable(referencesGroupedByResourceGroup.entrySet())
+                                .filter(entry -> entry.getKey() != null && entry.getValue() != null && !entry.getValue().isEmpty())
                                 .concatMap(entry ->
-                                        {
-                                            try {
-                                                return referenceHandler.handleReferences(
-                                                        entry.getValue(),
-                                                        patientBundle,
-                                                        coreBundle,
-                                                        groupMap
-                                                );
-                                            } catch (MustHaveViolatedException e) {
-                                                return Flux.empty();
-                                            }
-                                        }
+                                        Flux.fromIterable(referenceHandler.handleReferences(
+                                                entry.getValue(),
+                                                patientBundle,
+                                                coreBundle,
+                                                groupMap
+                                        ))
                                 )
                 )
                 .collect(Collectors.toSet())
                 .flatMap(set -> set.isEmpty() ? Mono.empty() : Mono.just(set));
     }
 
-
     /**
      * Extracts for every ResourceGroup the ReferenceWrappers and collects them ordered by
      *
@@ -179,9 +171,9 @@ public Map<ResourceGroup, List<ReferenceWrapper>> loadReferencesByResourceGroup(
             ResourceBundle coreBundle,
             Map<String, AnnotatedAttributeGroup> groupMap) {
 
-        return resourceGroups.parallelStream()
+        return resourceGroups.stream()
                 .map(resourceGroup -> processResourceGroup(resourceGroup, patientBundle, coreBundle, groupMap))
-                .filter(entry -> !entry.getValue().isEmpty())
+                .filter(entry -> entry.getKey() != null && entry.getValue() != null && !entry.getValue().isEmpty())
                 .collect(Collectors.toMap(
                         Map.Entry::getKey,
                         Map.Entry::getValue,
@@ -219,11 +211,7 @@ private Map.Entry<ResourceGroup, List<ReferenceWrapper>> processResourceGroup(
                 ? patientBundle.bundle().get(resourceGroup.resourceId())
                 : coreBundle.get(resourceGroup.resourceId());
 
-        if (resource.isPresent()) {
-            return extractReferences(resourceGroup, resource.get(), groupMap, processingBundle);
-        } else {
-            return handleMissingResource(resourceGroup, processingBundle);
-        }
+        return resource.map(value -> extractReferences(resourceGroup, value, groupMap, processingBundle)).orElseGet(() -> handleMissingResource(resourceGroup, processingBundle));
     }
 
     /**
@@ -253,9 +241,7 @@ private Map.Entry<ResourceGroup, List<ReferenceWrapper>> extractReferences(
             List<ReferenceWrapper> extracted = referenceExtractor.extract(resource, groupMap, resourceGroup.groupId());
             return Map.entry(resourceGroup, extracted);
         } catch (MustHaveViolatedException e) {
-            synchronized (processingBundle) {
-                processingBundle.addResourceGroupValidity(resourceGroup, false);
-            }
+            processingBundle.addResourceGroupValidity(resourceGroup, false);
             return Map.entry(resourceGroup, Collections.emptyList());
         }
     }
@@ -270,11 +256,8 @@ private Map.Entry<ResourceGroup, List<ReferenceWrapper>> extractReferences(
     private Map.Entry<ResourceGroup, List<ReferenceWrapper>> handleMissingResource(
             ResourceGroup resourceGroup,
             ResourceBundle processingBundle) {
-
-        synchronized (processingBundle) {
-            logger.warn("Empty resource marked as valid for group {}", resourceGroup);
-            processingBundle.addResourceGroupValidity(resourceGroup, false);
-        }
+        logger.warn("Empty resource marked as valid for group {}", resourceGroup);
+        processingBundle.addResourceGroupValidity(resourceGroup, false);
         return Map.entry(resourceGroup, Collections.emptyList());
     }