Merge pull request #286 from jembi/CU-86bzpxg72_Update-Kafka-key---Co…

…mbine-given-and-family-name Cu 86bzpxg72 update kafka key combine given and family name
jembi · Jul 25, 2024 · c9765e0 · c9765e0
2 parents acd5d0a + 4e5eb87
commit c9765e0
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/JeMPI_Apps/JeMPI_AsyncReceiver/src/main/java/org/jembi/jempi/async_receiver/Main.java b/JeMPI_Apps/JeMPI_AsyncReceiver/src/main/java/org/jembi/jempi/async_receiver/Main.java
@@ -164,13 +164,15 @@ private void apacheReadCSV(
                   .findFirst()
                   .orElse("");
 
-               String partitionKey;
+               String partitionKey = "";
                if (!givenName.isEmpty()) {
-                  partitionKey = new Soundex().soundex(givenName);
-               } else if (!familyName.isEmpty()) {
-                  partitionKey = new Soundex().soundex(familyName);
-               } else {
-                  partitionKey = "Unknown";
+                  partitionKey += new Soundex().soundex(givenName);
+               }
+               if (!familyName.isEmpty()) {
+                  partitionKey += new Soundex().soundex(familyName);
+               }
+               if (givenName.isEmpty() && familyName.isEmpty()) {
+                  partitionKey += "Unknown";
                }
                LOGGER.info("Kafka topic/partition for patient: " + partitionKey);
 

diff --git a/JeMPI_Apps/JeMPI_Linker/src/main/java/org/jembi/jempi/linker/SPInteractions.java b/JeMPI_Apps/JeMPI_Linker/src/main/java/org/jembi/jempi/linker/SPInteractions.java
@@ -95,6 +95,7 @@ private void closeInteractionStream() {
 
    private Properties loadConfig() {
       final Properties props = new Properties();
+      props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
       props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, AppConfig.KAFKA_BOOTSTRAP_SERVERS);
       props.put(StreamsConfig.APPLICATION_ID_CONFIG, AppConfig.KAFKA_APPLICATION_ID_INTERACTIONS + topic);
       return props;