https://github.com/gbif/pipelines/issues/1078

adding data warehouse message processing

Author: fmendezh

pom.xml

@@ -98,7 +98,7 @@
     <gbif-httputils.version>0.10</gbif-httputils.version>
     <gbif-metadata-profile.version>2.2.1-SNAPSHOT</gbif-metadata-profile.version>
     <gbif-parsers.version>0.67</gbif-parsers.version>
-    <gbif-postal-service.version>1.8.0</gbif-postal-service.version>
+    <gbif-postal-service.version>1.9.0-SNAPSHOT</gbif-postal-service.version>
     <xoai.version>4.1.0-gbif-4</xoai.version>
     <gbif-occurrence.version>0.195.1</gbif-occurrence.version>
     <gbif-vocabulary.version>1.0.12</gbif-vocabulary.version>

registry-pipelines/src/main/java/org/gbif/registry/pipelines/DefaultRegistryPipelinesHistoryTrackingService.java

@@ -459,6 +459,7 @@ private Optional<? extends PipelineBasedMessage> createStepMessage(StepType step
     switch (stepName) {
       case INTERPRETED_TO_INDEX:
       case HDFS_VIEW:
+      case DATA_WAREHOUSE:
       case FRAGMENTER:
         return createInterpretedMessage(prefix, jsonMessage, stepName);
       case VERBATIM_TO_INTERPRETED:
@@ -487,7 +488,7 @@ protected Set<StepType> getStepTypes(Set<StepType> stepsToSend) {
     if (stepsToSend.stream().anyMatch(StepType::isOccurrenceType)) {
       finalSteps.addAll(PipelinesWorkflow.getOccurrenceWorkflow().getAllNodesFor(stepsToSend));
     }
-    // TODO: Events from dwca will be ignorred, add dataset type?
+    // TODO: Events from dwca will be ignored, add dataset type?
     if (stepsToSend.stream().anyMatch(StepType::isVerbatimType)) {
       finalSteps.addAll(PipelinesWorkflow.getOccurrenceWorkflow().getAllNodesFor(stepsToSend));
     }
@@ -523,12 +524,14 @@ private Optional<PipelineBasedMessage> createVerbatimMessage(
       steps.add(StepType.VERBATIM_TO_INTERPRETED.name());
       steps.add(StepType.INTERPRETED_TO_INDEX.name());
       steps.add(StepType.HDFS_VIEW.name());
+      steps.add(StepType.DATA_WAREHOUSE.name());
     }
 
     if (message.getPipelineSteps().contains(StepType.EVENTS_VERBATIM_TO_INTERPRETED.name())) {
       steps.add(StepType.EVENTS_VERBATIM_TO_INTERPRETED.name());
       steps.add(StepType.EVENTS_INTERPRETED_TO_INDEX.name());
       steps.add(StepType.EVENTS_HDFS_VIEW.name());
+      steps.add(StepType.DATA_WAREHOUSE.name());
     }
 
     message.setPipelineSteps(steps);