CNDIT 1703: Rollback Investigation Service loop debug changes

.github/workflows/Build-and-deploy-reporting-services.yaml

@@ -6,7 +6,7 @@ on:
       - master
       - rel-**
 #       Uncomment the following line only to test the build deploy from private branches.
-      - CNDIT-*
+#      - CNDIT-*
     paths-ignore:
       - "docker-compose.yml"
       - "**.md"

investigation-service/src/main/java/gov/cdc/etldatapipeline/investigation/service/InvestigationService.java

@@ -104,14 +104,12 @@ public String processInvestigation(String value) {
                     InvestigationReporting reportingModel = modelMapper.map(investigation, InvestigationReporting.class);
                     InvestigationTransformed investigationTransformed = processDataUtil.transformInvestigationData(investigation);
                     buildReportingModelForTransformedData(reportingModel, investigationTransformed);
-                    pushKeyValuePairToKafka(investigationKey, reportingModel, investigationTopicReporting);
-
-                    //ToDo: Revert this after the loop debugging issue
-                    // only process and send notifications when investigation data has been sent
-                      /*  .whenComplete((res, ex) ->
+                    pushKeyValuePairToKafka(investigationKey, reportingModel, investigationTopicReporting)
+                        // only process and send notifications when investigation data has been sent
+                        .whenComplete((res, ex) ->
                             logger.info("Investigation data (uid={}) sent to {}", investigation.getPublicHealthCaseUid(), investigationTopicReporting))
                         .thenRunAsync(() -> processDataUtil.processNotifications(investigation.getInvestigationNotifications(), objectMapper))
-                        .join();*/
+                        .join();
                     return objectMapper.writeValueAsString(investigation);
                 } else {
                     throw new NoDataException("No investigation data found for id: " + publicHealthCaseUid);
@@ -141,10 +139,10 @@ private void processPhcFactDatamart(String publicHealthCaseUid) {
     }
 
     // This same method can be used for elastic search as well and that is why the generic model is present
-    private void pushKeyValuePairToKafka(InvestigationKey investigationKey, Object model, String topicName) {
+    private CompletableFuture<SendResult<String, String>> pushKeyValuePairToKafka(InvestigationKey investigationKey, Object model, String topicName) {
         String jsonKey = jsonGenerator.generateStringJson(investigationKey);
         String jsonValue = jsonGenerator.generateStringJson(model);
-         kafkaTemplate.send(topicName, jsonKey, jsonValue);
+        return kafkaTemplate.send(topicName, jsonKey, jsonValue);
     }
 
     private void buildReportingModelForTransformedData(InvestigationReporting reportingModel, InvestigationTransformed investigationTransformed) {

investigation-service/src/main/java/gov/cdc/etldatapipeline/investigation/util/ProcessInvestigationDataUtil.java

@@ -68,9 +68,8 @@ public void processNotifications(String investigationNotifications, ObjectMapper
 
                     String jsonKey = jsonGenerator.generateStringJson(investigationNotificationsKey);
                     String jsonValue = jsonGenerator.generateStringJson(tempInvestigationNotificationsObject);
-                    // ToDo: Revert this
-//                    kafkaTemplate.send(investigationNotificationsOutputTopicName, jsonKey, jsonValue)
-//                            .whenComplete((res, e) -> logger.info("Notification data (uid={}) sent to {}", notificationUid, investigationNotificationsOutputTopicName));
+                    kafkaTemplate.send(investigationNotificationsOutputTopicName, jsonKey, jsonValue)
+                            .whenComplete((res, e) -> logger.info("Notification data (uid={}) sent to {}", notificationUid, investigationNotificationsOutputTopicName));
                 }
             }
             else {
@@ -187,8 +186,7 @@ private void transformObservationIds(String observationNotificationIds, Investig
                 for(Long id : observationIds) {
                     investigationObservation.setObservationId(id);
                     String jsonValue = jsonGenerator.generateStringJson(investigationObservation);
-                    //ToDo: Revert this after the loop debugging issue
-//                    kafkaTemplate.send(investigationObservationOutputTopicName, jsonValue, jsonValue);
+                    kafkaTemplate.send(investigationObservationOutputTopicName, jsonValue, jsonValue);
                 }
             }
             else {
@@ -244,8 +242,7 @@ private void transformInvestigationConfirmationMethod(String investigationConfir
                     investigationConfirmationMethodKey.setConfirmationMethodCd(key);
                     String jsonKey = jsonGenerator.generateStringJson(investigationConfirmationMethodKey);
                     String jsonValue = jsonGenerator.generateStringJson(investigationConfirmation);
-                    //ToDo: Revert this after the loop debugging issue
-//                    kafkaTemplate.send(investigationConfirmationOutputTopicName, jsonKey, jsonValue);
+                    kafkaTemplate.send(investigationConfirmationOutputTopicName, jsonKey, jsonValue);
                 }
             }
             else {

investigation-service/src/test/java/gov/cdc/etldatapipeline/investigation/InvestigationDataProcessingTests.java

@@ -6,7 +6,6 @@
 import gov.cdc.etldatapipeline.investigation.repository.model.dto.*;
 import gov.cdc.etldatapipeline.investigation.repository.rdb.InvestigationCaseAnswerRepository;
 import gov.cdc.etldatapipeline.investigation.util.ProcessInvestigationDataUtil;
-import org.junit.Ignore;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
@@ -28,7 +27,6 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.*;
 
-@Ignore
 class InvestigationDataProcessingTests {
     @Mock
     KafkaTemplate<String, String> kafkaTemplate;
@@ -65,8 +63,7 @@ void setUp() {
         transformer = new ProcessInvestigationDataUtil(kafkaTemplate, investigationCaseAnswerRepository);
     }
 
-    @Ignore
-    //@Test
+    @Test
     void testConfirmationMethod() {
         Investigation investigation = new Investigation();
 
@@ -102,7 +99,7 @@ void testConfirmationMethod() {
         assertTrue(containsWords.apply(messageCaptor.getValue(), cmDetailsFn.apply(confirmationMethod)));
     }
 
-    //@Test
+    @Test
     void testObservationNotificationIds() {
         Investigation investigation = new Investigation();
 
@@ -127,7 +124,7 @@ void testObservationNotificationIds() {
         assertTrue(containsWords.apply(actualCombined, oDetailsFn.apply(observation)));
     }
 
-   // @Test
+    @Test
     void testNotifications() {
         Investigation investigation = new Investigation();
 

investigation-service/src/test/java/gov/cdc/etldatapipeline/investigation/service/InvestigationServiceTest.java

@@ -8,7 +8,6 @@
 import gov.cdc.etldatapipeline.investigation.repository.model.reporting.InvestigationReporting;
 import gov.cdc.etldatapipeline.investigation.repository.rdb.InvestigationCaseAnswerRepository;
 import gov.cdc.etldatapipeline.investigation.util.ProcessInvestigationDataUtil;
-import org.junit.Ignore;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
@@ -24,7 +23,6 @@
 import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
-@Ignore
 class InvestigationServiceTest {
 
     @Mock
@@ -54,7 +52,7 @@ void setUp() {
         transformer = new ProcessInvestigationDataUtil(kafkaTemplate, investigationCaseAnswerRepository);
     }
 
-   // @Test
+    @Test
     void testProcessMessage() {
         String investigationTopic = "Investigation";
         String investigationTopicOutput = "InvestigationOutput";