Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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<String, Object> variables) {
RuntimeService runtimeService = processEngine.getRuntimeService();
runtimeService.signalEventReceived(signal, variables);
try (FreshReadScope.Handle ignored = FreshReadScope.enter()) {
runtimeService.signalEventReceived(signal, variables);
}
}

private void unlockJobsOnStartup() {
Expand Down Expand Up @@ -922,6 +932,15 @@ public boolean resolveLegacyThreadTask(UUID customTaskId, Map<String, Object> va

private boolean resolveTaskInternal(
UUID customTaskId, Map<String, Object> 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<String, Object> 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
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
*
* <p>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.
*
* <p>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();
}
}
Comment thread
gitar-bot[bot] marked this conversation as resolved.

/** 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;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
Loading
Loading