CNDE-1834: Fix persistence for multivalued observation payloads (#62)

CDCgov · Oct 18, 2024 · 3254c5e · 3254c5e
1 parent 4a5d514
commit 3254c5e
Show file tree

Hide file tree

Showing 4 changed files with 54 additions and 6 deletions.
diff --git a/...a/gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationCodedKey.java b/...a/gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationCodedKey.java
@@ -0,0 +1,14 @@
+package gov.cdc.etldatapipeline.observation.repository.model.reporting;
+
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+public class ObservationCodedKey {
+    private Long observationUid;
+    private String ovcCode;
+}
diff --git a/.../gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationReasonKey.java b/.../gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationReasonKey.java
@@ -0,0 +1,14 @@
+package gov.cdc.etldatapipeline.observation.repository.model.reporting;
+
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+public class ObservationReasonKey {
+    private Long observationUid;
+    private String reasonCd;
+}
diff --git a/...ava/gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationTxtKey.java b/...ava/gov/cdc/etldatapipeline/observation/repository/model/reporting/ObservationTxtKey.java
@@ -0,0 +1,14 @@
+package gov.cdc.etldatapipeline.observation.repository.model.reporting;
+
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+public class ObservationTxtKey {
+    private Long observationUid;
+    private Integer ovtSeq;
+}
diff --git a/...ce/src/main/java/gov/cdc/etldatapipeline/observation/util/ProcessObservationDataUtil.java b/...ce/src/main/java/gov/cdc/etldatapipeline/observation/util/ProcessObservationDataUtil.java
@@ -323,10 +323,12 @@ private void transformObservationCoded(String observationCoded) {
         try {
             JsonNode observationCodedJsonArray = parseJsonArray(observationCoded);
 
+            ObservationCodedKey codedKey = new ObservationCodedKey();
             for (JsonNode jsonNode : observationCodedJsonArray) {
                 ObservationCoded coded = objectMapper.treeToValue(jsonNode, ObservationCoded.class);
-                observationKey.setObservationUid(coded.getObservationUid());
-                sendToKafka(observationKey, coded, codedTopicName, coded.getObservationUid(), "Observation Coded data (uid={}) sent to {}");
+                codedKey.setObservationUid(coded.getObservationUid());
+                codedKey.setOvcCode(coded.getOvcCode());
+                sendToKafka(codedKey, coded, codedTopicName, coded.getObservationUid(), "Observation Coded data (uid={}) sent to {}");
             }
         } catch (IllegalArgumentException ex) {
             logger.info(ex.getMessage(), "ObservationCoded");
@@ -388,10 +390,12 @@ private void transformObservationReasons(String observationReasons) {
         try {
             JsonNode observationReasonsJsonArray = parseJsonArray(observationReasons);
 
+            ObservationReasonKey reasonKey = new ObservationReasonKey();
             for (JsonNode jsonNode : observationReasonsJsonArray) {
                 ObservationReason reason = objectMapper.treeToValue(jsonNode, ObservationReason.class);
-                observationKey.setObservationUid(reason.getObservationUid());
-                sendToKafka(observationKey, reason, reasonTopicName, reason.getObservationUid(), "Observation Reason data (uid={}) sent to {}");
+                reasonKey.setObservationUid(reason.getObservationUid());
+                reasonKey.setReasonCd(reason.getReasonCd());
+                sendToKafka(reasonKey, reason, reasonTopicName, reason.getObservationUid(), "Observation Reason data (uid={}) sent to {}");
             }
         } catch (IllegalArgumentException ex) {
             logger.info(ex.getMessage(), "ObservationReasons");
@@ -404,10 +408,12 @@ private void transformObservationTxt(String observationTxt) {
         try {
             JsonNode observationTxtJsonArray = parseJsonArray(observationTxt);
 
+            ObservationTxtKey txtKey = new ObservationTxtKey();
             for (JsonNode jsonNode : observationTxtJsonArray) {
                 ObservationTxt txt = objectMapper.treeToValue(jsonNode, ObservationTxt.class);
-                observationKey.setObservationUid(txt.getObservationUid());
-                sendToKafka(observationKey, txt, txtTopicName, txt.getObservationUid(), "Observation Txt data (uid={}) sent to {}");
+                txtKey.setObservationUid(txt.getObservationUid());
+                txtKey.setOvtSeq(txt.getOvtSeq());
+                sendToKafka(txtKey, txt, txtTopicName, txt.getObservationUid(), "Observation Txt data (uid={}) sent to {}");
             }
         } catch (IllegalArgumentException ex) {
             logger.info(ex.getMessage(), "ObservationTxt");