Skip to content

Commit

Permalink
Merge branch 'dev' into CU-86bzu8m1g_Increase-Lease-in-Dgraph-while-r…
Browse files Browse the repository at this point in the history
…estoring-Backup-using-API
  • Loading branch information
MatthewErispe authored Oct 15, 2024
2 parents f7fb857 + ad7083b commit 66f0152
Show file tree
Hide file tree
Showing 13 changed files with 107 additions and 40 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import org.jembi.jempi.shared.serdes.JsonPojoSerializer;
import org.jembi.jempi.shared.utils.AppUtils;
import org.apache.commons.codec.language.Soundex;
import org.jetbrains.annotations.NotNull;

import java.io.IOException;
import java.nio.file.*;
Expand Down Expand Up @@ -153,28 +154,7 @@ private void apacheReadCSV(

for (CSVRecord csvRecord : csvParser) {
final var patientRecord = demographicData(csvRecord);
String givenName = patientRecord.fields.stream()
.filter(field -> "given_name".equals(field.ccTag()))
.map(DemographicData.DemographicField::value)
.findFirst()
.orElse("");
String familyName = patientRecord.fields.stream()
.filter(field -> "family_name".equals(field.ccTag()))
.map(DemographicData.DemographicField::value)
.findFirst()
.orElse("");

String partitionKey = "";
if (!givenName.isEmpty()) {
partitionKey += new Soundex().soundex(givenName);
}
if (!familyName.isEmpty()) {
partitionKey += new Soundex().soundex(familyName);
}
if (givenName.isEmpty() && familyName.isEmpty()) {
partitionKey += "Unknown";
}
LOGGER.info("Using Kafka topic/partition: " + partitionKey);
String partitionKey = generateKafkaPartitionKey(patientRecord);

final var interactionEnvelop = new InteractionEnvelop(InteractionEnvelop.ContentType.BATCH_INTERACTION,
tag,
Expand Down Expand Up @@ -203,6 +183,30 @@ private void apacheReadCSV(
}
}

private static @NotNull String generateKafkaPartitionKey(final DemographicData patientRecord) {
StringBuilder partitionKey = new StringBuilder();

for (String field : FIELDS_CONFIG.fieldsForKafkaKeyGen) {
String value = patientRecord.fields.stream()
.filter(fieldData -> field.equals(fieldData.ccTag()))
.map(DemographicData.DemographicField::value)
.findFirst()
.orElse("");

if (!value.isEmpty()) {
partitionKey.append(new Soundex().soundex(value));
}
}

if (partitionKey.length() == 0) {
partitionKey.append("Unknown");
}

String partitionKeyStr = partitionKey.toString();
LOGGER.info("Using Kafka topic/partition: {}", partitionKeyStr);
return partitionKeyStr;
}

private String updateStan(
final String stanDate,
final int recCount) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -61,9 +61,11 @@ case class Rules(
)

case class Config(
auxInteractionFields: Option[Array[auxField]],
auxGoldenRecordFields: Option[Array[auxField]],
additionalNodes: Option[Array[AdditionalNode]],
demographicFields: Array[DemographicField],
rules: Rules
auxInteractionFields: Option[Array[auxField]],
auxGoldenRecordFields: Option[Array[auxField]],
additionalNodes: Option[Array[AdditionalNode]],
demographicFields: Array[DemographicField],
rules: Rules,
fieldsForKafkaKeyGen: Option[List[String]],
nameFieldsForNotificationDisplay: Option[List[String]]
)
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@ public final class FieldsConfig {
public final List<AuxField> userAuxInteractionFields;
public final List<AuxField> userAuxGoldenRecordFields;
public final List<DemographicField> demographicFields;
public final List<String> fieldsForKafkaKeyGen;
public final List<String> nameFieldsForNotificationDisplay;
public final List<AdditionalNode> additionalNodes;

public FieldsConfig(final JsonConfig jsonConfig) {
Expand Down Expand Up @@ -88,6 +90,13 @@ public FieldsConfig(final JsonConfig jsonConfig) {
);
optionalInteractionAuxIdIdx = auxInteractionAuxIdIdx[0];

fieldsForKafkaKeyGen = jsonConfig.fieldsForKafkaKeyGen() != null
? List.copyOf(jsonConfig.fieldsForKafkaKeyGen())
: List.of();

nameFieldsForNotificationDisplay = jsonConfig.nameFieldsForNotificationDisplay() != null
? List.copyOf(jsonConfig.nameFieldsForNotificationDisplay())
: List.of();

demographicFields = jsonConfig.demographicFields()
.stream()
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@ public record JsonConfig(
@JsonProperty("auxGoldenRecordFields") List<AuxGoldenRecordField> auxGoldenRecordFields,
List<AdditionalNode> additionalNodes,
List<DemographicField> demographicFields,
@JsonProperty("fieldsForKafkaKeyGen") List<String> fieldsForKafkaKeyGen,
@JsonProperty("nameFieldsForNotificationDisplay") List<String> nameFieldsForNotificationDisplay,
Rules rules) {

private static final Logger LOGGER = LogManager.getLogger(JsonConfig.class);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package org.jembi.jempi.shared.config.input;
import com.fasterxml.jackson.annotation.JsonInclude;

import java.util.List;

@JsonInclude(JsonInclude.Include.NON_NULL)
public record NameFieldsForKafkaKeyGen(
List<String> nameFieldsForKafkaKeyGen
) {
}
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@ public record Configuration(
List<Field> auxGoldenRecordFields,
List<Node> additionalNodes,
List<DemographicField> demographicFields,
List<String> fieldsForKafkaKeyGen,
List<String> nameFieldsForNotificationDisplay,
Rules rules) {
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,17 +95,17 @@ private static boolean helperUpdateGoldenRecordField(

private static String patientName(final Interaction interaction) {
var patientRecord = interaction.demographicData();
String givenName = patientRecord.fields.stream()
.filter(field -> "given_name".equals(field.ccTag()))
.map(DemographicData.DemographicField::value)
.findFirst()
.orElse("");
String familyName = patientRecord.fields.stream()
.filter(field -> "family_name".equals(field.ccTag()))
.map(DemographicData.DemographicField::value)
.findFirst()
.orElse("");
return (givenName + " " + familyName).trim();
Map<String, String> fieldMap = patientRecord.fields.stream()
.collect(Collectors.toMap(DemographicData.DemographicField::ccTag,
DemographicData.DemographicField::value));

String patientDisplayName = FIELDS_CONFIG.nameFieldsForNotificationDisplay.stream()
.map(fieldName -> fieldMap.getOrDefault(fieldName,
""))
.filter(StringUtils::isNotBlank)
.collect(Collectors.joining(" "))
.trim();
return patientDisplayName;
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,14 @@
"indexGoldenRecord": "@index(exact)"
}
],
"fieldsForKafkaKeyGen": [
"given_name",
"family_name"
],
"nameFieldsForNotificationDisplay": [
"given_name",
"family_name"
],
"rules": {
"link": {
"deterministic": [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,14 @@
"matchMetaData": null
}
],
"fieldsForKafkaKeyGen": [
"given_name",
"family_name"
],
"nameFieldsForNotificationDisplay": [
"given_name",
"family_name"
],
"rules": {
"link": {
"deterministic": [
Expand Down
8 changes: 8 additions & 0 deletions devops/linux/docker/data-config/config-reference-link-d.json
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,14 @@
"indexGoldenRecord": "@index(exact)"
}
],
"fieldsForKafkaKeyGen": [
"given_name",
"family_name"
],
"nameFieldsForNotificationDisplay": [
"given_name",
"family_name"
],
"rules": {
"link": {
"deterministic": [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,14 @@
}
}
],
"fieldsForKafkaKeyGen": [
"given_name",
"family_name"
],
"nameFieldsForNotificationDisplay": [
"given_name",
"family_name"
],
"rules": {
"link": {
"deterministic": [
Expand Down
8 changes: 8 additions & 0 deletions devops/linux/docker/data-config/config-reference-link-dp.json
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,14 @@
}
}
],
"fieldsForKafkaKeyGen": [
"given_name",
"family_name"
],
"nameFieldsForNotificationDisplay": [
"given_name",
"family_name"
],
"rules": {
"link": {
"deterministic": [
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,5 +30,3 @@ source ../scripts/d-stack-up-keycloak-containers.sh
sleep 2
source ../scripts/d-stack-up-app-containers.sh
sleep 2
source ../../helper/keycloak/start-keycloak-test-server.sh
sleep 2

0 comments on commit 66f0152

Please sign in to comment.