test(misc): misc test updates (#6890)

.github/workflows/docker-unified.yml

@@ -312,8 +312,7 @@ jobs:
           fetch-depth: 0
       - name: Pre-build artifacts for docker image
         run: |
-          export USE_SYSTEM_NODE="true"
-          ./gradlew :datahub-frontend:dist -PuseSystemNode=${USE_SYSTEM_NODE} -x test -x yarnTest -x yarnLint --parallel
+          ./gradlew :datahub-frontend:dist -x test -x yarnTest -x yarnLint --parallel
           mv ./datahub-frontend/build/distributions/datahub-frontend-*.zip datahub-frontend.zip
       - name: Build and push
         uses: ./.github/actions/docker-custom-build-and-push

datahub-web-react/build.gradle

@@ -77,7 +77,7 @@ task yarnQuickBuild(type: YarnTask, dependsOn: [yarnInstall, yarnGenerate]) {
 }
 
 task cleanExtraDirs {
-  delete 'node_modules'
+  delete 'node_modules/.yarn-integrity'
   delete 'dist'
   delete 'tmp'
   delete 'just'

metadata-io/src/main/java/com/linkedin/metadata/entity/EntityService.java

@@ -83,6 +83,7 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import javax.persistence.EntityNotFoundException;
 
 import io.ebean.PagedList;
 import lombok.Value;
@@ -1588,7 +1589,12 @@ private Map<Urn, List<EnvelopedAspect>> getCorrespondingAspects(Set<EntityAspect
     final AspectSpec keySpec = spec.getKeyAspectSpec();
     String keyAspectName = getKeyAspectName(urn);
 
-    EntityAspect latestKey = _aspectDao.getLatestAspect(urn.toString(), keyAspectName);
+    EntityAspect latestKey = null;
+    try {
+      latestKey = _aspectDao.getLatestAspect(urn.toString(), keyAspectName);
+    } catch (EntityNotFoundException e) {
+      log.warn("Entity to delete does not exist. {}", urn.toString());
+    }
     if (latestKey == null || latestKey.getSystemMetadata() == null) {
       return new RollbackRunResult(removedAspects, rowsDeletedFromEntityDeletion);
     }

smoke-test/requirements.txt

@@ -4,4 +4,5 @@ psutil
 tenacity
 -e ../metadata-ingestion[datahub-rest,datahub-kafka,mysql]
 slack-sdk==3.18.1
-aiohttp
+aiohttp
+joblib

smoke-test/run-quickstart.sh

@@ -15,5 +15,4 @@ echo "test_user:test_pass" >> ~/.datahub/plugins/frontend/auth/user.props
 echo "DATAHUB_VERSION = $DATAHUB_VERSION"
 DATAHUB_TELEMETRY_ENABLED=false  \
 DOCKER_COMPOSE_BASE="file://$( dirname "$DIR" )" \
-
 datahub docker quickstart --standalone_consumers --dump-logs-on-failure

smoke-test/tests/utils.py

@@ -3,6 +3,7 @@ import os
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Tuple
 from time import sleep
+from joblib import Parallel, delayed
 
 import requests_wrapper as requests
 
@@ -112,7 +113,7 @@ def ingest_file_via_rest(filename: str) -> Pipeline:
     return pipeline
 
 
-def delete_urns_from_file(filename: str) -> None:
+def delete_urns_from_file(filename: str, shared_data: bool = False) -> None:
     if not cli_utils.get_boolean_env_variable("CLEANUP_DATA", True):
         print("Not cleaning data to save time")
         return
@@ -124,26 +125,33 @@ def delete_urns_from_file(filename: str) -> None:
         }
     )
 
+    def delete(entry):
+        is_mcp = "entityUrn" in entry
+        urn = None
+        # Kill Snapshot
+        if is_mcp:
+            urn = entry["entityUrn"]
+        else:
+            snapshot_union = entry["proposedSnapshot"]
+            snapshot = list(snapshot_union.values())[0]
+            urn = snapshot["urn"]
+        payload_obj = {"urn": urn}
+
+        cli_utils.post_delete_endpoint_with_session_and_url(
+            session,
+            get_gms_url() + "/entities?action=delete",
+            payload_obj,
+        )
+
     with open(filename) as f:
         d = json.load(f)
-        for entry in d:
-            is_mcp = "entityUrn" in entry
-            urn = None
-            # Kill Snapshot
-            if is_mcp:
-                urn = entry["entityUrn"]
-            else:
-                snapshot_union = entry["proposedSnapshot"]
-                snapshot = list(snapshot_union.values())[0]
-                urn = snapshot["urn"]
-            payload_obj = {"urn": urn}
+        Parallel(n_jobs=10)(delayed(delete)(entry) for entry in d)
 
-            cli_utils.post_delete_endpoint_with_session_and_url(
-                session,
-                get_gms_url() + "/entities?action=delete",
-                payload_obj,
-            )
-    sleep(requests.ELASTICSEARCH_REFRESH_INTERVAL_SECONDS)
+    # Deletes require 60 seconds when run between tests operating on common data, otherwise standard sync wait
+    if shared_data:
+        sleep(60)
+    else:
+        sleep(requests.ELASTICSEARCH_REFRESH_INTERVAL_SECONDS)
 
 
 # Fixed now value