feat(config): support alternate hashing algorithm for doc id

docker/datahub-gms/env/docker-without-neo4j.env

@@ -23,6 +23,8 @@ PE_CONSUMER_ENABLED=true
 UI_INGESTION_ENABLED=true
 ENTITY_SERVICE_ENABLE_RETENTION=true
 
+ELASTIC_ID_HASH_ALGO=MD5
+
 # Uncomment to disable persistence of client-side analytics events
 # DATAHUB_ANALYTICS_ENABLED=false
 

docker/datahub-gms/env/docker.env

@@ -27,6 +27,8 @@ MCE_CONSUMER_ENABLED=true
 PE_CONSUMER_ENABLED=true
 UI_INGESTION_ENABLED=true
 
+ELASTIC_ID_HASH_ALGO=MD5
+
 # Uncomment to enable Metadata Service Authentication
 METADATA_SERVICE_AUTH_ENABLED=false
 

docker/quickstart/docker-compose-m1.quickstart.yml

@@ -17,6 +17,7 @@ services:
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
     - KAFKA_MESSAGE_MAX_BYTES=5242880
     - KAFKA_MAX_MESSAGE_BYTES=5242880
+    - ELASTIC_ID_HASH_ALGO=MD5
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose-without-neo4j-m1.quickstart.yml

@@ -17,6 +17,7 @@ services:
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
     - KAFKA_MESSAGE_MAX_BYTES=5242880
     - KAFKA_MAX_MESSAGE_BYTES=5242880
+    - ELASTIC_ID_HASH_ALGO=MD5
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose-without-neo4j.quickstart.yml

@@ -17,6 +17,7 @@ services:
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
     - KAFKA_MESSAGE_MAX_BYTES=5242880
     - KAFKA_MAX_MESSAGE_BYTES=5242880
+    - ELASTIC_ID_HASH_ALGO=MD5
     healthcheck:
       interval: 1s
       retries: 5

docker/quickstart/docker-compose.consumers-without-neo4j.quickstart.yml

@@ -19,6 +19,7 @@ services:
     - ES_BULK_REFRESH_POLICY=WAIT_UNTIL
     - GRAPH_SERVICE_IMPL=elasticsearch
     - ENTITY_REGISTRY_CONFIG_PATH=/datahub/datahub-mae-consumer/resources/entity-registry.yml
+    - ELASTIC_ID_HASH_ALGO=MD5
     hostname: datahub-mae-consumer
     image: ${DATAHUB_MAE_CONSUMER_IMAGE:-acryldata/datahub-mae-consumer}:${DATAHUB_VERSION:-head}
     ports:

docker/quickstart/docker-compose.consumers.quickstart.yml

@@ -26,6 +26,7 @@ services:
     - NEO4J_PASSWORD=datahub
     - GRAPH_SERVICE_IMPL=neo4j
     - ENTITY_REGISTRY_CONFIG_PATH=/datahub/datahub-mae-consumer/resources/entity-registry.yml
+    - ELASTIC_ID_HASH_ALGO=MD5
     hostname: datahub-mae-consumer
     image: ${DATAHUB_MAE_CONSUMER_IMAGE:-acryldata/datahub-mae-consumer}:${DATAHUB_VERSION:-head}
     ports:

docker/quickstart/docker-compose.quickstart.yml

@@ -17,6 +17,7 @@ services:
     - KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE=false
     - KAFKA_MESSAGE_MAX_BYTES=5242880
     - KAFKA_MAX_MESSAGE_BYTES=5242880
+    - ELASTIC_ID_HASH_ALGO=MD5
     healthcheck:
       interval: 1s
       retries: 5

entity-registry/src/main/java/com/linkedin/metadata/aspect/models/graph/Edge.java

@@ -72,8 +72,9 @@ public String toDocId() {
     }
 
     try {
+      String hashAlgo = System.getenv("ELASTIC_ID_HASH_ALGO");
       byte[] bytesOfRawDocID = rawDocId.toString().getBytes(StandardCharsets.UTF_8);
-      MessageDigest md = MessageDigest.getInstance("MD5");
+      MessageDigest md = MessageDigest.getInstance(hashAlgo);
       byte[] thedigest = md.digest(bytesOfRawDocID);
       return Base64.getEncoder().encodeToString(thedigest);
     } catch (NoSuchAlgorithmException e) {

metadata-io/build.gradle

@@ -130,6 +130,7 @@ test {
     // override, testng controlling parallelization
     // increasing >1 will merely run all tests extra times
     maxParallelForks = 1
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
   }
   useTestNG() {
     suites 'src/test/resources/testng.xml'

metadata-io/src/main/java/com/linkedin/metadata/systemmetadata/ElasticSearchSystemMetadataService.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.linkedin.common.urn.Urn;
 import com.linkedin.metadata.run.AspectRowSummary;
@@ -42,6 +43,7 @@
 import org.opensearch.search.aggregations.bucket.terms.ParsedStringTerms;
 import org.opensearch.search.aggregations.bucket.terms.Terms;
 import org.opensearch.search.aggregations.metrics.ParsedMax;
+import org.springframework.beans.factory.annotation.Value;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -71,6 +73,14 @@ public class ElasticSearchSystemMetadataService
               FIELD_REGISTRY_NAME,
               FIELD_REGISTRY_VERSION));
 
+  @Value("${elasticsearch.idHashAlgo}")
+  private String hashAlgo;
+
+  @VisibleForTesting
+  public void setIdHashAlgo(String algo) {
+    hashAlgo = algo;
+  }
+
   private String toDocument(SystemMetadata systemMetadata, String urn, String aspect) {
     final ObjectNode document = JsonNodeFactory.instance.objectNode();
 
@@ -86,10 +96,10 @@ private String toDocument(SystemMetadata systemMetadata, String urn, String aspe
 
   private String toDocId(@Nonnull final String urn, @Nonnull final String aspect) {
     String rawDocId = urn + DOC_DELIMETER + aspect;
-
+    String hashAlgo = System.getenv("ELASTIC_ID_HASH_ALGO");
     try {
       byte[] bytesOfRawDocID = rawDocId.getBytes(StandardCharsets.UTF_8);
-      MessageDigest md = MessageDigest.getInstance("MD5");
+      MessageDigest md = MessageDigest.getInstance(hashAlgo);
       byte[] thedigest = md.digest(bytesOfRawDocID);
       return Base64.getEncoder().encodeToString(thedigest);
     } catch (NoSuchAlgorithmException e) {

metadata-io/src/main/java/com/linkedin/metadata/timeseries/transformer/TimeseriesAspectTransformer.java

@@ -10,6 +10,7 @@
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
 import com.linkedin.common.urn.Urn;
 import com.linkedin.data.DataMap;
 import com.linkedin.data.schema.ArrayDataSchema;
@@ -32,6 +33,7 @@
 import javax.annotation.Nullable;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.codec.digest.DigestUtils;
+import org.springframework.beans.factory.annotation.Value;
 
 /** Class that provides a utility function that transforms the timeseries aspect into a document */
 @Slf4j
@@ -48,7 +50,15 @@ public class TimeseriesAspectTransformer {
         .setStreamReadConstraints(StreamReadConstraints.builder().maxStringLength(maxSize).build());
   }
 
-  private TimeseriesAspectTransformer() {}
+  public TimeseriesAspectTransformer() {}
+
+  @Value("${elasticsearch.idHashAlgo}")
+  private static String hashAlgo;
+
+  @VisibleForTesting
+  public void setIdHashAlgo(String algo) {
+    hashAlgo = algo;
+  }
 
   public static Map<String, JsonNode> transform(
       @Nonnull final Urn urn,
@@ -257,7 +267,9 @@ private static Pair<String, ObjectNode> getTimeseriesFieldCollectionDocument(
         finalDocument);
   }
 
-  private static String getDocId(@Nonnull JsonNode document, String collectionId) {
+  private static String getDocId(@Nonnull JsonNode document, String collectionId)
+      throws IllegalArgumentException {
+    String hashAlgo = System.getenv("ELASTIC_ID_HASH_ALGO");
     String docId = document.get(MappingsBuilder.TIMESTAMP_MILLIS_FIELD).toString();
     JsonNode eventGranularity = document.get(MappingsBuilder.EVENT_GRANULARITY);
     if (eventGranularity != null) {
@@ -276,6 +288,11 @@ private static String getDocId(@Nonnull JsonNode document, String collectionId)
       docId += partitionSpec.toString();
     }
 
-    return DigestUtils.md5Hex(docId);
+    if (hashAlgo.equalsIgnoreCase("SHA-256")) {
+      return DigestUtils.sha256Hex(docId);
+    } else if (hashAlgo.equalsIgnoreCase("MD5")) {
+      return DigestUtils.md5Hex(docId);
+    }
+    throw new IllegalArgumentException("Hash function not handled !");
   }
 }

metadata-io/src/test/java/com/linkedin/metadata/systemmetadata/SystemMetadataServiceTestBase.java

@@ -39,6 +39,8 @@ public abstract class SystemMetadataServiceTestBase extends AbstractTestNGSpring
   @BeforeClass
   public void setup() {
     _client = buildService();
+    _client.setIdHashAlgo("MD5");
+    _client.configure();
     _client.reindexAll(Collections.emptySet());
   }
 

metadata-io/src/test/java/com/linkedin/metadata/timeseries/search/TimeseriesAspectServiceTestBase.java

@@ -151,8 +151,10 @@ private ElasticSearchTimeseriesAspectService buildService() {
    */
 
   private void upsertDocument(TestEntityProfile dp, Urn urn) throws JsonProcessingException {
+    TimeseriesAspectTransformer timeseriesAspectTransformer = new TimeseriesAspectTransformer();
+    timeseriesAspectTransformer.setIdHashAlgo("MD5");
     Map<String, JsonNode> documents =
-        TimeseriesAspectTransformer.transform(urn, dp, aspectSpec, null);
+        timeseriesAspectTransformer.transform(urn, dp, aspectSpec, null);
     assertEquals(documents.size(), 3);
     documents.forEach(
         (key, value) ->

metadata-service/configuration/src/main/resources/application.yaml

@@ -174,6 +174,7 @@ elasticsearch:
   opensearchUseAwsIamAuth: ${OPENSEARCH_USE_AWS_IAM_AUTH:false}
   region: ${AWS_REGION:#{null}}
   implementation: ${ELASTICSEARCH_IMPLEMENTATION:elasticsearch} # elasticsearch or opensearch, for handling divergent cases
+  idHashAlgo: ${ELASTIC_ID_HASH_ALGO:MD5}
   sslContext: # Required if useSSL is true
     protocol: ${ELASTICSEARCH_SSL_PROTOCOL:#{null}}
     secureRandomImplementation: ${ELASTICSEARCH_SSL_SECURE_RANDOM_IMPL:#{null}}

smoke-test/build.gradle

@@ -89,6 +89,7 @@ task noCypressSuite0(type: Exec, dependsOn: [installDev, ':metadata-ingestion:in
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'no_cypress_suite0'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -101,6 +102,7 @@ task noCypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:in
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'no_cypress_suite1'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -113,6 +115,7 @@ task cypressSuite1(type: Exec, dependsOn: [installDev, ':metadata-ingestion:inst
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'cypress_suite1'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -125,6 +128,7 @@ task cypressRest(type: Exec, dependsOn: [installDev, ':metadata-ingestion:instal
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'TEST_STRATEGY', 'cypress_rest'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -139,6 +143,7 @@ task cypressDev(type: Exec, dependsOn: [installDev, ':metadata-ingestion:install
     environment 'RUN_QUICKSTART', 'false'
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',
@@ -154,6 +159,7 @@ task cypressData(type: Exec, dependsOn: [installDev, ':metadata-ingestion:instal
     environment 'DATAHUB_KAFKA_SCHEMA_REGISTRY_URL', 'http://localhost:8080/schema-registry/api/'
     environment 'KAFKA_BROKER_CONTAINER', 'datahub-kafka-broker-1'
     environment 'RUN_UI', 'false'
+    environment "ELASTIC_ID_HASH_ALGO", "MD5"
 
     workingDir = project.projectDir
     commandLine 'bash', '-c',

smoke-test/cypress-dev.sh

@@ -12,6 +12,7 @@ source venv/bin/activate
 
 export KAFKA_BROKER_CONTAINER="datahub-kafka-broker-1"
 export KAFKA_BOOTSTRAP_SERVER="broker:9092"
+export ELASTIC_ID_HASH_ALGO="MD5"
 python -c 'from tests.cypress.integration_test import ingest_data; ingest_data()'
 
 cd tests/cypress

smoke-test/run-quickstart.sh

@@ -15,6 +15,7 @@ DATAHUB_SEARCH_TAG="${DATAHUB_SEARCH_TAG:=2.9.0}"
 XPACK_SECURITY_ENABLED="${XPACK_SECURITY_ENABLED:=plugins.security.disabled=true}"
 ELASTICSEARCH_USE_SSL="${ELASTICSEARCH_USE_SSL:=false}"
 USE_AWS_ELASTICSEARCH="${USE_AWS_ELASTICSEARCH:=true}"
+ELASTIC_ID_HASH_ALGO="${ELASTIC_ID_HASH_ALGO:=MD5}"
 
 echo "DATAHUB_VERSION = $DATAHUB_VERSION"
 DATAHUB_TELEMETRY_ENABLED=false  \

smoke-test/smoke.sh

@@ -25,6 +25,7 @@ source venv/bin/activate
 source ./set-cypress-creds.sh
 
 export DATAHUB_GMS_URL=http://localhost:8080
+export ELASTIC_ID_HASH_ALGO="MD5"
 
 # no_cypress_suite0, no_cypress_suite1, cypress_suite1, cypress_rest
 if [[ -z "${TEST_STRATEGY}" ]]; then