diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/GlossaryTermInheritedReviewerApprovalIT.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/GlossaryTermInheritedReviewerApprovalIT.java new file mode 100644 index 000000000000..b59c1ca4b5e3 --- /dev/null +++ b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/GlossaryTermInheritedReviewerApprovalIT.java @@ -0,0 +1,475 @@ +/* + * Copyright 2024 Collate + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.it.tests; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.stream.Collectors; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.openmetadata.it.bootstrap.SharedEntities; +import org.openmetadata.it.util.SdkClients; +import org.openmetadata.it.util.TestNamespace; +import org.openmetadata.it.util.TestNamespaceExtension; +import org.openmetadata.schema.api.data.CreateGlossary; +import org.openmetadata.schema.api.data.CreateGlossaryTerm; +import org.openmetadata.schema.entity.data.Glossary; +import org.openmetadata.schema.entity.data.GlossaryTerm; +import org.openmetadata.schema.entity.tasks.Task; +import org.openmetadata.schema.entity.teams.User; +import org.openmetadata.schema.type.EntityReference; +import org.openmetadata.schema.type.EntityStatus; +import org.openmetadata.schema.type.TaskEntityStatus; +import org.openmetadata.sdk.models.ListResponse; +import org.openmetadata.sdk.network.HttpMethod; +import org.openmetadata.sdk.network.RequestOptions; +import org.openmetadata.service.Entity; + +/** + * Reproduces the Glossary Approval regression where a term whose reviewers are inherited (from + * the parent Glossary) does NOT get an approval task, while a term with reviewers attached + * directly does. + * + *

Reported scenario (see the attached workflow-history screenshot on the issue): + * + *

    + *
  1. A reviewer is attached to the term itself → the {@code GlossaryTermApprovalWorkflow} + * triggers, the term moves to {@code In Review} and an open approval task is created for the + * reviewer. (Works.) + *
  2. The reviewer is removed from the term so the term inherits the reviewer from its + * Glossary → no approval task is created and the term settles in {@code Draft}. (Bug.) + *
+ * + *

The workflow's {@code CheckGlossaryTermHasReviewers} node (JsonLogic {@code some reviewers ...}) + * is expected to see inherited reviewers because the node re-fetches the term with {@code + * getEntity(entityLink, "*", Include.ALL)} — which applies {@code setInheritedFields}. If inheritance + * is not honored on that read path, the term is routed to a terminal status (Draft/Approved) and no + * task is ever assigned to the inherited reviewer. + * + *

{@link #test_directReviewerOnTerm_createsOpenApprovalTask} is the control (direct reviewers) and + * must pass on every build. {@link #test_inheritedReviewerFromGlossary_createsOpenApprovalTask} and + * {@link #test_inheritedReviewerFromParentTerm_createsOpenApprovalTask} assert the post-fix invariant + * — inherited reviewers must produce the same {@code In Review} + open-task outcome as direct + * reviewers. On an affected build they fail fast: the approval workflow reaches a terminal status + * without ever creating a task. + */ +@ExtendWith(TestNamespaceExtension.class) +@Execution(ExecutionMode.CONCURRENT) +public class GlossaryTermInheritedReviewerApprovalIT { + + private static final String APPROVAL_WORKFLOW = "GlossaryTermApprovalWorkflow"; + private static final Set TERMINAL_WORKFLOW_STATUSES = + Set.of("FINISHED", "EXCEPTION", "FAILURE"); + // The approval outcome is driven by the async Flowable job executor (BPMN trigger → reviewer + // check + // → status transition → approval task), which contends with the rest of the concurrent suite, so + // the budgets mirror the generous ones in GlossaryTermMoveApprovalIT. + private static final Duration TASK_TIMEOUT = Duration.ofMinutes(5); + private static final Duration STATUS_TIMEOUT = Duration.ofMinutes(3); + private static final Duration POLL_INTERVAL = Duration.ofSeconds(2); + + /** Terms created after the reviewer is added, to land on a poisoned executor thread. */ + private static final int POISONED_THREAD_PROBE_COUNT = 6; + + protected SharedEntities shared() { + return SharedEntities.get(); + } + + /** USER1 carries the AllowAll test-admin role, so it is a valid approval-task assignee. */ + protected User reviewer() { + return shared().USER1; + } + + /** A distinct user used as a term owner (and as a removable direct reviewer). */ + protected User otherUser() { + return shared().USER2; + } + + /** + * Control: a reviewer attached directly to the term drives it to {@code In Review} with an open + * approval task assigned to that reviewer. This path is known to work and must always pass. + */ + @Test + void test_directReviewerOnTerm_createsOpenApprovalTask(TestNamespace ns) throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ false); + GlossaryTerm term = + createTerm(glossary, "direct_reviewer_term", /* attachReviewerDirectly */ true); + + Task task = awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + assertAssigneeIsReviewer(task); + waitForTermStatus(term.getId(), EntityStatus.IN_REVIEW); + } + + /** + * Bug: the glossary carries the reviewer and the term attaches none, so the term inherits the + * reviewer. The approval workflow must produce the SAME outcome as the direct case — {@code In + * Review} plus an open approval task assigned to the inherited reviewer. On an affected build the + * workflow instead finishes in {@code Draft} without creating a task, and this fails fast. + */ + @Test + void test_inheritedReviewerFromGlossary_createsOpenApprovalTask(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ true); + GlossaryTerm term = + createTerm(glossary, "inherited_reviewer_term", /* attachReviewerDirectly */ false); + + // Sanity: the term reports the reviewer via inheritance on a normal GET (Include.ALL), which is + // exactly the read the workflow node performs. If this holds but no task is created, the bug is + // isolated to the workflow's reviewer-check read path. + assertTermInheritsReviewer(term.getId()); + + Task task = awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + assertAssigneeIsReviewer(task); + waitForTermStatus(term.getId(), EntityStatus.IN_REVIEW); + } + + /** + * Bug variant: the reviewer is attached to a parent term and the child attaches none, so the child + * inherits the reviewer from the parent term (a different inheritance source than the Glossary, + * resolved by a distinct branch of {@code resolveEffectiveReviewers}). The parent is created BY the + * reviewer so it auto-approves (updatedBy == reviewer) and leaves no lingering task under it. The + * child must reach {@code In Review} with an open task assigned to the inherited reviewer. + */ + @Test + void test_inheritedReviewerFromParentTerm_createsOpenApprovalTask(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ false); + GlossaryTerm parent = createReviewerOwnedTerm(glossary, "parent_with_reviewer"); + GlossaryTerm child = createChildTerm(glossary, parent, "child_inherits_reviewer"); + + assertTermInheritsReviewer(child.getId()); + + Task task = awaitOpenApprovalTask(child.getId(), child.getFullyQualifiedName()); + assertAssigneeIsReviewer(task); + waitForTermStatus(child.getId(), EntityStatus.IN_REVIEW); + } + + /** + * Stale inheritance-parent cache. {@code EntityRepository.inheritanceParentCache} is a static + * ThreadLocal cleared only by {@code ImpersonationCleanupFilter}, a JAX-RS response filter — so on + * Flowable's async job-executor threads it is never cleared for the life of the process. + * + *

This drives the reported ordering: the glossary starts with NO reviewers, a first term is + * created so the workflow caches that reviewer-less glossary snapshot on a job thread, and only then + * is the reviewer added. Every term created afterwards must still be gated as "has reviewers"; a + * term that lands in a terminal status instead means a job thread served the stale parent. + * + *

Several terms are created because the executor has a pool of threads and only the poisoned ones + * exhibit the bug — one term could be routed to a clean thread and pass by luck. + */ + @Test + void test_reviewerAddedToGlossaryAfterFirstTerm_stillGatesLaterTerms(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ false); + + // Poison: this term's workflow caches the glossary parent while it has no reviewers. + GlossaryTerm seed = createTerm(glossary, "seed_before_reviewer", false); + waitForTermStatus(seed.getId(), EntityStatus.APPROVED); + + addReviewerToGlossary(glossary.getId()); + + for (int i = 0; i < POISONED_THREAD_PROBE_COUNT; i++) { + GlossaryTerm term = createTerm(glossary, "after_reviewer_" + i, false); + assertTermInheritsReviewer(term.getId()); + awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + } + } + + private void addReviewerToGlossary(UUID glossaryId) throws Exception { + String patch = + String.format( + "[{\"op\":\"add\",\"path\":\"/reviewers\",\"value\":" + + "[{\"id\":\"%s\",\"type\":\"user\"}]}]", + reviewer().getId()); + SdkClients.adminClient() + .glossaries() + .patch(glossaryId.toString(), new ObjectMapper().readTree(patch)); + } + + /** + * Grandparent inheritance: the GLOSSARY holds the reviewer, an intermediate term holds none, and + * the leaf holds none. Effective-reviewer resolution takes a single hop to the parent term, so this + * only succeeds if reading that parent applies its own inheritance and surfaces the glossary's + * reviewer — the assumption the generic resolution rests on. + */ + @Test + void test_inheritedReviewerFromGrandparentGlossary_createsOpenApprovalTask(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ true); + GlossaryTerm middle = createTerm(glossary, "middle_no_reviewer", false); + GlossaryTerm leaf = createChildTerm(glossary, middle, "leaf_inherits_from_glossary"); + + assertTermInheritsReviewer(leaf.getId()); + + Task task = awaitOpenApprovalTask(leaf.getId(), leaf.getFullyQualifiedName()); + assertAssigneeIsReviewer(task); + waitForTermStatus(leaf.getId(), EntityStatus.IN_REVIEW); + } + + /** + * Faithful shape of the reported "hello world" term: an OWNER and a description are set, and the + * reviewer is inherited from the glossary (none on the term). This isolates whether an owner on the + * term suppresses inherited-reviewer resolution at the approval gate (it must not). + */ + @Test + void test_inheritedReviewerWithOwnerSet_createsOpenApprovalTask(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ true); + CreateGlossaryTerm create = + new CreateGlossaryTerm() + .withName(ns.shortPrefix("owned_inherited")) + .withGlossary(glossary.getFullyQualifiedName()) + .withDescription("hello world") + .withOwners(List.of(otherUser().getEntityReference())); + GlossaryTerm term = SdkClients.adminClient().glossaryTerms().create(create); + + assertTermInheritsReviewer(term.getId()); + Task task = awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + assertAssigneeIsReviewer(task); + waitForTermStatus(term.getId(), EntityStatus.IN_REVIEW); + } + + /** + * The reported Run-2 sequence: the glossary carries the reviewer; the term starts with a DIFFERENT + * direct reviewer, then that direct reviewer is removed so the term must fall back to the + * glossary-inherited reviewer and STILL be gated as "has reviewers" (task kept, not dropped to a + * terminal status). + */ + @Test + void test_removeDirectReviewer_fallsBackToGlossaryInheritedReviewer(TestNamespace ns) + throws Exception { + Glossary glossary = createGlossary(ns, /* withReviewer */ true); // glossary reviewer = USER1 + CreateGlossaryTerm create = + new CreateGlossaryTerm() + .withName(ns.shortPrefix("removed_direct")) + .withGlossary(glossary.getFullyQualifiedName()) + .withDescription("hello world") + .withReviewers(List.of(otherUser().getEntityReference())); // direct reviewer = USER2 + GlossaryTerm term = SdkClients.adminClient().glossaryTerms().create(create); + awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + + removeDirectReviewers(term.getId()); // now the term inherits USER1 from the glossary + assertTermInheritsReviewer(term.getId()); + + // Decisive signal: the workflow must NOT drop the term to a terminal status without a task once + // the only remaining reviewer is inherited. (Task re-assignment timing is asserted elsewhere.) + awaitOpenApprovalTask(term.getId(), term.getFullyQualifiedName()); + } + + private void removeDirectReviewers(UUID termId) throws Exception { + JsonNode patch = + new ObjectMapper().readTree("[{\"op\":\"replace\",\"path\":\"/reviewers\",\"value\":[]}]"); + SdkClients.adminClient().glossaryTerms().patch(termId.toString(), patch); + } + + private Glossary createGlossary(TestNamespace ns, boolean withReviewer) { + CreateGlossary create = + new CreateGlossary() + .withName(ns.shortPrefix("inh")) + .withDescription("Glossary for inherited-reviewer approval test"); + if (withReviewer) { + create.withReviewers(List.of(reviewer().getEntityReference())); + } + return ns.trackRoot(Entity.GLOSSARY, SdkClients.adminClient().glossaries().create(create)); + } + + private GlossaryTerm createTerm(Glossary glossary, String name, boolean attachReviewerDirectly) { + CreateGlossaryTerm create = + new CreateGlossaryTerm() + .withName(name) + .withGlossary(glossary.getFullyQualifiedName()) + // A non-empty description is required to pass CheckGlossaryTermIsReadyToBeReviewed. + .withDescription("Term created by inherited-reviewer approval test"); + if (attachReviewerDirectly) { + create.withReviewers(List.of(reviewer().getEntityReference())); + } + return SdkClients.adminClient().glossaryTerms().create(create); + } + + private GlossaryTerm createChildTerm(Glossary glossary, GlossaryTerm parent, String name) { + CreateGlossaryTerm create = + new CreateGlossaryTerm() + .withName(name) + .withGlossary(glossary.getFullyQualifiedName()) + .withParent(parent.getFullyQualifiedName()) + .withDescription("Child term created by inherited-reviewer approval test"); + return SdkClients.adminClient().glossaryTerms().create(create); + } + + /** + * Creates a term with the reviewer attached directly, AS that reviewer, so the approval workflow + * auto-approves it (the updatedBy-is-reviewer branch) and it holds no open task of its own. + */ + private GlossaryTerm createReviewerOwnedTerm(Glossary glossary, String name) { + CreateGlossaryTerm create = + new CreateGlossaryTerm() + .withName(name) + .withGlossary(glossary.getFullyQualifiedName()) + .withDescription("Parent term created by inherited-reviewer approval test") + .withReviewers(List.of(reviewer().getEntityReference())); + return SdkClients.user1Client().glossaryTerms().create(create); + } + + private void assertTermInheritsReviewer(UUID termId) { + GlossaryTerm term = + SdkClients.adminClient().glossaryTerms().get(termId.toString(), "reviewers"); + Set reviewerIds = entityReferenceIds(term.getReviewers()); + assertTrue( + reviewerIds.contains(reviewer().getId()), + "Term " + + termId + + " should inherit reviewer " + + reviewer().getId() + + " but resolved reviewers were " + + reviewerIds); + } + + /** + * Waits until the term has an OPEN approval task, or fails fast the moment the approval workflow + * settles in a terminal state without one (the inherited-reviewers bug). Failing on a terminal, + * task-less workflow surfaces the regression in seconds instead of hanging for the full timeout. + */ + private Task awaitOpenApprovalTask(UUID termId, String termFqn) { + Map filters = + Map.of("limit", "100", "status", TaskEntityStatus.Open.value(), "aboutEntity", termFqn); + Awaitility.await("open approval task for " + termFqn) + .atMost(TASK_TIMEOUT) + .pollInterval(POLL_INTERVAL) + .until( + () -> + !safeListTasks(filters).isEmpty() + || approvalWorkflowSettledWithoutTask(termFqn, filters)); + List tasks = safeListTasks(filters); + if (tasks.isEmpty()) { + fail( + "Expected the Glossary Approval Workflow to create an OPEN approval task for " + + termFqn + + " (reviewer resolved via inheritance), but the workflow settled without one. " + + "term status=" + + safeCurrentStatus(termId) + + ", workflow instance statuses=" + + safeWorkflowStatuses(termFqn) + + ". Inherited reviewers were not honored — the term went straight to a terminal " + + "status instead of 'In Review' with an approval task."); + } + Task task = tasks.get(0); + assertNotNull(task.getId()); + return task; + } + + private boolean approvalWorkflowSettledWithoutTask(String termFqn, Map filters) { + boolean settledWithoutTask = false; + if (safeListTasks(filters).isEmpty()) { + List statuses = safeWorkflowStatuses(termFqn); + settledWithoutTask = + !statuses.isEmpty() && statuses.stream().allMatch(TERMINAL_WORKFLOW_STATUSES::contains); + } + return settledWithoutTask; + } + + private void assertAssigneeIsReviewer(Task task) { + Set assigneeIds = entityReferenceIds(task.getAssignees()); + assertTrue( + assigneeIds.contains(reviewer().getId()), + "Approval task " + + task.getId() + + " should be assigned to the (inherited) reviewer " + + reviewer().getId() + + " but assignees were " + + assigneeIds); + } + + private void waitForTermStatus(UUID termId, EntityStatus expected) { + Awaitility.await("glossary term " + termId + " should reach status " + expected) + .atMost(STATUS_TIMEOUT) + .pollInterval(POLL_INTERVAL) + .ignoreExceptions() + .untilAsserted(() -> assertEquals(expected, safeCurrentStatus(termId))); + } + + private List safeListTasks(Map filters) { + List tasks; + try { + ListResponse response = SdkClients.adminClient().tasks().listWithFilters(filters); + tasks = response.getData() == null ? List.of() : response.getData(); + } catch (RuntimeException e) { + tasks = List.of(); + } + return tasks; + } + + private EntityStatus safeCurrentStatus(UUID termId) { + EntityStatus status; + try { + status = SdkClients.adminClient().glossaryTerms().get(termId.toString()).getEntityStatus(); + } catch (RuntimeException e) { + status = null; + } + return status; + } + + private List safeWorkflowStatuses(String termFqn) { + List statuses; + try { + statuses = approvalWorkflowStatuses(termFqn); + } catch (Exception e) { + statuses = List.of(); + } + return statuses; + } + + private List approvalWorkflowStatuses(String termFqn) throws Exception { + long now = System.currentTimeMillis(); + RequestOptions options = + RequestOptions.builder() + .queryParam("entityLink", String.format("<#E::%s::%s>", Entity.GLOSSARY_TERM, termFqn)) + .queryParam("workflowDefinitionName", APPROVAL_WORKFLOW) + .queryParam("startTs", String.valueOf(now - Duration.ofHours(1).toMillis())) + .queryParam("endTs", String.valueOf(now + Duration.ofHours(1).toMillis())) + .queryParam("limit", "50") + .build(); + String response = + SdkClients.adminClient() + .getHttpClient() + .executeForString(HttpMethod.GET, "/v1/governance/workflowInstances", null, options); + JsonNode data = new ObjectMapper().readTree(response).path("data"); + List statuses = new ArrayList<>(); + if (data.isArray()) { + for (JsonNode instance : data) { + statuses.add(instance.path("status").asText()); + } + } + return statuses; + } + + private Set entityReferenceIds(List references) { + return references == null + ? Set.of() + : references.stream().map(EntityReference::getId).collect(Collectors.toSet()); + } +} diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java b/openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java index 6cb36a62dd11..a982b69c7dab 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java @@ -397,9 +397,9 @@ public void run(OpenMetadataApplicationConfig catalogConfig, Environment environ environment.jersey().register(ETagRequestFilter.class); environment.jersey().register(ETagResponseFilter.class); - // Clears per-request ThreadLocals (inheritanceParentCache, ReadBundleContext, - // RequestEntityCache, impersonation context) after every response so state - // cannot leak across requests that share a Jetty worker thread. + // Clears per-request ThreadLocals (ReadBundleContext, RequestEntityCache, impersonation + // context) after every response so state cannot leak across requests that share a Jetty + // worker thread. Non-HTTP pools clear the same set via PerRequestContextCleaner. environment.jersey().register(ImpersonationCleanupFilter.class); // Register User Activity Tracking diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java index 39978464c333..d1e3b082706a 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumer.java @@ -46,6 +46,7 @@ import org.openmetadata.service.notifications.recipients.RecipientResolver; import org.openmetadata.service.notifications.recipients.context.Recipient; import org.openmetadata.service.util.DIContainer; +import org.openmetadata.service.util.PerRequestContextCleaner; import org.quartz.DisallowConcurrentExecution; import org.quartz.Job; import org.quartz.JobDetail; @@ -490,6 +491,20 @@ record CursorPlan(long offset, long pendingGapSince, int recordCount, boolean sk @Override public void execute(JobExecutionContext jobExecutionContext) { + // Quartz worker threads are long lived, shared with every other scheduled job, and never pass + // through the JAX-RS response filter. Per-request ThreadLocal caches left behind here would be + // served to whatever runs next on this thread — indefinitely stale. Destinations on this thread + // read entities (governance workflows resolve inherited reviewers here), so bracket the whole + // tick: start clean, and leave clean however this exits. + PerRequestContextCleaner.clear(); + try { + executeTick(jobExecutionContext); + } finally { + PerRequestContextCleaner.clear(); + } + } + + private void executeTick(JobExecutionContext jobExecutionContext) { this.init(jobExecutionContext); if (this.eventSubscription == null) { LOG.error("Skipping job execution - EventSubscription could not be loaded"); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java index a217a5c8c81c..0946f52badab 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowHandler.java @@ -83,6 +83,7 @@ import org.openmetadata.service.jdbi3.WorkflowInstanceStateRepository; import org.openmetadata.service.jdbi3.locator.ConnectionType; import org.openmetadata.service.resources.services.ingestionpipelines.IngestionPipelineMapper; +import org.openmetadata.service.util.FreshReadScope; @Slf4j public class WorkflowHandler { @@ -292,8 +293,9 @@ public void initializeNewProcessEngine( // Add Expression Manager processEngineConfiguration.setExpressionManager(new DefaultExpressionManager(expressionMap)); - // Add Global Failure Listener - processEngineConfiguration.setEventListeners(List.of(new WorkflowFailureListener())); + // Add Global Failure Listener + per-job ThreadLocal cleanup for the async executor pool + processEngineConfiguration.setEventListeners( + List.of(new WorkflowFailureListener(), new WorkflowThreadCleanupListener())); boolean engineBuilt = false; try { @@ -707,9 +709,17 @@ public ProcessInstance triggerByKey( } } + /** + * Signals are delivered synchronously, so the whole workflow — filters, attribute gates, status + * transitions — runs inline on the caller's thread. Those gates decide whether an entity gets an + * approval task at all, so they read fresh rather than trusting an in-process cache that a write on + * another node may not have invalidated. + */ public void triggerWithSignal(String signal, Map variables) { RuntimeService runtimeService = processEngine.getRuntimeService(); - runtimeService.signalEventReceived(signal, variables); + try (FreshReadScope.Handle ignored = FreshReadScope.enter()) { + runtimeService.signalEventReceived(signal, variables); + } } private void unlockJobsOnStartup() { @@ -922,6 +932,15 @@ public boolean resolveLegacyThreadTask(UUID customTaskId, Map va private boolean resolveTaskInternal( UUID customTaskId, Map variables, boolean legacyThreadTask) { + // Completing a user task continues the workflow inline, re-running the attribute gates that + // decide the entity's next status — same freshness requirement as triggerWithSignal. + try (FreshReadScope.Handle ignored = FreshReadScope.enter()) { + return resolveTaskWithFreshReads(customTaskId, variables, legacyThreadTask); + } + } + + private boolean resolveTaskWithFreshReads( + UUID customTaskId, Map variables, boolean legacyThreadTask) { TaskService taskService = processEngine.getTaskService(); LOG.debug("[WorkflowTask] RESOLVE: customTaskId='{}' variables={}", customTaskId, variables); // Admission control: bound how many resolutions touch Flowable at once so an approval burst diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListener.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListener.java new file mode 100644 index 000000000000..0325a979653c --- /dev/null +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListener.java @@ -0,0 +1,62 @@ +/* + * Copyright 2024 Collate + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.governance.workflows; + +import org.flowable.common.engine.api.delegate.event.FlowableEngineEventType; +import org.flowable.common.engine.api.delegate.event.FlowableEvent; +import org.flowable.common.engine.api.delegate.event.FlowableEventListener; +import org.openmetadata.service.util.PerRequestContextCleaner; + +/** + * Clears per-request ThreadLocal caches once an async job finishes. + * + *

Flowable's async-executor threads are pooled and long lived, and never pass through the JAX-RS + * response filter that clears these ThreadLocals for HTTP requests. Without this, a parent entity + * read by one job is served to every later job on the same thread for the life of the process. + * + *

Only the job-completion events are handled: both are dispatched on the async-executor thread + * itself, after the job's delegates have finished, which mirrors the response filter's "clear once + * the unit of work completes" semantics. Note the synchronous workflow path (a change event + * signalling a process inline) does not produce these events — that thread is cleaned by {@code + * AbstractEventConsumer}. + */ +public class WorkflowThreadCleanupListener implements FlowableEventListener { + + @Override + public void onEvent(FlowableEvent event) { + // Registered engine-wide, so every event lands here: keep the check cheap and first. Compare + // against the enum constants rather than their names so that if Flowable renames an event type, + // this fails to compile instead of silently never clearing again. + if (event.getType() == FlowableEngineEventType.JOB_EXECUTION_SUCCESS + || event.getType() == FlowableEngineEventType.JOB_EXECUTION_FAILURE) { + PerRequestContextCleaner.clear(); + } + } + + /** Cleanup must never fail the job that triggered it. */ + @Override + public boolean isFailOnException() { + return false; + } + + @Override + public boolean isFireOnTransactionLifecycleEvent() { + return false; + } + + @Override + public String getOnTransaction() { + return null; + } +} diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DatabaseSchemaRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DatabaseSchemaRepository.java index 9cbbe0afe43f..a849eedadc79 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DatabaseSchemaRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/DatabaseSchemaRepository.java @@ -550,7 +550,7 @@ public void setInheritedFields(DatabaseSchema schema, Fields fields) { ? (needsRetention ? "owners,domains,retentionPeriod" : "owners,domains") : "retentionPeriod"; Database database = - getOrLoadInheritanceParent(schema.getDatabase(), inheritanceFields, Database.class); + loadInheritanceParentLeniently(schema.getDatabase(), inheritanceFields, Database.class); if (database == null) { return; } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java index 3ad3a46105ed..8e855a6db25e 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/EntityRepository.java @@ -271,6 +271,7 @@ import org.openmetadata.service.util.EntityUtil; import org.openmetadata.service.util.EntityUtil.Fields; import org.openmetadata.service.util.EntityUtil.RelationIncludes; +import org.openmetadata.service.util.FreshReadScope; import org.openmetadata.service.util.FullyQualifiedName; import org.openmetadata.service.util.LineageUtil; import org.openmetadata.service.util.ListWithOffsetFunction; @@ -343,8 +344,6 @@ public abstract class EntityRepository { public record EntityHistoryWithOffset(EntityHistory entityHistory, int nextOffset) {} - private record InheritanceCacheKey(String entityType, UUID entityId, String fieldsKey) {} - private static final int STRING_OBJECT_OVERHEAD_BYTES = 40; // Conservative upper-bound weight for a String: length() * 2 (UTF-16 worst-case) + 40 (header). @@ -607,9 +606,6 @@ public Integer load(String key) { */ private final ThreadLocal> parentCacheForPrepare = new ThreadLocal<>(); - private static final ThreadLocal> - inheritanceParentCache = ThreadLocal.withInitial(HashMap::new); - protected final ChangeSummarizer changeSummarizer; // Lock manager for preventing orphaned entities during cascade deletion @@ -958,19 +954,12 @@ protected void clearRelationshipsForUpdateMany(List entities) { */ @SuppressWarnings("unused") protected void setInheritedFields(T entity, Fields fields) { - if (!requiresParentForInheritance(entity, fields)) { - return; - } - String inheritableFields = getInheritableFields(); - EntityReference parentRef = getParentReference(entity); - EntityInterface parent = getCachedInheritanceParent(parentRef, inheritableFields); - if (parent == null) { - parent = resolveInheritanceParentLeniently(entity, inheritableFields); - cacheInheritanceParent(parentRef, inheritableFields, parent); - } - if (parent != null) { - // Keep single-entity inheritance path aligned with batch/recursive inheritance path. - applyInheritance(entity, fields, parent); + if (requiresParentForInheritance(entity, fields)) { + EntityInterface parent = resolveInheritanceParentLeniently(entity, getInheritableFields()); + if (parent != null) { + // Keep single-entity inheritance path aligned with batch/recursive inheritance path. + applyInheritance(entity, fields, parent); + } } } @@ -1176,51 +1165,23 @@ protected void setParentCache(Map cache) { /** Clear the parent cache after bulk prepare. */ public void clearParentCache() { parentCacheForPrepare.remove(); - inheritanceParentCache.remove(); - } - - public static void clearInheritanceParentCache() { - inheritanceParentCache.remove(); } - private EntityInterface getCachedInheritanceParent(EntityReference parentRef, String fields) { - if (parentRef == null || parentRef.getId() == null || nullOrEmpty(parentRef.getType())) { - return null; - } - Map cache = inheritanceParentCache.get(); - InheritanceCacheKey directKey = inheritanceCacheKey(parentRef, fields); - EntityInterface direct = cache.get(directKey); - if (direct != null) { - return direct; - } - - // Reuse a superset entry when the same parent was already loaded with broader fields - // (for example "owners,domains,retentionPeriod" can serve "owners,domains"). - Set requestedFields = parseFieldSet(directKey.fieldsKey()); - for (Entry entry : cache.entrySet()) { - InheritanceCacheKey cachedKey = entry.getKey(); - if (!cachedKey.entityType().equals(parentRef.getType()) - || !cachedKey.entityId().equals(parentRef.getId())) { - continue; - } - if (parseFieldSet(cachedKey.fieldsKey()).containsAll(requestedFields)) { - return entry.getValue(); - } - } - return null; - } - - protected final

P getOrLoadInheritanceParent( + /** + * Loads an inheritance parent, tolerating a parent that has been hard-deleted since the child was + * read. Returns null (skip inheritance) rather than propagating, matching {@link + * #resolveInheritanceParentLeniently}. + */ + protected final

P loadInheritanceParentLeniently( EntityReference parentRef, String fields, Class

parentClass) { - if (parentRef == null || parentRef.getId() == null || nullOrEmpty(parentRef.getType())) { - return null; - } - EntityInterface parent = getCachedInheritanceParent(parentRef, fields); - if (parent == null) { + P result = null; + if (parentRef != null && parentRef.getId() != null && !nullOrEmpty(parentRef.getType())) { try { - parent = + EntityInterface parent = Entity.getEntityForInheritance(parentRef.getType(), parentRef.getId(), fields, ALL); - cacheInheritanceParent(parentRef, fields, parent); + if (parentClass.isInstance(parent)) { + result = parentClass.cast(parent); + } } catch (EntityNotFoundException e) { LOG.debug( "Inheritance parent {} {} no longer exists; skipping inheritance", @@ -1228,52 +1189,7 @@ protected final

P getOrLoadInheritanceParent( parentRef.getId()); } } - if (!parentClass.isInstance(parent)) { - return null; - } - return parentClass.cast(parent); - } - - private void cacheInheritanceParent( - EntityReference parentRef, String fields, EntityInterface parent) { - if (parentRef == null || parentRef.getId() == null || nullOrEmpty(parentRef.getType())) { - return; - } - if (parent == null || parent.getId() == null) { - return; - } - inheritanceParentCache.get().put(inheritanceCacheKey(parentRef, fields), parent); - } - - private InheritanceCacheKey inheritanceCacheKey(EntityReference parentRef, String fields) { - return new InheritanceCacheKey( - parentRef.getType(), parentRef.getId(), normalizeFieldList(fields)); - } - - private String normalizeFieldList(String fields) { - if (fields == null || fields.isBlank()) { - return ""; - } - // Canonicalize field order so cache keys are stable across equivalent requests - // (e.g. "owners,domains" and "domains, owners" should share the same parent entry). - return fields - .lines() - .flatMap(line -> Stream.of(line.split(","))) - .map(String::trim) - .filter(field -> !field.isEmpty()) - .distinct() - .sorted() - .collect(Collectors.joining(",")); - } - - private Set parseFieldSet(String fields) { - if (fields == null || fields.isBlank()) { - return Collections.emptySet(); - } - return Stream.of(fields.split(",")) - .map(String::trim) - .filter(field -> !field.isEmpty()) - .collect(Collectors.toSet()); + return result; } /** @@ -1307,32 +1223,14 @@ protected void setInheritedFields(List entities, Fields fields) { var refsByType = parentRefMap.values().stream().collect(Collectors.groupingBy(EntityReference::getType)); + // One bulk load per parent type keeps list endpoints off an N+1, which is what the removed + // thread-local parent cache was really buying on this path. var loadedParents = new HashMap(); - var missingRefsByType = new HashMap>(); for (var entry : refsByType.entrySet()) { - var missingRefs = new ArrayList(); - for (var ref : entry.getValue()) { - var cachedParent = getCachedInheritanceParent(ref, inheritableFields); - if (cachedParent != null) { - loadedParents.put(ref.getId(), cachedParent); - } else { - missingRefs.add(ref); - } - } - if (!missingRefs.isEmpty()) { - missingRefsByType.put(entry.getKey(), missingRefs); - } - } - - for (var entry : missingRefsByType.entrySet()) { List parents = Entity.getEntitiesForInheritance(entry.getValue(), inheritableFields, ALL); for (var parent : parents) { loadedParents.put(parent.getId(), parent); - var parentRef = parentRefMap.get(parent.getId()); - if (parentRef != null) { - cacheInheritanceParent(parentRef, inheritableFields, parent); - } } } @@ -1542,18 +1440,18 @@ public final T get( Fields fields, RelationIncludes relationIncludes, boolean fromCache) { + final boolean useCache = cacheAllowed(fromCache); T requestCachedEntity = - RequestEntityCache.getById( - entityType, id, fields, relationIncludes, fromCache, entityClass); + RequestEntityCache.getById(entityType, id, fields, relationIncludes, useCache, entityClass); if (requestCachedEntity != null) { return withHref(uriInfo, requestCachedEntity); } - if (!fromCache) { + if (!useCache) { CACHE_WITH_ID.invalidate(new ImmutablePair<>(entityType, id)); } T entity = - withPhase("entityLookup", () -> find(id, relationIncludes.getDefaultInclude(), fromCache)); + withPhase("entityLookup", () -> find(id, relationIncludes.getDefaultInclude(), useCache)); ReadPlan readPlan = withPhase("readCreatePlan", () -> createReadPlan(entity, fields, relationIncludes)); ReadBundle readBundle = withPhase("buildReadBundle", () -> buildReadBundle(entity, readPlan)); @@ -1570,7 +1468,7 @@ public final T get( "requestCachePutById", () -> RequestEntityCache.putById( - entityType, id, fields, relationIncludes, fromCache, hydratedEntity, entityClass)); + entityType, id, fields, relationIncludes, useCache, hydratedEntity, entityClass)); if (hydratedEntity.getFullyQualifiedName() != null) { withPhase( "requestCachePutByName", @@ -1580,7 +1478,7 @@ public final T get( hydratedEntity.getFullyQualifiedName(), fields, relationIncludes, - fromCache, + useCache, hydratedEntity, entityClass)); } @@ -1628,7 +1526,17 @@ public final T find(UUID id, Include include) throws EntityNotFoundException { return find(id, include, true); } + /** + * Callers ask for cached reads by default; a fresh-read scope overrides them. Governance workflows + * open that scope because their reads drive gating decisions, where a stale answer routes an entity + * to the wrong terminal status silently. + */ + private static boolean cacheAllowed(boolean fromCache) { + return fromCache && !FreshReadScope.isActive(); + } + public final T find(UUID id, Include include, boolean fromCache) throws EntityNotFoundException { + fromCache = cacheAllowed(fromCache); var notFoundCache = CacheBundle.getNotFoundCache(); if (!fromCache) { // On the explicit-bypass path the L1 cache is being skipped entirely, so checking the @@ -1758,6 +1666,7 @@ public final T getByName( Fields fields, RelationIncludes relationIncludes, boolean fromCache) { + fromCache = cacheAllowed(fromCache); fqn = quoteFqn ? quoteName(fqn) : fqn; T requestCachedEntity = RequestEntityCache.getByName( @@ -2274,6 +2183,7 @@ public final T findByName(String fqn, Include include) { } public final T findByName(String fqn, Include include, boolean fromCache) { + fromCache = cacheAllowed(fromCache); fqn = quoteFqn ? quoteName(fqn) : fqn; var notFoundCache = CacheBundle.getNotFoundCache(); if (!fromCache) { @@ -4941,23 +4851,6 @@ protected final void cleanup(String deletedBy, T entityInterface) { // Delete the extension data storing custom properties removeExtension(entityInterface); - // Cancel any governance workflow instances tied to this entity before the row - // goes away, so downstream nodes do not throw EntityNotFoundException. The - // Flowable query and cancel calls must never abort the delete transaction — - // a stray engine failure here shouldn't wedge entity deletes on a data plane - // the workflow engine has no authority over. - if (WorkflowHandler.isInitialized()) { - try { - WorkflowHandler.getInstance() - .cancelInstancesForEntity(entityInterface.getId(), "Entity deleted"); - } catch (Exception cancelEx) { - LOG.warn( - "Failed to cancel workflow instances for entity {}: {}", - entityInterface.getId(), - cancelEx.getMessage()); - } - } - // Delete all the threads that are about this entity Entity.getFeedRepository().deleteByAbout(entityInterface.getId()); @@ -4971,6 +4864,10 @@ protected final void cleanup(String deletedBy, T entityInterface) { return null; }); + // Flowable uses a separate transaction. Cancelling only after this one commits prevents a + // rolled-back entity delete from leaving a live entity without its workflow, and keeps the + // workflow queries out of the entity transaction's lock-hold time. + cancelWorkflowInstances(List.of(entityInterface.getId())); // Re-invalidate after the transaction commits. Any read that slipped in between the // pre-delete invalidate and the commit could have re-populated the cache from the // still-visible DB row; clearing again here guarantees the next read goes back to the @@ -6941,6 +6838,7 @@ private void bulkDeleteReferencesAndRows(List entities) { if (jdbi == null) { bulkCleanupReferences(entities); bulkDeleteEntityRows(entities); + cancelWorkflowInstances(entityIds(entities)); return; } jdbi.inTransaction( @@ -6949,6 +6847,32 @@ private void bulkDeleteReferencesAndRows(List entities) { bulkDeleteEntityRows(entities); return null; }); + // Keep Flowable's separate transaction outside the entity delete transaction. See cleanup(). + cancelWorkflowInstances(entityIds(entities)); + } + + private List entityIds(List entities) { + List entityIds = new ArrayList<>(entities.size()); + for (T entity : entities) { + entityIds.add(entity.getId()); + } + return entityIds; + } + + private void cancelWorkflowInstances(Collection entityIds) { + // Workflow-engine failures must never wedge a hard-delete; the entity rows are the source of + // truth and the workflow cleanup is best effort. + if (!WorkflowHandler.isInitialized()) { + return; + } + try (var ignored = phase("bulkHardDeleteWorkflows")) { + WorkflowHandler.getInstance().cancelInstancesForEntities(entityIds, "Entity deleted"); + } catch (Exception cancelEx) { + LOG.warn( + "Failed to cancel workflow instances for {} entities: {}", + entityIds.size(), + cancelEx.getMessage()); + } } private void bulkCleanupReferences(List entities) { @@ -6987,20 +6911,6 @@ private void bulkCleanupReferences(List entities) { // must be cleared here — but in one IN-list delete per chunk instead of one per entity. daoCollection.usageDAO().deleteByIds(entityIds); } - try (var ignored = phase("bulkHardDeleteWorkflows")) { - // Workflow-engine failures must never wedge a bulk hard-delete; the workflow - // instances are best-effort cleanup, the entity rows are the source of truth. - if (WorkflowHandler.isInitialized()) { - try { - WorkflowHandler.getInstance().cancelInstancesForEntities(entityIds, "Entity deleted"); - } catch (Exception cancelEx) { - LOG.warn( - "Failed to cancel workflow instances for {} entities: {}", - entityIds.size(), - cancelEx.getMessage()); - } - } - } try (var ignored = phase("bulkHardDeleteFeedThreads")) { Entity.getFeedRepository().deleteByAbout(entityIds); } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java index a4852345223f..2d486633af13 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/TableRepository.java @@ -351,7 +351,7 @@ public void setInheritedFields(Table table, Fields fields) { ? (needsRetention ? "owners,domains,retentionPeriod" : "owners,domains") : "retentionPeriod"; DatabaseSchema schema = - getOrLoadInheritanceParent( + loadInheritanceParentLeniently( table.getDatabaseSchema(), inheritanceFields, DatabaseSchema.class); if (schema == null) { return; diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/security/ImpersonationCleanupFilter.java b/openmetadata-service/src/main/java/org/openmetadata/service/security/ImpersonationCleanupFilter.java index fca87cf2cb1a..adc634637d4b 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/security/ImpersonationCleanupFilter.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/security/ImpersonationCleanupFilter.java @@ -19,11 +19,7 @@ import jakarta.ws.rs.container.ContainerResponseContext; import jakarta.ws.rs.container.ContainerResponseFilter; import jakarta.ws.rs.ext.Provider; -import org.openmetadata.service.Entity; -import org.openmetadata.service.jdbi3.EntityRepository; -import org.openmetadata.service.jdbi3.ReadBundleContext; -import org.openmetadata.service.resources.filters.ETagRequestFilter; -import org.openmetadata.service.util.RequestEntityCache; +import org.openmetadata.service.util.PerRequestContextCleaner; /** * Response filter to clean up ThreadLocal impersonation context after each request. This prevents @@ -37,12 +33,6 @@ public class ImpersonationCleanupFilter implements ContainerResponseFilter { public void filter( ContainerRequestContext requestContext, ContainerResponseContext responseContext) { // Always clear ThreadLocal after request completes (success or failure) - ImpersonationContext.clear(); - ActivePersonaContext.clear(); - ETagRequestFilter.clearIfMatchHeader(); - RequestEntityCache.clear(); - ReadBundleContext.clear(); - EntityRepository.clearInheritanceParentCache(); - Entity.clearRepositoryThreadLocals(); + PerRequestContextCleaner.clear(); } } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/FreshReadScope.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/FreshReadScope.java new file mode 100644 index 000000000000..9f05fac620b5 --- /dev/null +++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/FreshReadScope.java @@ -0,0 +1,61 @@ +/* + * Copyright 2024 Collate + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.util; + +/** + * Thread-scoped marker that forces entity reads to bypass the in-JVM caches and go to the database. + * + *

Governance workflows use this. Their decisions are gates — "does this term have reviewers?" — + * and a stale answer silently routes an entity to the wrong terminal status with no error and no + * retry. The in-JVM L1 is invalidated on write locally and, when Redis is configured, across nodes + * via pub/sub; but with Redis disabled a multi-node deployment has no cross-node invalidation at + * all, so another node's L1 can answer from before the write. Workflow volume is low, so paying for + * a fresh read is the cheaper side of that trade. + * + *

Scope it with try-with-resources so the marker is always restored, including on exceptions: + * + * 

{@code
+ * try (FreshReadScope.Handle ignored = FreshReadScope.enter()) {
+ *   ...
+ * }
+ * }
+ * + *

{@link #enter()} restores the previous value rather than clearing, so nesting is safe. This is + * the same shape as {@link org.openmetadata.service.cache.EntityCacheBypass}, which opts out of the + * Redis layer; this one opts out of the in-process caches. + */ +public final class FreshReadScope { + + private static final ThreadLocal ACTIVE = ThreadLocal.withInitial(() -> Boolean.FALSE); + + private FreshReadScope() {} + + /** True when the calling thread is inside a fresh-read scope. */ + public static boolean isActive() { + return Boolean.TRUE.equals(ACTIVE.get()); + } + + /** Enters a fresh-read scope. Close the returned handle to restore the previous state. */ + public static Handle enter() { + boolean previous = ACTIVE.get(); + ACTIVE.set(Boolean.TRUE); + return () -> ACTIVE.set(previous); + } + + @FunctionalInterface + public interface Handle extends AutoCloseable { + @Override + void close(); + } +} diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/util/PerRequestContextCleaner.java b/openmetadata-service/src/main/java/org/openmetadata/service/util/PerRequestContextCleaner.java new file mode 100644 index 000000000000..282288d6acb8 --- /dev/null +++ b/openmetadata-service/src/main/java/org/openmetadata/service/util/PerRequestContextCleaner.java @@ -0,0 +1,46 @@ +/* + * Copyright 2024 Collate + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.util; + +import org.openmetadata.service.Entity; +import org.openmetadata.service.jdbi3.ReadBundleContext; +import org.openmetadata.service.resources.filters.ETagRequestFilter; +import org.openmetadata.service.security.ActivePersonaContext; +import org.openmetadata.service.security.ImpersonationContext; + +/** + * Clears the ThreadLocal state that is scoped to a single unit of work. + * + *

These ThreadLocals are caches and request context, not pending work, so dropping them can only + * cost a re-read. They were originally cleared only by the JAX-RS response filter, which meant any + * pool that never serves an HTTP request — the Quartz change-event consumer, Flowable's async job + * executor — accumulated them for the life of the process and served indefinitely stale reads. + * + *

Deliberately excluded: {@code LineageUtil.DEFERRED_LINEAGE_ES} and {@code + * SearchRepository.DEFERRED_SEARCH_WRITES}. Those hold pending writes, so clearing them + * would silently drop search and lineage updates rather than merely forcing a re-read. + */ +public final class PerRequestContextCleaner { + + private PerRequestContextCleaner() {} + + public static void clear() { + ImpersonationContext.clear(); + ActivePersonaContext.clear(); + ETagRequestFilter.clearIfMatchHeader(); + RequestEntityCache.clear(); + ReadBundleContext.clear(); + Entity.clearRepositoryThreadLocals(); + } +} diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumerTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumerTest.java index 1392f00856d4..4cc31c584557 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumerTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/apps/bundles/changeEvent/AbstractEventConsumerTest.java @@ -38,6 +38,7 @@ import org.openmetadata.service.notifications.recipients.RecipientResolver; import org.openmetadata.service.notifications.recipients.context.EmailRecipient; import org.openmetadata.service.notifications.recipients.context.Recipient; +import org.openmetadata.service.security.ImpersonationContext; import org.openmetadata.service.util.DIContainer; import org.quartz.JobDataMap; import org.quartz.JobDetail; @@ -141,6 +142,35 @@ void testConstructor() { assertNotNull(testEventConsumer.dependencies); } + /** + * Quartz worker threads are pooled and shared with every other scheduled job, and never pass + * through the JAX-RS response filter that clears these ThreadLocals for HTTP requests. Whatever + * runs next on the thread inherits anything left behind, so a tick must leave it clean however it + * exits — including an early return when the subscription cannot be loaded. + * + *

Scope of this test: it pins the end-to-end invariant and fails if the cleanup is removed + * altogether. It does not isolate the exit-side clear from the entry-side one, because the + * only paths that previously skipped cleanup run inside the private {@code init}, which offers no + * seam for a test to populate the ThreadLocals mid-tick. That the exit clear is unconditional is + * enforced structurally, by the try/finally in {@code execute}. + */ + @Test + void execute_leavesThreadCleanForTheNextJob() { + ImpersonationContext.setImpersonatedBy("someone"); + + try { + testEventConsumer.execute(jobExecutionContext); + } catch (RuntimeException expectedInThisHarness) { + // The subscription cannot be resolved here, so the tick either returns early or throws. + // Either way the cleanup guarantee below must hold. + } + + assertNull( + ImpersonationContext.getImpersonatedBy(), + "A tick must leave the thread clean, or the next job scheduled onto it reads stale " + + "per-request state"); + } + @Test void testSendAlertMethod() { UUID receiverId = UUID.randomUUID(); diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListenerTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListenerTest.java new file mode 100644 index 000000000000..3fc3cfbdacf7 --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowThreadCleanupListenerTest.java @@ -0,0 +1,84 @@ +/* + * Copyright 2024 Collate. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.governance.workflows; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.flowable.common.engine.api.delegate.event.FlowableEngineEventType; +import org.flowable.common.engine.api.delegate.event.FlowableEvent; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.openmetadata.service.security.ImpersonationContext; + +/** + * Flowable's async-executor threads are pooled and never pass through the JAX-RS response filter, so + * without a job-boundary hook the per-request ThreadLocals survive from one job to the next and serve + * stale reads for the life of the process. + */ +class WorkflowThreadCleanupListenerTest { + + private final WorkflowThreadCleanupListener listener = new WorkflowThreadCleanupListener(); + + @AfterEach + void tearDown() { + ImpersonationContext.clear(); + } + + @Test + void jobCompletion_clearsPerRequestContext() { + for (FlowableEngineEventType type : + new FlowableEngineEventType[] { + FlowableEngineEventType.JOB_EXECUTION_SUCCESS, + FlowableEngineEventType.JOB_EXECUTION_FAILURE + }) { + ImpersonationContext.setImpersonatedBy("someone"); + assertNotNull(ImpersonationContext.getImpersonatedBy(), "precondition for " + type); + + listener.onEvent(eventOfType(type)); + + assertNull( + ImpersonationContext.getImpersonatedBy(), + type + " must clear per-request context so the next job on this thread starts clean"); + } + } + + @Test + void unrelatedEvent_leavesContextAlone() { + ImpersonationContext.setImpersonatedBy("someone"); + + listener.onEvent(eventOfType(FlowableEngineEventType.ENTITY_CREATED)); + + assertNotNull( + ImpersonationContext.getImpersonatedBy(), + "Only job boundaries are cleanup points; the listener sees every engine event"); + } + + /** Cleanup is best-effort: it must never be able to fail the job that triggered it. */ + @Test + void listenerNeverFailsTheJob() { + assertFalse(listener.isFailOnException()); + assertFalse(listener.isFireOnTransactionLifecycleEvent()); + assertNull(listener.getOnTransaction()); + } + + private FlowableEvent eventOfType(FlowableEngineEventType type) { + FlowableEvent event = mock(FlowableEvent.class); + when(event.getType()).thenReturn(type); + return event; + } +} diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImplTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImplTest.java new file mode 100644 index 000000000000..6f17c08a40ed --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImplTest.java @@ -0,0 +1,173 @@ +/* + * Copyright 2024 Collate. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.governance.workflows.elements.nodes.automatedTask.impl; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.flowable.common.engine.api.delegate.Expression; +import org.flowable.engine.delegate.DelegateExecution; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; +import org.openmetadata.schema.entity.data.GlossaryTerm; +import org.openmetadata.schema.type.EntityReference; +import org.openmetadata.schema.type.Include; +import org.openmetadata.service.Entity; +import org.openmetadata.service.resources.feeds.MessageParser; + +/** + * Covers the reviewers gate of the Glossary Approval Workflow. + * + *

{@code CheckGlossaryTermHasReviewers} decides whether an approval task is created at all: it + * evaluates the shipped JsonLogic rule below against the term and routes a {@code false} result to a + * terminal status, leaving the term in Draft with no task. These tests pin how that rule reads the + * entity it is handed. + * + *

Whether a term that inherits its reviewers arrives here with them populated is a property + * of the read path, not of this delegate — it is covered by the inheritance tests and by {@code + * GlossaryTermInheritedReviewerApprovalIT}. + */ +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +class CheckEntityAttributesImplTest { + + /** The rule shipped in GlossaryApprovalWorkflow.json for CheckGlossaryTermHasReviewers. */ + private static final String HAS_REVIEWERS_RULE = + "{\"and\":[{\"some\":[{\"var\":\"reviewers\"},{\"!=\":[{\"var\":\"fullyQualifiedName\"},null]}]}]}"; + + private static final String NODE_ID = "CheckGlossaryTermHasReviewers"; + private static final String RESULT_KEY = NODE_ID + "_result"; + + @Mock private DelegateExecution execution; + @Mock private Expression rulesExpr; + @Mock private Expression inputNamespaceMapExpr; + + private CheckEntityAttributesImpl delegate; + private MockedStatic mockedEntity; + private Map capturedVars; + + @BeforeEach + void setUp() throws Exception { + delegate = new CheckEntityAttributesImpl(); + injectField(delegate, "rulesExpr", rulesExpr); + injectField(delegate, "inputNamespaceMapExpr", inputNamespaceMapExpr); + + when(inputNamespaceMapExpr.getValue(execution)).thenReturn("{\"relatedEntity\":\"global\"}"); + when(rulesExpr.getValue(execution)).thenReturn(HAS_REVIEWERS_RULE); + when(execution.getProcessDefinitionId()).thenReturn("GlossaryTermApprovalWorkflow:1:1"); + when(execution.getCurrentActivityId()).thenReturn(NODE_ID); + when(execution.getVariable("global_relatedEntity")) + .thenReturn("<#E::glossaryTerm::Property.hello world>"); + + mockedEntity = mockStatic(Entity.class); + + capturedVars = new HashMap<>(); + doAnswer( + invocation -> { + capturedVars.put(invocation.getArgument(0), invocation.getArgument(1)); + return null; + }) + .when(execution) + .setVariable(anyString(), any()); + } + + @AfterEach + void tearDown() { + mockedEntity.close(); + } + + /** Reviewers present on the entity — whether set directly or applied by inheritance. */ + @Test + void reviewersRule_reviewersPresentOnEntity_evaluatesTrue() { + givenRelatedEntity(termWithReviewers(List.of(reviewer("manoj")))); + + delegate.execute(execution); + + assertTrue(result(), "A term carrying a reviewer must satisfy the gate"); + } + + /** No reviewers anywhere: the gate must be false so the term auto-approves. */ + @Test + void reviewersRule_noReviewers_evaluatesFalse() { + givenRelatedEntity(termWithReviewers(null)); + + delegate.execute(execution); + + assertFalse(result(), "With no reviewers the gate must be false"); + } + + /** + * The shipped rule tests {@code fullyQualifiedName != null} per element, so a reference missing its + * FQN does not satisfy it. Pinned because inheritance copies references between entities. + */ + @Test + void reviewersRule_reviewerWithoutFqn_evaluatesFalse() { + givenRelatedEntity(termWithReviewers(List.of(new EntityReference().withType(Entity.USER)))); + + delegate.execute(execution); + + assertFalse(result(), "A reviewer reference with no FQN must not satisfy the rule"); + } + + private void givenRelatedEntity(GlossaryTerm term) { + mockedEntity + .when( + () -> + Entity.getEntity( + any(MessageParser.EntityLink.class), anyString(), any(Include.class))) + .thenReturn(term); + } + + private GlossaryTerm termWithReviewers(List reviewers) { + return new GlossaryTerm() + .withName("hello world") + .withFullyQualifiedName("Property.hello world") + .withDescription("hello world") + .withReviewers(reviewers); + } + + private EntityReference reviewer(String name) { + return new EntityReference() + .withType(Entity.USER) + .withName(name) + .withFullyQualifiedName(name) + .withInherited(true); + } + + private boolean result() { + return Boolean.TRUE.equals(capturedVars.get(RESULT_KEY)); + } + + private static void injectField(Object target, String fieldName, Object value) throws Exception { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } +} diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/EntityRepositoryInheritanceParentTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/EntityRepositoryInheritanceParentTest.java new file mode 100644 index 000000000000..3c1d83eca480 --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/jdbi3/EntityRepositoryInheritanceParentTest.java @@ -0,0 +1,174 @@ +/* + * Copyright 2024 Collate. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.openmetadata.service.jdbi3; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.openmetadata.schema.EntityInterface; +import org.openmetadata.schema.entity.data.Pipeline; +import org.openmetadata.schema.type.EntityReference; +import org.openmetadata.service.Entity; +import org.openmetadata.service.util.EntityUtil.Fields; +import org.openmetadata.service.util.EntityUtil.RelationIncludes; + +/** + * Inheritance must reflect the parent's current state, not a snapshot taken earlier on the same + * thread. + * + *

Parent lookups used to be memoized in a {@code static ThreadLocal} on {@code EntityRepository} + * that had no TTL, no size bound, and no eviction, and was cleared only by the JAX-RS response + * filter. Any pool that never serves an HTTP request — the Quartz change-event consumer that runs + * governance workflows, Flowable's async job executor — therefore kept the first snapshot forever. + * In production that meant a glossary read before its reviewer was added kept reporting no + * reviewers, so the approval workflow created no task and left terms in Draft until the pod + * restarted. + * + *

{@link #inheritance_reflectsParentMutation_onRepeatedReadsFromOneThread()} fails against that + * memoization: the second read returns the stale parent. + */ +class EntityRepositoryInheritanceParentTest { + + private CollectionDAO daoCollection; + private ParentTrackingPipelineRepo repository; + private Pipeline parent; + + /** Serves a single mutable parent and counts how many times inheritance asked for it. */ + private static class ParentTrackingPipelineRepo extends EntityRepository { + private final Pipeline parent; + private int parentLoads; + + ParentTrackingPipelineRepo(CollectionDAO.PipelineDAO dao, Pipeline parent) { + super("pipelines", Entity.PIPELINE, Pipeline.class, dao, "domains", "domains"); + this.parent = parent; + } + + @Override + public EntityReference getParentReference(Pipeline entity) { + return parent.getEntityReference(); + } + + /** + * Returns a fresh snapshot each time, as a real read does. Handing back the same mutable + * instance would let a memoized reference appear up to date and hide staleness. + */ + @Override + public EntityInterface getParentEntity(Pipeline entity, String fields) { + parentLoads++; + return new Pipeline() + .withId(parent.getId()) + .withName(parent.getName()) + .withFullyQualifiedName(parent.getFullyQualifiedName()) + .withDomains(parent.getDomains() == null ? null : new ArrayList<>(parent.getDomains())); + } + + @Override + protected void setFields(Pipeline entity, Fields fields, RelationIncludes r) {} + + @Override + protected void clearFields(Pipeline entity, Fields fields) {} + + @Override + protected void prepare(Pipeline entity, boolean update) {} + + @Override + protected void storeEntity(Pipeline entity, boolean update) {} + + @Override + protected void storeRelationships(Pipeline entity) {} + } + + @BeforeEach + void setUp() { + daoCollection = mock(CollectionDAO.class); + when(daoCollection.relationshipDAO()) + .thenReturn(mock(CollectionDAO.EntityRelationshipDAO.class)); + Entity.setCollectionDAO(daoCollection); + + parent = + new Pipeline() + .withId(UUID.randomUUID()) + .withName("parent") + .withFullyQualifiedName("parent") + .withDomains(new ArrayList<>()); + repository = new ParentTrackingPipelineRepo(mock(CollectionDAO.PipelineDAO.class), parent); + } + + @AfterEach + void tearDown() { + Entity.cleanup(); + } + + @Test + void inheritance_reflectsParentMutation_onRepeatedReadsFromOneThread() { + Fields domains = new Fields(Set.of(Entity.FIELD_DOMAINS)); + + // First read on this thread: the parent has no domains yet, so nothing is inherited. + Pipeline before = childPipeline(); + repository.setInheritedFields(before, domains); + assertTrue( + before.getDomains() == null || before.getDomains().isEmpty(), + "Nothing to inherit before the parent has a domain"); + + // The parent gains a domain — as a glossary gains a reviewer. + EntityReference domain = + new EntityReference() + .withId(UUID.randomUUID()) + .withType(Entity.DOMAIN) + .withName("finance") + .withFullyQualifiedName("finance"); + parent.setDomains(new ArrayList<>(List.of(domain))); + + // Second read on the SAME thread must see it. A memoized parent would still report none. + Pipeline after = childPipeline(); + repository.setInheritedFields(after, domains); + + assertEquals( + 1, + after.getDomains() == null ? 0 : after.getDomains().size(), + "Second read on the same thread must reflect the parent's current domains, " + + "not a snapshot cached during the first read"); + assertEquals("finance", after.getDomains().get(0).getFullyQualifiedName()); + } + + @Test + void inheritance_loadsParentOnEveryRead_soNoStaleSnapshotCanSurvive() { + Fields domains = new Fields(Set.of(Entity.FIELD_DOMAINS)); + + repository.setInheritedFields(childPipeline(), domains); + repository.setInheritedFields(childPipeline(), domains); + + assertEquals( + 2, + repository.parentLoads, + "Each read resolves the parent; memoizing it across reads is what made inherited " + + "fields go stale on long-lived non-HTTP threads"); + } + + private Pipeline childPipeline() { + return new Pipeline() + .withId(UUID.randomUUID()) + .withName("child") + .withFullyQualifiedName("child"); + } +} diff --git a/openmetadata-ui/src/main/resources/ui/playwright/e2e/Features/Glossary/GlossaryInheritedReviewerApproval.spec.ts b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Features/Glossary/GlossaryInheritedReviewerApproval.spec.ts new file mode 100644 index 000000000000..68c3c7fd7089 --- /dev/null +++ b/openmetadata-ui/src/main/resources/ui/playwright/e2e/Features/Glossary/GlossaryInheritedReviewerApproval.spec.ts @@ -0,0 +1,313 @@ +/* + * Copyright 2024 Collate. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import { APIRequestContext, expect, test } from '@playwright/test'; +import { Glossary } from '../../../support/glossary/Glossary'; +import { GlossaryTerm } from '../../../support/glossary/GlossaryTerm'; +import { UserClass } from '../../../support/user/UserClass'; +import { performAdminLogin } from '../../../utils/admin'; +import { uuid } from '../../../utils/common'; + +/** + * Reproduction for the Glossary Approval bug where a term whose reviewers are INHERITED from its + * glossary is treated as having none: no approval task is created and the term settles in Draft. + * + * The ordering matters. The test creates a glossary WITHOUT reviewers first, lets a term complete + * its workflow, and only then adds the reviewer. Every term created afterwards must inherit that + * reviewer and be gated as "has reviewers". + * + * Terms are created in a batch because the executor uses a thread pool and only threads that cached + * the stale parent misbehave — a single term could be routed to a clean thread and pass by luck. + */ + +const REVIEWER_ADDED_PROBE_COUNT = 6; +const STATUS_TIMEOUT = 120_000; + +test.use({ storageState: 'playwright/.auth/admin.json' }); + +type TermStatus = 'Draft' | 'In Review' | 'Approved' | 'Rejected' | string; + +const createTermWithoutReviewers = async ( + apiContext: APIRequestContext, + glossaryFqn: string, + name: string +) => { + const response = await apiContext.post('/api/v1/glossaryTerms', { + data: { + name, + displayName: name, + // A non-empty description is required to pass the "ready to be reviewed" gate. + description: 'Term used by the inherited-reviewer approval reproduction', + glossary: glossaryFqn, + }, + }); + + expect(response.status()).toBe(201); + + return response.json(); +}; + +/** + * Polls the term until the approval workflow has committed a status. + * + *

`Draft` is deliberately NOT a settled status. A term under a reviewed parent is written as + * `Draft` at creation, before the workflow runs, so accepting it here would return on the very first + * sample and every assertion would read the pre-workflow value. Waiting for a status the workflow + * itself sets means a term that never leaves `Draft` — the bug under test — times out with the + * message below rather than being misreported as a wrong-status failure. + */ +const waitForSettledStatus = async ( + apiContext: APIRequestContext, + termId: string +): Promise => { + let status: TermStatus = ''; + + await expect + .poll( + async () => { + const response = await apiContext.get( + `/api/v1/glossaryTerms/${termId}?fields=reviewers` + ); + const term = await response.json(); + status = term.entityStatus; + + return status; + }, + { + message: + `Glossary term ${termId} never left Draft — the approval workflow did not commit a ` + + `status. For a term inheriting a reviewer this is the inherited-reviewer bug.`, + timeout: STATUS_TIMEOUT, + } + ) + .toMatch(/In Review|Approved|Rejected/); + + return status; +}; + +const getOpenApprovalTaskCount = async ( + apiContext: APIRequestContext, + termFqn: string +) => { + const response = await apiContext.get( + `/api/v1/tasks?limit=100&status=Open&aboutEntity=${encodeURIComponent( + termFqn + )}` + ); + const body = await response.json(); + + return (body.data ?? []).length; +}; + +const addReviewerToGlossary = async ( + apiContext: APIRequestContext, + glossary: Glossary, + reviewer: UserClass +) => { + const response = await apiContext.patch( + `/api/v1/glossaries/${glossary.responseData.id}`, + { + data: [ + { + op: 'add', + path: '/reviewers', + value: [{ id: reviewer.responseData.id, type: 'user' }], + }, + ], + headers: { 'Content-Type': 'application/json-patch+json' }, + } + ); + + expect(response.status()).toBe(200); +}; + +const waitForInheritedReviewer = async ( + apiContext: APIRequestContext, + termId: string, + reviewerId: string +) => { + await expect + .poll( + async () => { + const response = await apiContext.get( + `/api/v1/glossaryTerms/${termId}?fields=reviewers` + ); + const term = await response.json(); + + return term.reviewers?.some( + (termReviewer: { id: string; inherited?: boolean }) => + termReviewer.id === reviewerId && termReviewer.inherited === true + ); + }, + { + message: `Term ${termId} must expose the reviewer inherited from its glossary`, + timeout: STATUS_TIMEOUT, + intervals: [3_000, 5_000, 10_000], + } + ) + .toBe(true); +}; + +test.describe( + 'Glossary Approval - inherited reviewers', + { tag: ['@Features', '@Governance'] }, + () => { + test('term inherits reviewer added to the glossary after an earlier term ran the workflow', async ({ + browser, + }) => { + test.slow(); + + const { apiContext, afterAction } = await performAdminLogin(browser); + const reviewer = new UserClass(); + const glossary = new Glossary(); + + try { + await reviewer.create(apiContext); + await glossary.create(apiContext); + const glossaryFqn = glossary.responseData.fullyQualifiedName; + + await test.step('A term created before any reviewer exists is auto-approved', async () => { + const seed = await createTermWithoutReviewers( + apiContext, + glossaryFqn, + `seed_before_reviewer_${uuid()}` + ); + + const status = await waitForSettledStatus(apiContext, seed.id); + + expect(status).toBe('Approved'); + }); + + await test.step('Add a reviewer to the glossary', async () => { + await addReviewerToGlossary(apiContext, glossary, reviewer); + }); + + await test.step('Every term created afterwards inherits the reviewer and gets an approval task', async () => { + for (let index = 0; index < REVIEWER_ADDED_PROBE_COUNT; index++) { + const term = await createTermWithoutReviewers( + apiContext, + glossaryFqn, + `after_reviewer_${index}_${uuid()}` + ); + + const status = await waitForSettledStatus(apiContext, term.id); + + expect( + status, + `Term ${term.name} inherits a reviewer from the glossary, so the approval workflow ` + + `must move it to "In Review". "Draft" means the reviewers gate saw none — the ` + + `inherited-reviewer bug.` + ).toBe('In Review'); + await waitForInheritedReviewer( + apiContext, + term.id, + reviewer.responseData.id + ); + + await expect + .poll( + () => + getOpenApprovalTaskCount(apiContext, term.fullyQualifiedName), + { + message: + `Term ${term.name} must have an open approval task for the inherited ` + + `reviewer`, + timeout: STATUS_TIMEOUT, + intervals: [3_000, 5_000, 10_000], + } + ) + .toBeGreaterThan(0); + } + }); + } finally { + await glossary.delete(apiContext); + await reviewer.delete(apiContext); + await afterAction(); + } + }); + + test('inherited reviewer is shown on the term page and it is not left in Draft', async ({ + browser, + page, + }) => { + test.slow(); + + const { apiContext, afterAction } = await performAdminLogin(browser); + const reviewer = new UserClass(); + const glossary = new Glossary(); + let term: GlossaryTerm | undefined; + + try { + await reviewer.create(apiContext); + await glossary.create(apiContext); + + await test.step('A term created before any reviewer exists is auto-approved', async () => { + const seed = await createTermWithoutReviewers( + apiContext, + glossary.responseData.fullyQualifiedName, + `seed_before_reviewer_${uuid()}` + ); + + expect(await waitForSettledStatus(apiContext, seed.id)).toBe( + 'Approved' + ); + }); + + await test.step('Add a reviewer to the glossary', async () => { + await addReviewerToGlossary(apiContext, glossary, reviewer); + }); + + // Built from a Glossary constructed without reviewers, so the term is created with an empty + // reviewers list and must inherit the one added to the glossary server-side. + term = new GlossaryTerm(glossary); + await term.create(apiContext); + + await test.step('Term exposes the inherited reviewer and reaches In Review', async () => { + expect( + await waitForSettledStatus(apiContext, term.responseData.id) + ).toBe('In Review'); + await waitForInheritedReviewer( + apiContext, + term.responseData.id, + reviewer.responseData.id + ); + }); + + await page.goto( + `/glossary/${encodeURIComponent( + term.responseData.fullyQualifiedName + )}` + ); + + await expect(page.locator('[data-testid="loader"]')).toHaveCount(0); + + await test.step('Inherited reviewer is displayed', async () => { + await expect( + page.getByTestId('glossary-reviewer').getByTestId('owner-link') + ).toContainText( + reviewer.responseData.displayName ?? reviewer.responseData.name + ); + }); + + await test.step('Term reached In Review, not Draft', async () => { + await expect(page.locator('.status-badge-label')).toContainText( + 'In Review' + ); + }); + } finally { + await glossary.delete(apiContext); + await reviewer.delete(apiContext); + await afterAction(); + } + }); + } +);