CTHLU-56 - clean up kafka topic config

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/investigation/controller/InvestigationController.java

@@ -12,22 +12,22 @@
 public class InvestigationController {
   private final KafkaProducerService producerService;
 
-  @Value("${spring.kafka.input.topic-name-phc}")
+  @Value("${spring.kafka.topics.nbs.public-health-case}")
   private String investigationTopic;
 
-  @Value("${spring.kafka.input.topic-name-ntf}")
+  @Value("${spring.kafka.topics.nbs.notification}")
   private String notificationTopic;
 
-  @Value("${spring.kafka.input.topic-name-int}")
+  @Value("${spring.kafka.topics.nbs.interview}")
   private String interviewTopic;
 
-  @Value("${spring.kafka.input.topic-name-ctr}")
+  @Value("${spring.kafka.topics.nbs.ct-contact}")
   private String contactTopic;
 
-  @Value("${spring.kafka.input.topic-name-vac}")
+  @Value("${spring.kafka.topics.nbs.intervention}")
   private String vaccinationTopic;
 
-  @Value("${spring.kafka.input.topic-name-tmt}")
+  @Value("${spring.kafka.topics.nbs.treatment}")
   private String treatmentTopic;
 
   @PostMapping("/reporting/investigation-svc/investigation")

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

@@ -51,31 +51,31 @@ public class InvestigationService {
       Executors.newFixedThreadPool(nProc * 2, new CustomizableThreadFactory("phc-"));
   private ExecutorService invExecutor;
 
-  @Value("${spring.kafka.input.topic-name-phc}")
+  @Value("${spring.kafka.topics.nbs.public-health-case}")
   private String investigationTopic;
 
-  @Value("${spring.kafka.input.topic-name-ntf}")
+  @Value("${spring.kafka.topics.nbs.notification}")
   private String notificationTopic;
 
-  @Value("${spring.kafka.input.topic-name-int}")
+  @Value("${spring.kafka.topics.nbs.interview}")
   private String interviewTopic;
 
-  @Value("${spring.kafka.input.topic-name-ctr}")
+  @Value("${spring.kafka.topics.nbs.ct-contact}")
   private String contactTopic;
 
-  @Value("${spring.kafka.input.topic-name-vac}")
+  @Value("${spring.kafka.topics.nbs.intervention}")
   private String vaccinationTopic;
 
-  @Value("${spring.kafka.input.topic-name-tmt}")
+  @Value("${spring.kafka.topics.nbs.treatment}")
   private String treatmentTopic;
 
-  @Value("${spring.kafka.input.topic-name-ar}")
+  @Value("${spring.kafka.topics.nbs.act-relationship}")
   private String actRelationshipTopic;
 
-  @Value("${spring.kafka.output.topic-name-treatment}")
+  @Value("${spring.kafka.topics.nrt.treatment}")
   private String treatmentOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-reporting}")
+  @Value("${spring.kafka.topics.nrt.investigation}")
   private String investigationTopicReporting;
 
   @Value("${featureFlag.phc-datamart-disable}")
@@ -147,13 +147,13 @@ void initMetrics() {
       kafkaTemplate = "investigationKafkaTemplate")
   @KafkaListener(
       topics = {
-        "${spring.kafka.input.topic-name-phc}",
-        "${spring.kafka.input.topic-name-ntf}",
-        "${spring.kafka.input.topic-name-int}",
-        "${spring.kafka.input.topic-name-ctr}",
-        "${spring.kafka.input.topic-name-vac}",
-        "${spring.kafka.input.topic-name-tmt}",
-        "${spring.kafka.input.topic-name-ar}"
+        "${spring.kafka.topics.nbs.public-health-case}",
+        "${spring.kafka.topics.nbs.notification}",
+        "${spring.kafka.topics.nbs.interview}",
+        "${spring.kafka.topics.nbs.ct-contact}",
+        "${spring.kafka.topics.nbs.intervention}",
+        "${spring.kafka.topics.nbs.treatment}",
+        "${spring.kafka.topics.nbs.act-relationship}"
       },
       containerFactory = "investigationKafkaListenerContainerFactory")
   public CompletableFuture<Void> processMessage(ConsumerRecord<String, String> rec) {

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

@@ -34,46 +34,46 @@ public class ProcessInvestigationDataUtil {
       new ObjectMapper().registerModule(new JavaTimeModule());
   public static final String ENTITY_ID = "entity_id";
 
-  @Value("${spring.kafka.output.topic-name-confirmation}")
+  @Value("${spring.kafka.topics.nrt.investigation-confirmation}")
   public String investigationConfirmationOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-observation}")
+  @Value("${spring.kafka.topics.nrt.investigation-observation}")
   public String investigationObservationOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-notifications}")
+  @Value("${spring.kafka.topics.nrt.investigation-notification}")
   public String investigationNotificationsOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-page-case-answer}")
+  @Value("${spring.kafka.topics.nrt.page-case-answer}")
   public String pageCaseAnswerOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-aggregate}")
+  @Value("${spring.kafka.topics.nrt.investigation-aggregate}")
   public String investigationAggregateOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-case-management}")
+  @Value("${spring.kafka.topics.nrt.investigation-case-management}")
   public String investigationCaseManagementTopicName;
 
-  @Value("${spring.kafka.output.topic-name-interview}")
+  @Value("${spring.kafka.topics.nrt.interview}")
   private String interviewOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-interview-answer}")
+  @Value("${spring.kafka.topics.nrt.interview-answer}")
   private String interviewAnswerOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-interview-note}")
+  @Value("${spring.kafka.topics.nrt.interview-note}")
   private String interviewNoteOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-contact}")
+  @Value("${spring.kafka.topics.nrt.contact}")
   private String contactOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-contact-answer}")
+  @Value("${spring.kafka.topics.nrt.contact-answer}")
   private String contactAnswerOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-vaccination}")
+  @Value("${spring.kafka.topics.nrt.vaccination}")
   private String vaccinationOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-vaccination-answer}")
+  @Value("${spring.kafka.topics.nrt.vaccination-answer}")
   private String vaccinationAnswerOutputTopicName;
 
-  @Value("${spring.kafka.output.topic-name-rdb-metadata-columns}")
+  @Value("${spring.kafka.topics.nrt.metadata-columns}")
   private String rdbMetadataColumnsOutputTopicName;
 
   @Qualifier("investigationKafkaTemplate")

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/ldfdata/controller/LdfDataController.java

@@ -12,7 +12,7 @@
 public class LdfDataController {
   private final KafkaProducerService producerService;
 
-  @Value("${spring.kafka.input.topic-name-sdfd}")
+  @Value("${spring.kafka.topics.nbs.state-defined-field-data}")
   private String topicName;
 
   @PostMapping("/reporting/ldfdata-svc/publish")

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/ldfdata/service/LdfDataService.java

@@ -47,10 +47,10 @@ public class LdfDataService {
       new ObjectMapper().registerModule(new JavaTimeModule());
   private ExecutorService ldfExecutor;
 
-  @Value("${spring.kafka.input.topic-name-sdfd}")
+  @Value("${spring.kafka.topics.nbs.state-defined-field-data}")
   private String ldfDataTopic;
 
-  @Value("${spring.kafka.output.topic-name-reporting-ldf-data}")
+  @Value("${spring.kafka.topics.nrt.ldf-data}")
   public String ldfDataTopicReporting;
 
   @Value("${featureFlag.thread-pool-size:1}")
@@ -106,7 +106,7 @@ void initMetrics() {
       },
       kafkaTemplate = "ldfdataKafkaTemplate")
   @KafkaListener(
-      topics = "${spring.kafka.input.topic-name-sdfd}",
+      topics = "${spring.kafka.topics.nbs.state-defined-field-data}",
       containerFactory = "ldfdataKafkaListenerContainerFactory")
   public CompletableFuture<Void> processMessage(ConsumerRecord<String, String> rec) {
     String topic = rec.topic();

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/observation/controller/ObservationServiceController.java

@@ -17,7 +17,7 @@ public class ObservationServiceController {
   @Qualifier("observationKafkaTemplate")
   private final KafkaTemplate<String, String> kafkaTemplate;
 
-  @Value("${spring.kafka.input.topic-name-observation}")
+  @Value("${spring.kafka.topics.nbs.observation}")
   private String observationTopic = "nbs_Observation";
 
   @PostMapping(value = "/reporting/observation-svc/observation")

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/observation/service/ObservationService.java

@@ -63,13 +63,13 @@ public class ObservationService {
   private static final Logger logger = LoggerFactory.getLogger(ObservationService.class);
   private static final String BEFORE_PATH = "before";
 
-  @Value("${spring.kafka.input.topic-name-observation}")
+  @Value("${spring.kafka.topics.nbs.observation}")
   private String observationTopic;
 
-  @Value("${spring.kafka.input.topic-name-ar}")
+  @Value("${spring.kafka.topics.nbs.act-relationship}")
   private String actRelationshipTopic;
 
-  @Value("${spring.kafka.output.observationReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation}")
   private String observationTopicOutputReporting;
 
   @Value("${featureFlag.thread-pool-size:1}")
@@ -131,8 +131,8 @@ void initMetrics() {
       kafkaTemplate = "observationKafkaTemplate")
   @KafkaListener(
       topics = {
-        "${spring.kafka.input.topic-name-observation}",
-        "${spring.kafka.input.topic-name-ar}"
+        "${spring.kafka.topics.nbs.observation}",
+        "${spring.kafka.topics.nbs.act-relationship}"
       },
       containerFactory = "observationKafkaListenerContainerFactory")
   public CompletableFuture<Void> processMessage(ConsumerRecord<String, String> rec) {

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/observation/transformer/ProcessObservationDataUtil.java

@@ -35,25 +35,25 @@ public class ProcessObservationDataUtil {
 
   private final CustomJsonGeneratorImpl jsonGenerator = new CustomJsonGeneratorImpl();
 
-  @Value("${spring.kafka.output.observationCoded.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-coded}")
   public String codedTopicName;
 
-  @Value("${spring.kafka.output.observationDate.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-date}")
   public String dateTopicName;
 
-  @Value("${spring.kafka.output.observationEdx.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-edx}")
   public String edxTopicName;
 
-  @Value("${spring.kafka.output.observationMaterial.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-material}")
   public String materialTopicName;
 
-  @Value("${spring.kafka.output.observationNumeric.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-numeric}")
   public String numericTopicName;
 
-  @Value("${spring.kafka.output.observationReason.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-reason}")
   public String reasonTopicName;
 
-  @Value("${spring.kafka.output.observationTxt.topic-name}")
+  @Value("${spring.kafka.topics.nrt.observation-txt}")
   public String txtTopicName;
 
   ObservationKey observationKey = new ObservationKey();

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/organization/controller/OrganizationServiceController.java

@@ -17,10 +17,10 @@ public class OrganizationServiceController {
   @Qualifier("organizationKafkaTemplate")
   private final KafkaTemplate<String, String> kafkaTemplate;
 
-  @Value("${spring.kafka.input.topic-name-organization}")
+  @Value("${spring.kafka.topics.nbs.organization}")
   private String orgTopicName = "nbs_Organization";
 
-  @Value("${spring.kafka.input.topic-name-place}")
+  @Value("${spring.kafka.topics.nbs.place}")
   private String placeTopicName = "nbs_Place";
 
   @PostMapping(value = "/reporting/organization-svc/organization")

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/organization/service/OrganizationService.java

@@ -71,22 +71,22 @@ public class OrganizationService {
   @Qualifier("organizationKafkaTemplate")
   private final KafkaTemplate<String, String> kafkaTemplate;
 
-  @Value("${spring.kafka.input.topic-name-organization}")
+  @Value("${spring.kafka.topics.nbs.organization}")
   private String orgTopic;
 
-  @Value("${spring.kafka.input.topic-name-place}")
+  @Value("${spring.kafka.topics.nbs.place}")
   private String placeTopic;
 
-  @Value("${spring.kafka.output.organizationElastic.topic-name}")
+  @Value("${spring.kafka.topics.elastic.organization}")
   private String orgElasticSearchTopic;
 
-  @Value("${spring.kafka.output.organizationReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.organization}")
   private String orgReportingOutputTopic;
 
-  @Value("${spring.kafka.output.placeReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.place}")
   private String placeReportingOutputTopic;
 
-  @Value("${spring.kafka.output.placeTeleReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.place-tele}")
   private String teleOutputTopic;
 
   @Value("${featureFlag.elastic-search-enable}")
@@ -144,10 +144,7 @@ void initMetrics() {
         NoDataException.class
       })
   @KafkaListener(
-      topics = {
-        "${spring.kafka.input.topic-name-organization}",
-        "${spring.kafka.input.topic-name-place}"
-      },
+      topics = {"${spring.kafka.topics.nbs.organization}", "${spring.kafka.topics.nbs.place}"},
       containerFactory = "organizationKafkaListenerContainerFactory")
   public CompletableFuture<Void> processMessage(
       String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/person/controller/PersonServiceController.java

@@ -19,10 +19,10 @@ public class PersonServiceController {
 
   private static final String PRODUCED = "Produced : ";
 
-  @Value("${spring.kafka.input.topic-name-person}")
+  @Value("${spring.kafka.topics.nbs.person}")
   private String personTopicName = "nbs_Person";
 
-  @Value("${spring.kafka.input.topic-name-user}")
+  @Value("${spring.kafka.topics.nbs.auth-user}")
   private String userTopicName = "nbs_Auth_user";
 
   @PostMapping(value = "/reporting/person-svc/provider")

reporting-hydration-service/src/main/java/gov/cdc/etldatapipeline/person/service/PersonService.java

@@ -76,25 +76,25 @@ public class PersonService {
   @Qualifier("personKafkaTemplate")
   private final KafkaTemplate<String, String> kafkaTemplate;
 
-  @Value("${spring.kafka.input.topic-name-person}")
+  @Value("${spring.kafka.topics.nbs.person}")
   private String personTopic;
 
-  @Value("${spring.kafka.input.topic-name-user}")
+  @Value("${spring.kafka.topics.nbs.auth-user}")
   private String userTopic;
 
-  @Value("${spring.kafka.output.patientElastic.topic-name}")
+  @Value("${spring.kafka.topics.elastic.patient}")
   private String patientElasticSearchOutputTopic;
 
-  @Value("${spring.kafka.output.patientReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.patient}")
   private String patientReportingOutputTopic;
 
-  @Value("${spring.kafka.output.providerElastic.topic-name}")
+  @Value("${spring.kafka.topics.elastic.provider}")
   private String providerElasticSearchOutputTopic;
 
-  @Value("${spring.kafka.output.providerReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.provider}")
   private String providerReportingOutputTopic;
 
-  @Value("${spring.kafka.output.userReporting.topic-name}")
+  @Value("${spring.kafka.topics.nrt.auth-user}")
   private String userReportingOutputTopic;
 
   @Value("${featureFlag.elastic-search-enable}")
@@ -153,7 +153,7 @@ void initMetrics() {
       },
       kafkaTemplate = "personKafkaTemplate")
   @KafkaListener(
-      topics = {"${spring.kafka.input.topic-name-person}", "${spring.kafka.input.topic-name-user}"},
+      topics = {"${spring.kafka.topics.nbs.person}", "${spring.kafka.topics.nbs.auth-user}"},
       containerFactory = "personKafkaListenerContainerFactory")
   public CompletableFuture<Void> processMessage(
       String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {