fix(domains): cross-check OpenSearch results against primary store in hasChildDomains() (#17732)

javabrett · web-flow · commit 31ed631a731b · 2026-06-05T08:02:55.000+10:00
DeleteDomainResolver blocked parent deletion even after all children
were removed. hasChildDomains() queried OpenSearch (eventually
consistent) -- recently-deleted children could still appear there
while already gone from MySQL.

Cross-check each OpenSearch candidate via filterExistingUrns(), a
batched MySQL read. Discard stale hits; only block if MySQL confirms
at least one child still exists. Fall back to blocking when OpenSearch
reports more candidates than the fetched page, to avoid a false allow
on large hierarchies.

Smoke test reproduces the race by firing child and parent deletes with
no consistency sleep between them.
diff --git a/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/util/DomainUtils.java b/datahub-graphql-core/src/main/java/com/linkedin/datahub/graphql/resolvers/mutate/util/DomainUtils.java
@@ -201,12 +201,36 @@ public static boolean hasChildDomains(
       @Nonnull final EntityClient entityClient)
       throws RemoteInvocationException {
     Filter parentDomainFilter = buildParentDomainFilter(domainUrn);
-    // Search for entities matching parent domain
-    // Limit count to 1 for existence check
+    // Fetch candidate child domains from OpenSearch. OpenSearch is eventually
+    // consistent -- a recently-deleted child may still appear here.
+    final int pageSize = 200;
     final SearchResult searchResult =
         entityClient.filter(
-            context.getOperationContext(), DOMAIN_ENTITY_NAME, parentDomainFilter, null, 0, 1);
-    return (searchResult.getNumEntities() > 0);
+            context.getOperationContext(),
+            DOMAIN_ENTITY_NAME,
+            parentDomainFilter,
+            null,
+            0,
+            pageSize);
+    // Cross-check each candidate against the primary store (MySQL) which is
+    // strongly consistent. Discard stale OpenSearch entries whose deletions
+    // have not yet been indexed. Only block deletion if at least one child
+    // still exists in the primary store.
+    final Set<Urn> candidateUrns =
+        searchResult.getEntities().stream()
+            .map(SearchEntity::getEntity)
+            .collect(Collectors.toSet());
+    if (candidateUrns.isEmpty()) {
+      return false;
+    }
+    if (!entityClient.filterExistingUrns(context.getOperationContext(), candidateUrns).isEmpty()) {
+      return true;
+    }
+    // If OpenSearch reported more candidates than we fetched in this page, we
+    // cannot confirm the domain is childless without checking the remaining
+    // entries. Fall back to treating the domain as having children to prevent
+    // accidental deletion when the index is stale on a large hierarchy.
+    return searchResult.getNumEntities() > searchResult.getEntities().size();
   }
 
   private static Map<Urn, EntityResponse> getDomainsByNameAndParent(
diff --git a/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/domain/DeleteDomainResolverTest.java b/datahub-graphql-core/src/test/java/com/linkedin/datahub/graphql/resolvers/domain/DeleteDomainResolverTest.java
@@ -5,39 +5,44 @@
 import static org.testng.Assert.*;
 
 import com.linkedin.common.urn.Urn;
+import com.linkedin.common.urn.UrnUtils;
 import com.linkedin.datahub.graphql.QueryContext;
 import com.linkedin.entity.client.EntityClient;
+import com.linkedin.metadata.search.SearchEntity;
+import com.linkedin.metadata.search.SearchEntityArray;
 import com.linkedin.metadata.search.SearchResult;
 import graphql.schema.DataFetchingEnvironment;
+import java.util.Collections;
+import java.util.Set;
 import java.util.concurrent.CompletionException;
 import org.mockito.Mockito;
 import org.testng.annotations.Test;
 
 public class DeleteDomainResolverTest {
 
   private static final String TEST_URN = "urn:li:domain:test-id";
+  private static final String CHILD_URN = "urn:li:domain:child-id";
 
   @Test
   public void testGetSuccess() throws Exception {
     EntityClient mockClient = Mockito.mock(EntityClient.class);
     DeleteDomainResolver resolver = new DeleteDomainResolver(mockClient);
 
-    // Execute resolver
     QueryContext mockContext = getMockAllowContext();
     DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
     Mockito.when(mockEnv.getArgument(Mockito.eq("urn"))).thenReturn(TEST_URN);
     Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
 
-    // Domain has 0 child domains
+    // Domain has 0 child domains -- early exit before filterExistingUrns.
     Mockito.when(
             mockClient.filter(
                 any(),
                 Mockito.eq("domain"),
                 Mockito.any(),
                 Mockito.any(),
                 Mockito.eq(0),
-                Mockito.eq(1)))
-        .thenReturn(new SearchResult().setNumEntities(0));
+                Mockito.eq(200)))
+        .thenReturn(new SearchResult().setNumEntities(0).setEntities(new SearchEntityArray()));
 
     assertTrue(resolver.get(mockEnv).get());
 
@@ -50,35 +55,117 @@ public void testDeleteWithChildDomains() throws Exception {
     EntityClient mockClient = Mockito.mock(EntityClient.class);
     DeleteDomainResolver resolver = new DeleteDomainResolver(mockClient);
 
-    // Execute resolver
     QueryContext mockContext = getMockAllowContext();
     DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
     Mockito.when(mockEnv.getArgument(Mockito.eq("urn"))).thenReturn(TEST_URN);
     Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
 
-    // Domain has child domains
+    // OpenSearch returns one child candidate.
+    Urn childUrn = UrnUtils.getUrn(CHILD_URN);
+    SearchEntity childEntity = new SearchEntity().setEntity(childUrn);
     Mockito.when(
             mockClient.filter(
                 any(),
                 Mockito.eq("domain"),
                 Mockito.any(),
                 Mockito.any(),
                 Mockito.eq(0),
-                Mockito.eq(1)))
-        .thenReturn(new SearchResult().setNumEntities(1));
+                Mockito.eq(200)))
+        .thenReturn(
+            new SearchResult().setNumEntities(1).setEntities(new SearchEntityArray(childEntity)));
+
+    // Primary store (MySQL) confirms the child still exists.
+    Mockito.when(mockClient.filterExistingUrns(any(), Mockito.eq(Set.of(childUrn))))
+        .thenReturn(Set.of(childUrn));
 
     assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
 
     Mockito.verify(mockClient, Mockito.times(0)).deleteEntity(any(), Mockito.any());
   }
 
+  @Test
+  public void testDeleteBlockedWhenPagedCandidatesAllStaleButMoreExist() throws Exception {
+    // When OpenSearch reports more total candidates than fit in one page and all
+    // fetched candidates are stale in MySQL, deletion must still be blocked.
+    // Without the fallback check (numEntities > entities.size()), the code would
+    // incorrectly return false -- potentially allowing deletion of a domain that
+    // still has real children in the un-fetched remainder of the OpenSearch result.
+    EntityClient mockClient = Mockito.mock(EntityClient.class);
+    DeleteDomainResolver resolver = new DeleteDomainResolver(mockClient);
+
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    Mockito.when(mockEnv.getArgument(Mockito.eq("urn"))).thenReturn(TEST_URN);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    // OpenSearch reports 300 total but only returns one entry in this page,
+    // simulating the case where numEntities > the fetched page size.
+    Urn childUrn = UrnUtils.getUrn(CHILD_URN);
+    Mockito.when(
+            mockClient.filter(
+                any(),
+                Mockito.eq("domain"),
+                Mockito.any(),
+                Mockito.any(),
+                Mockito.eq(0),
+                Mockito.eq(200)))
+        .thenReturn(
+            new SearchResult()
+                .setNumEntities(300)
+                .setEntities(new SearchEntityArray(new SearchEntity().setEntity(childUrn))));
+
+    // The fetched candidate is stale in MySQL.
+    Mockito.when(mockClient.filterExistingUrns(any(), any())).thenReturn(Collections.emptySet());
+
+    // Must still block deletion: we cannot confirm childlessness from one page.
+    assertThrows(CompletionException.class, () -> resolver.get(mockEnv).join());
+    Mockito.verify(mockClient, Mockito.times(0)).deleteEntity(any(), Mockito.any());
+  }
+
+  @Test
+  public void testDeleteWithStaleChildDomains() throws Exception {
+    // Regression test for the OpenSearch eventual-consistency race condition:
+    // OpenSearch still shows a child that was just deleted from MySQL.
+    // hasChildDomains() must allow the parent delete to proceed once the
+    // primary store (MySQL) confirms no child actually exists.
+    EntityClient mockClient = Mockito.mock(EntityClient.class);
+    DeleteDomainResolver resolver = new DeleteDomainResolver(mockClient);
+
+    QueryContext mockContext = getMockAllowContext();
+    DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
+    Mockito.when(mockEnv.getArgument(Mockito.eq("urn"))).thenReturn(TEST_URN);
+    Mockito.when(mockEnv.getContext()).thenReturn(mockContext);
+
+    // OpenSearch returns a stale child candidate.
+    Urn childUrn = UrnUtils.getUrn(CHILD_URN);
+    SearchEntity childEntity = new SearchEntity().setEntity(childUrn);
+    Mockito.when(
+            mockClient.filter(
+                any(),
+                Mockito.eq("domain"),
+                Mockito.any(),
+                Mockito.any(),
+                Mockito.eq(0),
+                Mockito.eq(200)))
+        .thenReturn(
+            new SearchResult().setNumEntities(1).setEntities(new SearchEntityArray(childEntity)));
+
+    // Primary store (MySQL) confirms the child no longer exists -- stale OpenSearch hit.
+    Mockito.when(mockClient.filterExistingUrns(any(), Mockito.eq(Set.of(childUrn))))
+        .thenReturn(Collections.emptySet());
+
+    // Deletion should succeed because the only OpenSearch candidate is stale.
+    assertTrue(resolver.get(mockEnv).get());
+
+    Mockito.verify(mockClient, Mockito.times(1))
+        .deleteEntity(any(), Mockito.eq(Urn.createFromString(TEST_URN)));
+  }
+
   @Test
   public void testGetUnauthorized() throws Exception {
-    // Create resolver
     EntityClient mockClient = Mockito.mock(EntityClient.class);
     DeleteDomainResolver resolver = new DeleteDomainResolver(mockClient);
 
-    // Execute resolver
     DataFetchingEnvironment mockEnv = Mockito.mock(DataFetchingEnvironment.class);
     Mockito.when(mockEnv.getArgument(Mockito.eq("urn"))).thenReturn(TEST_URN);
     QueryContext mockContext = getMockDenyContext();
diff --git a/smoke-test/tests/domains/domains_test.py b/smoke-test/tests/domains/domains_test.py
@@ -1,4 +1,5 @@
 import logging
+import uuid
 from typing import Any, Dict
 
 import pytest
@@ -169,3 +170,97 @@ def test_set_unset_domain(auth_session, ingest_cleanup_data):
         res_data["data"]["dataset"]["domain"]["domain"]["properties"]["name"]
         == "Engineering"
     )
+
+
+_CREATE_DOMAIN_MUTATION = """
+mutation createDomain($input: CreateDomainInput!) {
+  createDomain(input: $input)
+}
+"""
+
+_DELETE_DOMAIN_MUTATION = """
+mutation deleteDomain($urn: String!) {
+  deleteDomain(urn: $urn)
+}
+"""
+
+
+def test_delete_parent_domain_immediately_after_child_deletion(auth_session):
+    """
+    A parent domain whose only child has just been deleted should be
+    immediately deletable -- no sleep or page-refresh should be required.
+
+    Demonstrates a race condition in DomainUtils.hasChildDomains(): it
+    queries OpenSearch (eventually consistent) rather than the primary
+    store (MySQL). When a child is deleted and the parent delete follows
+    immediately, OpenSearch may not yet have indexed the child's removal,
+    causing the parent delete to be rejected with "Cannot delete domain
+    which has child domains" even though the child is already gone.
+
+    The test bypasses TestSessionWrapper's post-mutation consistency sleep
+    so both deletes fire back-to-back with no gap for OpenSearch to catch up.
+
+    References
+    ----------
+    - DomainUtils.java: datahub-graphql-core/.../resolvers/mutate/util/DomainUtils.java
+    - DeleteDomainResolver.java: datahub-graphql-core/.../resolvers/domain/DeleteDomainResolver.java
+    """
+    run_id = uuid.uuid4().hex[:8]
+    parent_id = f"test-domain-race-parent-{run_id}"
+    child_id = f"test-domain-race-child-{run_id}"
+    parent_urn = f"urn:li:domain:{parent_id}"
+    child_urn = f"urn:li:domain:{child_id}"
+
+    try:
+        res = execute_graphql(
+            auth_session,
+            _CREATE_DOMAIN_MUTATION,
+            {"input": {"id": parent_id, "name": f"Race Test Parent {run_id}"}},
+        )
+        assert res["data"]["createDomain"] == parent_urn
+
+        res = execute_graphql(
+            auth_session,
+            _CREATE_DOMAIN_MUTATION,
+            {
+                "input": {
+                    "id": child_id,
+                    "name": f"Race Test Child {run_id}",
+                    "parentDomain": parent_urn,
+                }
+            },
+        )
+        assert res["data"]["createDomain"] == child_urn
+
+        # Delete child then immediately delete parent using the underlying
+        # session directly, bypassing TestSessionWrapper's post-mutation
+        # consistency sleep. This preserves the race window between the
+        # child deletion (MySQL write) and the parent deletion attempt
+        # (OpenSearch child-guard check).
+        endpoint = f"{auth_session.frontend_url()}/api/v2/graphql"
+        headers = {"Authorization": f"Bearer {auth_session.gms_token()}"}
+
+        def raw_graphql(query, variables):
+            resp = auth_session._upstream.post(
+                endpoint,
+                json={"query": query, "variables": variables},
+                headers=headers,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            assert "errors" not in data, f"GraphQL errors: {data.get('errors')}"
+            return data
+
+        res = raw_graphql(_DELETE_DOMAIN_MUTATION, {"urn": child_urn})
+        assert res["data"]["deleteDomain"] is True
+
+        res = raw_graphql(_DELETE_DOMAIN_MUTATION, {"urn": parent_urn})
+        assert res["data"]["deleteDomain"] is True
+
+    finally:
+        # Best-effort cleanup via REST, which bypasses the GraphQL child guard.
+        for urn in (child_urn, parent_urn):
+            try:
+                delete_entity(auth_session, urn)
+            except Exception:
+                pass
