diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandler.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandler.java index 76cc9944752d..c0b3127d2263 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandler.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandler.java @@ -5,25 +5,68 @@ import static org.openmetadata.service.governance.workflows.Workflow.GLOBAL_NAMESPACE; import static org.openmetadata.service.governance.workflows.Workflow.RELATED_ENTITY_ID_VARIABLE; +import com.fasterxml.jackson.core.type.TypeReference; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.Optional; import java.util.UUID; import lombok.extern.slf4j.Slf4j; +import org.flowable.common.engine.api.delegate.Expression; import org.flowable.engine.delegate.DelegateExecution; import org.flowable.task.service.delegate.DelegateTask; import org.flowable.variable.api.delegate.VariableScope; import org.openmetadata.schema.EntityInterface; import org.openmetadata.schema.type.Include; +import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.resources.feeds.MessageParser; @Slf4j public class WorkflowVariableHandler { private final VariableScope varScope; + private static final TypeReference> INPUT_NAMESPACE_TYPE = + new TypeReference<>() {}; public WorkflowVariableHandler(VariableScope varScope) { this.varScope = varScope; } + /** + * Per-input variable namespace map for a workflow node. A node can read several inputs from + * different node/global namespaces, so the wrapper is plural: InputNamespaces. + */ + public record InputNamespaces(Map namespaces) { + public InputNamespaces { + namespaces = + namespaces != null + ? Collections.unmodifiableMap(new LinkedHashMap<>(namespaces)) + : Map.of(); + } + + public String namespaceFor(String variable) { + return namespaces.get(variable); + } + + public String namespaceForOrDefault(String variable, String defaultNamespace) { + return namespaces.getOrDefault(variable, defaultNamespace); + } + + public static InputNamespaces from(Expression expr, DelegateExecution execution) { + return read(expr.getValue(execution)); + } + + public static InputNamespaces read(Object rawValue) { + // Flowable delivers process variables as untyped Object — a String when the caller stored + // the JSON form (taskService.setVariable(name, json)) or a Map when set via a delegate. + Map map = + rawValue instanceof String str + ? JsonUtils.readValue(str, INPUT_NAMESPACE_TYPE) + : JsonUtils.convertValue(rawValue, INPUT_NAMESPACE_TYPE); + return new InputNamespaces(map); + } + } + public static String getNamespacedVariableName(String namespace, String varName) { if (namespace != null) { return String.format("%s_%s", namespace, varName); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/CreateIngestionPipelineDelegate.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/CreateIngestionPipelineDelegate.java index 2b5ae64d0767..01e33e7ebf01 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/CreateIngestionPipelineDelegate.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/CreateIngestionPipelineDelegate.java @@ -8,7 +8,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.getResultFromBoolean; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -18,9 +17,9 @@ import org.openmetadata.schema.ServiceEntityInterface; import org.openmetadata.schema.entity.services.ingestionPipelines.PipelineType; import org.openmetadata.schema.type.Include; -import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; import org.openmetadata.service.resources.services.ingestionpipelines.IngestionPipelineMapper; @@ -35,8 +34,7 @@ public class CreateIngestionPipelineDelegate implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); PipelineType pipelineType = PipelineType.fromValue((String) pipelineTypeExpr.getValue(execution)); @@ -48,7 +46,8 @@ public void execute(DelegateExecution execution) { MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); ServiceEntityInterface service = Entity.getEntity(entityLink, "owners,ingestionRunner", Include.NON_DELETED); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineDelegate.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineDelegate.java index 58b3ecd8af2a..27ebca725d49 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineDelegate.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/createAndRunIngestionPipeline/RunIngestionPipelineDelegate.java @@ -7,7 +7,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.getResultFromBoolean; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import java.util.UUID; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; @@ -15,7 +14,6 @@ import org.flowable.engine.delegate.BpmnError; import org.flowable.engine.delegate.DelegateExecution; import org.flowable.engine.delegate.JavaDelegate; -import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.sdk.PipelineServiceClientInterface; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; @@ -31,9 +29,6 @@ public class RunIngestionPipelineDelegate implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); - boolean shouldRun = Boolean.parseBoolean((String) shouldRunExpr.getValue(execution)); boolean waitForCompletion = diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/ApplyRecognizerFeedbackImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/ApplyRecognizerFeedbackImpl.java index 52830acdb9d8..385b4ae801f8 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/ApplyRecognizerFeedbackImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/ApplyRecognizerFeedbackImpl.java @@ -6,7 +6,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -17,6 +16,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.RecognizerFeedbackRepository; @Slf4j @@ -27,17 +27,15 @@ public class ApplyRecognizerFeedbackImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - @SuppressWarnings("unchecked") - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String feedbackJson = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); + inputNamespaces.namespaceFor(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); RecognizerFeedback feedback = JsonUtils.readValue(feedbackJson, RecognizerFeedback.class); - String updatedByNamespace = inputNamespaceMap.get(UPDATED_BY_VARIABLE); + String updatedByNamespace = inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE); String reviewedBy = "governance-bot"; if (updatedByNamespace != null) { String actualUser = diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckChangeDescriptionTaskImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckChangeDescriptionTaskImpl.java index 32a60a85c88f..060592bbc18a 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckChangeDescriptionTaskImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckChangeDescriptionTaskImpl.java @@ -22,6 +22,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.governance.workflows.util.FieldChangeValueExtractor; import org.openmetadata.service.resources.feeds.MessageParser; @@ -35,12 +36,11 @@ public class CheckChangeDescriptionTaskImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String entityLinkStr = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); boolean result = checkChangeDescription(execution, entityLinkStr); varHandler.setNodeVariable(RESULT_VARIABLE, result); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImpl.java index 4751e34d5c90..999dda28fbe3 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImpl.java @@ -6,7 +6,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -17,6 +16,7 @@ import org.openmetadata.schema.type.Include; import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; import org.openmetadata.service.rules.RuleEngine; @@ -29,14 +29,14 @@ public class CheckEntityAttributesImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String rules = (String) rulesExpr.getValue(execution); MessageParser.EntityLink entityLink = MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); varHandler.setNodeVariable(RESULT_VARIABLE, checkAttributes(varHandler, entityLink, rules)); } catch (Exception exc) { LOG.error( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/DataCompletenessImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/DataCompletenessImpl.java index 1e9de4c16c58..e1f28bd7a71c 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/DataCompletenessImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/DataCompletenessImpl.java @@ -22,6 +22,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; @Slf4j @@ -35,8 +36,7 @@ public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { // Get configuration - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); List fieldsToCheck = JsonUtils.readOrConvertValue(fieldsToCheckExpr.getValue(execution), List.class); List> qualityBandMaps = @@ -53,7 +53,8 @@ public void execute(DelegateExecution execution) { MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); EntityInterface entity = Entity.getEntity(entityLink, "*", Include.ALL); Map entityMap = JsonUtils.getMap(entity); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RejectRecognizerFeedbackImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RejectRecognizerFeedbackImpl.java index fbc5a4d59f41..31e4fdb19a14 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RejectRecognizerFeedbackImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RejectRecognizerFeedbackImpl.java @@ -7,7 +7,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -18,6 +17,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.RecognizerFeedbackRepository; @Slf4j @@ -28,17 +28,15 @@ public class RejectRecognizerFeedbackImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - @SuppressWarnings("unchecked") - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String feedbackJson = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); + inputNamespaces.namespaceFor(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); RecognizerFeedback feedback = JsonUtils.readValue(feedbackJson, RecognizerFeedback.class); - String updatedByNamespace = inputNamespaceMap.get(UPDATED_BY_VARIABLE); + String updatedByNamespace = inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE); String reviewedBy = "governance-bot"; if (updatedByNamespace != null) { String actualUser = diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RollbackEntityImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RollbackEntityImpl.java index 07b2a311e038..ed6a16cbdac1 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RollbackEntityImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/RollbackEntityImpl.java @@ -6,7 +6,6 @@ import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.UUID; import lombok.extern.slf4j.Slf4j; import org.flowable.common.engine.api.delegate.Expression; @@ -20,6 +19,7 @@ import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.Workflow; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.EntityRepository; import org.openmetadata.service.resources.feeds.MessageParser; @@ -37,8 +37,7 @@ public void execute(DelegateExecution execution) { try { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); Object workflowInstanceExecutionIdObj = execution.getVariable(WORKFLOW_INSTANCE_EXECUTION_ID_VARIABLE); @@ -51,12 +50,13 @@ public void execute(DelegateExecution execution) { MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); String updatedBy = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE); + inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE); if (updatedBy == null || updatedBy.isEmpty()) { updatedBy = "governance-bot"; } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityAttributeImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityAttributeImpl.java index e1e0bc91db01..a4a32c2366c8 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityAttributeImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityAttributeImpl.java @@ -6,7 +6,6 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; @@ -16,8 +15,8 @@ import org.flowable.engine.delegate.JavaDelegate; import org.openmetadata.schema.EntityInterface; import org.openmetadata.schema.type.Include; -import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; import org.openmetadata.service.util.EntityFieldUtils; @@ -31,10 +30,8 @@ public class SetEntityAttributeImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - // Extract entity from workflow context - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); - String relatedEntityNamespace = (String) inputNamespaceMap.get(RELATED_ENTITY_VARIABLE); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); + String relatedEntityNamespace = inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE); String relatedEntityValue = (String) varHandler.getNamespacedVariable(relatedEntityNamespace, RELATED_ENTITY_VARIABLE); @@ -45,7 +42,6 @@ public void execute(DelegateExecution execution) { String fieldName = fieldNameExpr != null ? (String) fieldNameExpr.getValue(execution) : ""; - // Simple null check - if fieldValueExpr is null, treat as empty/null value String fieldValue = null; if (fieldValueExpr != null) { Object value = fieldValueExpr.getValue(execution); @@ -54,22 +50,18 @@ public void execute(DelegateExecution execution) { } } - String updatedByNamespace = (String) inputNamespaceMap.get(UPDATED_BY_VARIABLE); + String updatedByNamespace = inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE); String actualUser = Optional.ofNullable(updatedByNamespace) .map(ns -> (String) varHandler.getNamespacedVariable(ns, UPDATED_BY_VARIABLE)) .orElse(null); - // Apply the field change using shared utility with bot impersonation - // Note: fieldValue can be null to clear/remove a field value - // When actualUser is available, use it as the user and mark 'governance-bot' as impersonator - // Otherwise, use 'governance-bot' directly (for system-initiated workflows) + // fieldValue==null clears the field. When we have an acting user, preserve it and mark + // governance-bot as impersonator; otherwise attribute the write to governance-bot directly. if (actualUser != null && !actualUser.isEmpty()) { - // User-initiated workflow: preserve actual user, mark bot as impersonator EntityFieldUtils.setEntityField( entity, entityType, actualUser, fieldName, fieldValue, true, "governance-bot"); } else { - // System-initiated workflow: use governance-bot directly EntityFieldUtils.setEntityField( entity, entityType, "governance-bot", fieldName, fieldValue, true, null); } diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityCertificationImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityCertificationImpl.java index 3c2951962c08..0bb2ec994b02 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityCertificationImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetEntityCertificationImpl.java @@ -7,7 +7,6 @@ import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; import jakarta.json.JsonPatch; -import java.util.Map; import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; @@ -22,6 +21,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.EntityRepository; import org.openmetadata.service.resources.feeds.MessageParser; @@ -35,13 +35,13 @@ public class SetEntityCertificationImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); MessageParser.EntityLink entityLink = MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); String entityType = entityLink.getEntityType(); EntityInterface entity = Entity.getEntity(entityLink, "certification", Include.ALL); @@ -53,7 +53,7 @@ public void execute(DelegateExecution execution) { Optional.ofNullable( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE)) + inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE)) .orElse("governance-bot"); setStatus(entity, entityType, user, certification); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetGlossaryTermStatusImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetGlossaryTermStatusImpl.java index ca3eb27c588f..59293b2b4c2b 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetGlossaryTermStatusImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/SetGlossaryTermStatusImpl.java @@ -7,7 +7,6 @@ import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; import jakarta.json.JsonPatch; -import java.util.Map; import java.util.Optional; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; @@ -21,6 +20,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.GlossaryTermRepository; import org.openmetadata.service.resources.feeds.MessageParser; @@ -34,13 +34,13 @@ public class SetGlossaryTermStatusImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); MessageParser.EntityLink entityLink = MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); GlossaryTerm glossaryTerm = Entity.getEntity(entityLink, "", Include.ALL); String status = (String) statusExpr.getValue(execution); @@ -48,7 +48,7 @@ public void execute(DelegateExecution execution) { Optional.ofNullable( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE)) + inputNamespaces.namespaceFor(UPDATED_BY_VARIABLE), UPDATED_BY_VARIABLE)) .orElse("governance-bot"); setStatus(glossaryTerm, user, status); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/runApp/RunAppDelegate.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/runApp/RunAppDelegate.java index 4fdc9b45ff92..d2a4f2c0e00d 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/runApp/RunAppDelegate.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/runApp/RunAppDelegate.java @@ -6,16 +6,15 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; import org.flowable.engine.delegate.BpmnError; import org.flowable.engine.delegate.DelegateExecution; import org.flowable.engine.delegate.JavaDelegate; -import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.sdk.PipelineServiceClientInterface; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; @Slf4j @@ -30,8 +29,7 @@ public class RunAppDelegate implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String appName = (String) appNameExpr.getValue(execution); boolean waitForCompletion = @@ -45,7 +43,8 @@ public void execute(DelegateExecution execution) { MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); boolean wasSuccessful = new RunAppImpl() diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/sink/SinkTaskDelegate.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/sink/SinkTaskDelegate.java index 07baf9554318..6070c0f62591 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/sink/SinkTaskDelegate.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/sink/SinkTaskDelegate.java @@ -23,7 +23,6 @@ import com.google.common.collect.Lists; import java.util.ArrayList; import java.util.List; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -35,6 +34,7 @@ import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.resources.feeds.MessageParser; import org.openmetadata.service.workflows.searchIndex.ReindexingUtil; @@ -92,11 +92,10 @@ public void execute(DelegateExecution execution) { ? Integer.parseInt((String) timeoutSecondsExpr.getValue(execution)) : 300; // Default 5 minutes - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); // Check if we have an entity list for batch processing - String entityListNamespace = inputNamespaceMap.get(ENTITY_LIST_VARIABLE); + String entityListNamespace = inputNamespaces.namespaceFor(ENTITY_LIST_VARIABLE); List entityList = null; if (entityListNamespace != null) { Object entityListObj = @@ -143,7 +142,7 @@ public void execute(DelegateExecution execution) { result = executeBatchMode(context, sinkProvider, entityList); } else { // Single entity mode: process one entity - result = executeSingleEntityMode(context, sinkProvider, inputNamespaceMap, varHandler); + result = executeSingleEntityMode(context, sinkProvider, inputNamespaces, varHandler); } // Set output variables @@ -283,11 +282,11 @@ BatchAccumulator merge(SinkResult result, List fetchErrors private SinkResult executeSingleEntityMode( SinkContext context, SinkProvider sinkProvider, - Map inputNamespaceMap, + InputNamespaces inputNamespaces, WorkflowVariableHandler varHandler) { // Get entity from workflow context - String relatedEntityNamespace = inputNamespaceMap.get(RELATED_ENTITY_VARIABLE); + String relatedEntityNamespace = inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE); String relatedEntityValue = (String) varHandler.getNamespacedVariable(relatedEntityNamespace, RELATED_ENTITY_VARIABLE); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java index 2ca98ea42746..81fd977e9753 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java @@ -20,6 +20,7 @@ import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION; import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId; +import com.fasterxml.jackson.annotation.JsonInclude; import io.github.resilience4j.core.IntervalFunction; import io.github.resilience4j.retry.Retry; import io.github.resilience4j.retry.RetryConfig; @@ -29,9 +30,7 @@ import java.time.ZonedDateTime; import java.time.format.DateTimeParseException; import java.util.ArrayList; -import java.util.LinkedHashMap; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -50,6 +49,7 @@ import org.openmetadata.schema.entity.tasks.Task; import org.openmetadata.schema.governance.workflows.WorkflowDefinition; import org.openmetadata.schema.type.ChangeEvent; +import org.openmetadata.schema.type.DataAccessRequestPayload; import org.openmetadata.schema.type.EntityReference; import org.openmetadata.schema.type.EventType; import org.openmetadata.schema.type.Include; @@ -67,6 +67,7 @@ import org.openmetadata.service.exception.EntityNotFoundException; import org.openmetadata.service.governance.workflows.WorkflowHandler; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.governance.workflows.elements.TriggerFactory; import org.openmetadata.service.governance.workflows.elements.nodes.userTask.helper.WorkflowVariableResolver; import org.openmetadata.service.governance.workflows.util.ChangePreviewUtils; @@ -124,14 +125,15 @@ public class CreateTask implements TaskListener { public void notify(DelegateTask delegateTask) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(delegateTask); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(delegateTask), Map.class); + InputNamespaces inputNamespaces = + InputNamespaces.read(inputNamespaceMapExpr.getValue(delegateTask)); List assignees = getAssignees(delegateTask); MessageParser.EntityLink entityLink = MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); EntityInterface entity = Entity.getEntity(entityLink, "*", Include.ALL); // Get approval threshold, default to 1 if not set @@ -148,7 +150,7 @@ public void notify(DelegateTask delegateTask) { UUID workflowInstanceId = getWorkflowInstanceId(delegateTask); // Build workflow-specific payload for task types that need richer context. - Object payload = buildWorkflowPayload(taskType, inputNamespaceMap, varHandler); + Object payload = buildWorkflowPayload(taskType, inputNamespaces, varHandler); // Create or update the Task entity for the current workflow stage Task task = @@ -200,20 +202,17 @@ private TaskEntityType getTaskType(DelegateTask delegateTask) { if (variableTaskType != null && !variableTaskType.isEmpty()) { return TaskEntityType.fromValue(variableTaskType); } - if (taskTypeExpr != null) { String typeStr = (String) taskTypeExpr.getValue(delegateTask); if (typeStr != null && !typeStr.isEmpty()) { return TaskEntityType.fromValue(typeStr); } } - TaskEntityType inferredTaskType = inferTaskTypeFromWorkflow(delegateTask); if (inferredTaskType != null) { return inferredTaskType; } - - return TaskEntityType.GlossaryApproval; // Default for backward compatibility + return TaskEntityType.GlossaryApproval; // Backward-compat default } private TaskCategory getTaskCategory(DelegateTask delegateTask) { @@ -222,23 +221,21 @@ private TaskCategory getTaskCategory(DelegateTask delegateTask) { if (variableTaskCategory != null && !variableTaskCategory.isEmpty()) { return TaskCategory.fromValue(variableTaskCategory); } - if (taskCategoryExpr != null) { String categoryStr = (String) taskCategoryExpr.getValue(delegateTask); if (categoryStr != null && !categoryStr.isEmpty()) { return TaskCategory.fromValue(categoryStr); } } - TaskEntityType inferredTaskType = inferTaskTypeFromWorkflow(delegateTask); if (inferredTaskType != null) { - return TaskWorkflowLifecycleResolver.defaultTaskCategoryForWorkflowDefinitionRef( + String workflowRef = inferredTaskType == TaskEntityType.CustomTask ? "CustomTaskWorkflow" - : inferWorkflowDefinitionRef(delegateTask)); + : inferWorkflowDefinitionRef(delegateTask); + return TaskWorkflowLifecycleResolver.defaultTaskCategoryForWorkflowDefinitionRef(workflowRef); } - - return TaskCategory.Approval; // Default for backward compatibility + return TaskCategory.Approval; // Backward-compat default } private TaskEntityType inferTaskTypeFromWorkflow(DelegateTask delegateTask) { @@ -246,7 +243,6 @@ private TaskEntityType inferTaskTypeFromWorkflow(DelegateTask delegateTask) { if (workflowDefinitionRef == null || workflowDefinitionRef.isBlank()) { return null; } - return TaskWorkflowLifecycleResolver.defaultTaskTypeForWorkflowDefinitionRef( workflowDefinitionRef); } @@ -257,7 +253,6 @@ private String inferWorkflowDefinitionRef(DelegateTask delegateTask) { if (processDefinitionKey == null || processDefinitionKey.isBlank()) { return null; } - return processDefinitionKey.endsWith("Trigger") ? TriggerFactory.getMainWorkflowDefinitionNameFromTrigger(processDefinitionKey) : processDefinitionKey; @@ -268,12 +263,10 @@ private UUID resolveWorkflowDefinitionId( if (workflowDefinitionIdValue != null && !workflowDefinitionIdValue.isBlank()) { return UUID.fromString(workflowDefinitionIdValue); } - String workflowDefinitionRef = inferWorkflowDefinitionRef(delegateTask); if (workflowDefinitionRef == null || workflowDefinitionRef.isBlank()) { return null; } - WorkflowDefinition workflowDefinition = Entity.findByNameOrNull( Entity.WORKFLOW_DEFINITION, workflowDefinitionRef, Include.NON_DELETED); @@ -281,30 +274,25 @@ private UUID resolveWorkflowDefinitionId( } private UUID getWorkflowInstanceId(DelegateTask delegateTask) { - // First prefer an explicit runtime variable when one is present. Object workflowInstanceIdObj = delegateTask.getVariable("workflowInstanceId"); if (workflowInstanceIdObj != null) { return UUID.fromString(workflowInstanceIdObj.toString()); } + String businessKey = resolveProcessBusinessKey(delegateTask.getProcessInstanceId()); + return businessKey != null && !businessKey.isBlank() ? UUID.fromString(businessKey) : null; + } - String processInstanceId = delegateTask.getProcessInstanceId(); + private static String resolveProcessBusinessKey(String processInstanceId) { if (processInstanceId == null || processInstanceId.isBlank()) { return null; } - org.flowable.engine.runtime.ProcessInstance processInstance = WorkflowHandler.getInstance() .getRuntimeService() .createProcessInstanceQuery() .processInstanceId(processInstanceId) .singleResult(); - - String businessKey = processInstance != null ? processInstance.getBusinessKey() : null; - if (businessKey == null || businessKey.isBlank()) { - return null; - } - - return UUID.fromString(businessKey); + return processInstance != null ? processInstance.getBusinessKey() : null; } private List getAssignees(DelegateTask delegateTask) { @@ -434,7 +422,8 @@ private Task createOrUpdateTask( .withFullyQualifiedName(entity.getFullyQualifiedName()); // Build createdBy reference - EntityReference createdByRef = resolveCreatedByReference(requestedCreatedBy, entity, payload); + EntityReference createdByRef = + resolveCreatedByReference(requestedCreatedBy, entity, payload, taskType); String updatedBy = requestedUpdatedBy != null && !requestedUpdatedBy.isBlank() ? requestedUpdatedBy @@ -524,11 +513,12 @@ private Task createOrUpdateTask( updatedTask.setPriority(requestedPriority); } Long effectiveDueDate = - resolveEffectiveDueDate(stageStatus, requestedPayload, requestedDueDate); + resolveEffectiveDueDate(stageStatus, taskType, requestedPayload, requestedDueDate); if (effectiveDueDate != null) { updatedTask.setDueDate(effectiveDueDate); } - updatedTask.setPayload(withGrantExpirationDate(stageStatus, updatedTask.getPayload())); + updatedTask.setPayload( + withGrantExpirationDate(stageStatus, taskType, updatedTask.getPayload())); updatedTask.setPayload( applyProposedChangesIfApproval(taskType, entity, updatedTask.getPayload())); if (requestedExternalReference != null) { @@ -593,11 +583,11 @@ private Task createOrUpdateTask( task.setTaskFormSchemaVersion(taskFormSchemaVersion); } Long effectiveDueDate = - resolveEffectiveDueDate(stageStatus, requestedPayload, requestedDueDate); + resolveEffectiveDueDate(stageStatus, taskType, requestedPayload, requestedDueDate); if (effectiveDueDate != null) { task.setDueDate(effectiveDueDate); } - task.setPayload(withGrantExpirationDate(stageStatus, task.getPayload())); + task.setPayload(withGrantExpirationDate(stageStatus, taskType, task.getPayload())); task.setPayload(applyProposedChangesIfApproval(taskType, entity, task.getPayload())); if (requestedExternalReference != null) { task.setExternalReference( @@ -783,23 +773,20 @@ static List resolveExistingTaskAssignees( List requestedAssignees) { List existingAssignees = existingTask.getAssignees(); boolean hasExistingAssignees = existingAssignees != null && !existingAssignees.isEmpty(); - - // For API-created workflow-managed tasks, taskAssignees is seeded into the workflow start - // variables. Subsequent workflow callbacks must not overwrite the task row's current - // assignees with BPMN candidate users or the original start-variable snapshot; the persisted - // task row is the source of truth once assignees are present. - if (requestedAssignees != null && !requestedAssignees.isEmpty() && hasExistingAssignees) { + boolean hasRequestedAssignees = requestedAssignees != null && !requestedAssignees.isEmpty(); + boolean hasWorkflowAssignees = workflowAssignees != null && !workflowAssignees.isEmpty(); + // API-created workflow-managed tasks seed taskAssignees into workflow start variables. The + // persisted task row is the source of truth once assignees are set — subsequent workflow + // callbacks must not overwrite it with BPMN candidates or the original snapshot. + if (hasRequestedAssignees && hasExistingAssignees) { return null; } - - if (workflowAssignees != null && !workflowAssignees.isEmpty()) { + if (hasWorkflowAssignees) { return workflowAssignees; } - - if (requestedAssignees != null && !requestedAssignees.isEmpty() && !hasExistingAssignees) { + if (hasRequestedAssignees) { return requestedAssignees; } - return existingAssignees; } @@ -859,27 +846,12 @@ private UUID resolveRequestedTaskId(DelegateTask delegateTask) { if (taskId != null && !taskId.isBlank()) { return UUID.fromString(taskId); } - - String processInstanceId = delegateTask.getProcessInstanceId(); - if (processInstanceId == null || processInstanceId.isBlank()) { - return null; - } - - org.flowable.engine.runtime.ProcessInstance processInstance = - WorkflowHandler.getInstance() - .getRuntimeService() - .createProcessInstanceQuery() - .processInstanceId(processInstanceId) - .singleResult(); - String businessKey = processInstance != null ? processInstance.getBusinessKey() : null; - + String businessKey = resolveProcessBusinessKey(delegateTask.getProcessInstanceId()); if (businessKey == null || businessKey.isBlank()) { return null; } - LOG.debug( "[CreateTask] Falling back to process business key '{}' as requested task id", businessKey); - return UUID.fromString(businessKey); } @@ -892,66 +864,82 @@ private TaskEntityStatus resolveStageStatus(DelegateTask delegateTask) { } static Long resolveEffectiveDueDate( - TaskEntityStatus stageStatus, Object payload, Long requestedDueDate) { - if (stageStatus != TaskEntityStatus.Granted || payload == null) { - return requestedDueDate; - } - if (!(payload instanceof Map rawMap)) { + TaskEntityStatus stageStatus, + TaskEntityType taskType, + Object payload, + Long requestedDueDate) { + if (!isGrantedDataAccessRequest(stageStatus, taskType, payload)) { return requestedDueDate; } - Object durationValue = rawMap.get("duration"); - if (!(durationValue instanceof String duration) || duration.isBlank()) { + String duration = readDataAccessRequestPayload(payload).getDuration(); + if (duration == null || duration.isBlank()) { return requestedDueDate; } return parseMillisFromIso8601Duration(duration, requestedDueDate); } /** - * Compute {@code payload.expirationDate} when the workflow enters the Granted stage. - * - *

Anchoring this on stage entry (instead of on the user-invoked transition) makes - * both the manual path (Approved → markAsGranted → Granted) and the automated PolicyAgent - * path (Review → approve → PolicyAgent[granted] → Granted) populate expirationDate, since - * both routes funnel through this listener. - * - *

Returns null when the stage isn't Granted or the payload doesn't carry a parseable - * duration. When the payload already carries a numeric {@code expirationDate} (re-entry into - * the Granted stage, or upstream-set value), returns that existing value unchanged — silent - * overwrites would extend access on every workflow listener fire. + * Compute {@code payload.expirationDate} when the workflow enters the Granted stage. Anchoring + * on stage entry (instead of on the user-invoked transition) covers both the manual path + * (Approved → markAsGranted → Granted) and the PolicyAgent path (Review → approve → + * PolicyAgent[granted] → Granted). Returns null when the stage isn't Granted or the payload + * doesn't carry a parseable duration. Preserves an already-set {@code expirationDate} so a + * re-entry into Granted doesn't silently extend access. */ - static Long resolveEffectiveExpirationDate(TaskEntityStatus stageStatus, Object payload) { - if (stageStatus != TaskEntityStatus.Granted || !(payload instanceof Map rawMap)) { + static Long resolveEffectiveExpirationDate( + TaskEntityStatus stageStatus, TaskEntityType taskType, Object payload) { + if (!isGrantedDataAccessRequest(stageStatus, taskType, payload)) { return null; } - Object existing = rawMap.get("expirationDate"); - if (existing instanceof Number existingNum) { - return existingNum.longValue(); + DataAccessRequestPayload darPayload = readDataAccessRequestPayload(payload); + if (darPayload.getExpirationDate() != null) { + return darPayload.getExpirationDate(); } - Object durationValue = rawMap.get("duration"); - if (!(durationValue instanceof String duration) || duration.isBlank()) { + String duration = darPayload.getDuration(); + if (duration == null || duration.isBlank()) { return null; } return parseMillisFromIso8601Duration(duration, null); } /** - * Returns the payload with {@code expirationDate} merged in when the task enters - * Granted. Returns the original reference unchanged when there's nothing to add, so - * non-DAR workflows that target Granted aren't penalised and the input map is never - * mutated. + * Returns the payload with {@code expirationDate} merged in when the task enters Granted. + * Returns the original reference unchanged when there's nothing to add, so non-DAR workflows + * that target Granted aren't penalised and the input map is never mutated. */ - static Object withGrantExpirationDate(TaskEntityStatus stageStatus, Object payload) { - Long expiration = resolveEffectiveExpirationDate(stageStatus, payload); - if (expiration == null || !(payload instanceof Map rawMap)) { + static Object withGrantExpirationDate( + TaskEntityStatus stageStatus, TaskEntityType taskType, Object payload) { + Long expiration = resolveEffectiveExpirationDate(stageStatus, taskType, payload); + if (expiration == null) { return payload; } - if (rawMap.get("expirationDate") instanceof Number) { + DataAccessRequestPayload darPayload = readDataAccessRequestPayload(payload); + if (darPayload.getExpirationDate() != null) { return payload; } - Map merged = new LinkedHashMap<>(); - rawMap.forEach((k, v) -> merged.put(String.valueOf(k), v)); - merged.put("expirationDate", expiration); - return merged; + return darPayload.withExpirationDate(expiration); + } + + private static boolean isGrantedDataAccessRequest( + TaskEntityStatus stageStatus, TaskEntityType taskType, Object payload) { + return stageStatus == TaskEntityStatus.Granted + && taskType == TaskEntityType.DataAccessRequest + && payload != null; + } + + private static DataAccessRequestPayload readDataAccessRequestPayload(Object payload) { + // Task.payload is Object in the schema because the shape varies per taskType. The value + // reaches this method as a raw JSON String (Flowable variable serialization) or as a + // deserialized Map/POJO (repository read); handle both here rather than at every caller. + try { + if (payload instanceof String json) { + return JsonUtils.readValue(json, DataAccessRequestPayload.class); + } + return JsonUtils.convertValueLenient(payload, DataAccessRequestPayload.class); + } catch (RuntimeException invalidPayload) { + LOG.trace("[CreateTask] Payload is not a DataAccessRequestPayload", invalidPayload); + throw new IllegalArgumentException("Invalid DataAccessRequest task payload", invalidPayload); + } } /** @@ -970,17 +958,18 @@ static Object applyProposedChangesIfApproval( } static Long parseMillisFromIso8601Duration(String duration, Long fallback) { + // Duration.parse handles PT...H/M/S (time-based); Period covers PnD/M/Y (calendar-based). try { return System.currentTimeMillis() + Duration.parse(duration).toMillis(); - } catch (DateTimeParseException ignored) { - // Duration.parse does not support month/year designators; try Period + } catch (DateTimeParseException notDuration) { + // fall through to Period } try { return ZonedDateTime.now(ZoneOffset.UTC) .plus(Period.parse(duration)) .toInstant() .toEpochMilli(); - } catch (DateTimeParseException e) { + } catch (DateTimeParseException notPeriod) { LOG.warn( "[CreateTask] Could not parse ISO-8601 duration '{}'; falling back to caller default", duration); @@ -1001,21 +990,21 @@ private String buildTaskDescription(EntityInterface entity, TaskEntityType taskT } private EntityReference resolveCreatedByReference( - EntityReference requestedCreatedBy, EntityInterface entity, Object payload) { + EntityReference requestedCreatedBy, + EntityInterface entity, + Object payload, + TaskEntityType taskType) { if (requestedCreatedBy != null && requestedCreatedBy.getId() != null) { return requestedCreatedBy; } - - EntityReference payloadCreator = extractPayloadCreatedBy(payload); + EntityReference payloadCreator = extractPayloadCreatedBy(payload, taskType); if (payloadCreator != null) { return payloadCreator; } - String userName = entity != null ? entity.getUpdatedBy() : null; if (userName == null || userName.isEmpty()) { userName = DEFAULT_SYSTEM_USER; } - try { return Entity.getEntityReferenceByName(Entity.USER, userName, Include.NON_DELETED); } catch (Exception e) { @@ -1023,25 +1012,35 @@ private EntityReference resolveCreatedByReference( } } - private EntityReference extractPayloadCreatedBy(Object payload) { - if (!(payload instanceof Map payloadMap)) { + static EntityReference extractPayloadCreatedBy(Object payload, TaskEntityType taskType) { + if (taskType != TaskEntityType.RecognizerFeedbackApproval + && taskType != TaskEntityType.DataQualityReview) { return null; } - - Object feedback = payloadMap.get("feedback"); - if (feedback == null) { + RecognizerFeedbackTaskPayload feedbackPayload = readRecognizerFeedbackTaskPayload(payload); + RecognizerFeedback feedback = feedbackPayload != null ? feedbackPayload.feedback() : null; + if (feedback == null || feedback.getCreatedBy() == null) { return null; } + return feedback.getCreatedBy(); + } + private static RecognizerFeedbackTaskPayload readRecognizerFeedbackTaskPayload(Object payload) { + // Task.payload is Object in the schema because the shape varies per taskType. The value + // reaches this method as a raw JSON String (Flowable variable serialization), the concrete + // POJO (in-process reuse), or a deserialized Map (repository read); handle all three here. try { - RecognizerFeedback recognizerFeedback = - JsonUtils.convertValue(feedback, RecognizerFeedback.class); - if (recognizerFeedback == null || recognizerFeedback.getCreatedBy() == null) { - return null; + if (payload instanceof RecognizerFeedbackTaskPayload feedbackPayload) { + return feedbackPayload; } - return recognizerFeedback.getCreatedBy(); - } catch (Exception e) { - return null; + if (payload instanceof String json) { + return JsonUtils.readValueLenient(json, RecognizerFeedbackTaskPayload.class); + } + return JsonUtils.convertValueLenient(payload, RecognizerFeedbackTaskPayload.class); + } catch (RuntimeException invalidPayload) { + LOG.trace("[CreateTask] Payload is not a RecognizerFeedbackTaskPayload", invalidPayload); + throw new IllegalArgumentException( + "Invalid recognizer feedback task payload", invalidPayload); } } @@ -1059,39 +1058,32 @@ private String resolveUpdatedBy(EntityInterface entity, EntityReference createdB private Object buildWorkflowPayload( TaskEntityType taskType, - Map inputNamespaceMap, + InputNamespaces inputNamespaces, WorkflowVariableHandler varHandler) { - if ((taskType != TaskEntityType.RecognizerFeedbackApproval - && taskType != TaskEntityType.DataQualityReview) - || inputNamespaceMap == null) { + if (taskType != TaskEntityType.RecognizerFeedbackApproval + && taskType != TaskEntityType.DataQualityReview) { return null; } - String recognizerNamespace = - inputNamespaceMap.getOrDefault(RECOGNIZER_FEEDBACK, GLOBAL_NAMESPACE); - + inputNamespaces.namespaceForOrDefault(RECOGNIZER_FEEDBACK, GLOBAL_NAMESPACE); try { String feedbackJson = (String) varHandler.getNamespacedVariable(recognizerNamespace, RECOGNIZER_FEEDBACK); if (feedbackJson == null || feedbackJson.isEmpty()) { return null; } - RecognizerFeedback feedback = JsonUtils.readValue(feedbackJson, RecognizerFeedback.class); - Map payload = new LinkedHashMap<>(); - payload.put("feedback", feedback); - - TagLabelRecognizerMetadata recognizer = resolveRecognizerMetadata(feedback); - if (recognizer != null) { - payload.put("recognizer", recognizer); - } - return payload; + return new RecognizerFeedbackTaskPayload(feedback, resolveRecognizerMetadata(feedback)); } catch (Exception e) { LOG.warn("Failed to build recognizer feedback payload for task: {}", e.getMessage()); return null; } } + @JsonInclude(JsonInclude.Include.NON_NULL) + private record RecognizerFeedbackTaskPayload( + RecognizerFeedback feedback, TagLabelRecognizerMetadata recognizer) {} + private void terminateDeletedWorkflowManagedDraftTask( DelegateTask delegateTask, UUID requestedTaskId) { String processInstanceId = delegateTask.getProcessInstanceId(); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/AutoApproveServiceTaskImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/AutoApproveServiceTaskImpl.java index 425f3387349b..8a941ffd402a 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/AutoApproveServiceTaskImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/AutoApproveServiceTaskImpl.java @@ -2,14 +2,13 @@ import static org.openmetadata.service.governance.workflows.Workflow.RELATED_ENTITY_VARIABLE; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.flowable.common.engine.api.delegate.Expression; import org.flowable.engine.delegate.DelegateExecution; import org.flowable.engine.delegate.JavaDelegate; -import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.TaskRepository; import org.openmetadata.service.resources.feeds.MessageParser; @@ -38,12 +37,11 @@ public void execute(DelegateExecution execution) { if (inputNamespaceMapExpr != null) { try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String entityInfo = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); if (entityInfo != null) { MessageParser.EntityLink entityLink = MessageParser.EntityLink.parse(entityInfo); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CheckFeedbackSubmitterIsReviewerImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CheckFeedbackSubmitterIsReviewerImpl.java index 1d09434405dd..73820827268d 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CheckFeedbackSubmitterIsReviewerImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CheckFeedbackSubmitterIsReviewerImpl.java @@ -7,7 +7,6 @@ import com.fasterxml.jackson.core.type.TypeReference; import java.util.List; -import java.util.Map; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.exception.ExceptionUtils; import org.flowable.common.engine.api.delegate.Expression; @@ -17,6 +16,7 @@ import org.openmetadata.schema.type.RecognizerFeedback; import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; @Slf4j public class CheckFeedbackSubmitterIsReviewerImpl implements JavaDelegate { @@ -30,13 +30,12 @@ public void execute(DelegateExecution execution) { execution.getProcessInstanceId()); WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); String feedbackJson = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); + inputNamespaces.namespaceFor(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); RecognizerFeedback feedback = JsonUtils.readValue(feedbackJson, RecognizerFeedback.class); String assigneesVarName = assigneesVarNameExpr.getValue(execution).toString(); diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CreateRecognizerFeedbackApprovalTaskImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CreateRecognizerFeedbackApprovalTaskImpl.java index 4bfdfa87fcdc..e93b8134d099 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CreateRecognizerFeedbackApprovalTaskImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/CreateRecognizerFeedbackApprovalTaskImpl.java @@ -8,7 +8,6 @@ import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.UUID; import lombok.extern.slf4j.Slf4j; @@ -37,6 +36,7 @@ import org.openmetadata.service.exception.EntityNotFoundException; import org.openmetadata.service.governance.workflows.WorkflowHandler; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.CollectionDAO; import org.openmetadata.service.jdbi3.FeedRepository; import org.openmetadata.service.resources.feeds.FeedMapper; @@ -57,20 +57,20 @@ public void notify(DelegateTask delegateTask) { delegateTask.getId()); WorkflowVariableHandler varHandler = new WorkflowVariableHandler(delegateTask); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(delegateTask), Map.class); + InputNamespaces inputNamespaces = + InputNamespaces.read(inputNamespaceMapExpr.getValue(delegateTask)); List assignees = getAssignees(delegateTask); String feedbackJson = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); + inputNamespaces.namespaceFor(RECOGNIZER_FEEDBACK), RECOGNIZER_FEEDBACK); RecognizerFeedback feedback = JsonUtils.readValue(feedbackJson, RecognizerFeedback.class); String tagEntityLink = (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE); MessageParser.EntityLink tagLink = MessageParser.EntityLink.parse(tagEntityLink); Tag tag = Entity.getEntityByName( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/SetApprovalAssigneesImpl.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/SetApprovalAssigneesImpl.java index 66dde8cf7f72..472726aea4bc 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/SetApprovalAssigneesImpl.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/impl/SetApprovalAssigneesImpl.java @@ -32,6 +32,7 @@ import org.openmetadata.schema.utils.ResultList; import org.openmetadata.service.Entity; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.jdbi3.EntityRepository; import org.openmetadata.service.jdbi3.ListFilter; import org.openmetadata.service.jdbi3.TaskRepository; @@ -51,8 +52,7 @@ public class SetApprovalAssigneesImpl implements JavaDelegate { public void execute(DelegateExecution execution) { WorkflowVariableHandler varHandler = new WorkflowVariableHandler(execution); try { - Map inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); Map assigneesConfig = JsonUtils.readOrConvertValue(assigneesExpr.getValue(execution), Map.class); @@ -61,7 +61,8 @@ public void execute(DelegateExecution execution) { MessageParser.EntityLink.parse( (String) varHandler.getNamespacedVariable( - inputNamespaceMap.get(RELATED_ENTITY_VARIABLE), RELATED_ENTITY_VARIABLE)); + inputNamespaces.namespaceFor(RELATED_ENTITY_VARIABLE), + RELATED_ENTITY_VARIABLE)); EntityRepository entityRepository = Entity.getEntityRepository(entityLink.getEntityType()); boolean entitySupportsReviewers = entityRepository.isSupportsReviewers(); String relationshipFields = diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/BaseDelegate.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/BaseDelegate.java index ca87a584ece8..e53e2ff8eff6 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/BaseDelegate.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/BaseDelegate.java @@ -12,6 +12,7 @@ import org.flowable.engine.delegate.JavaDelegate; import org.openmetadata.schema.utils.JsonUtils; import org.openmetadata.service.governance.workflows.WorkflowVariableHandler; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; @Slf4j public abstract class BaseDelegate implements JavaDelegate { @@ -19,7 +20,7 @@ public abstract class BaseDelegate implements JavaDelegate { private Expression configMapExpr; protected WorkflowVariableHandler varHandler; - protected Map inputNamespaceMap; + protected InputNamespaces inputNamespaces; protected Map configMap; protected abstract void innerExecute(DelegateExecution execution); @@ -33,8 +34,7 @@ public void execute(DelegateExecution execution) { varHandler = new WorkflowVariableHandler(execution); try { - inputNamespaceMap = - JsonUtils.readOrConvertValue(inputNamespaceMapExpr.getValue(execution), Map.class); + inputNamespaces = InputNamespaces.from(inputNamespaceMapExpr, execution); configMap = JsonUtils.readOrConvertValue(configMapExpr.getValue(execution), Map.class); LOG.debug( diff --git a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/MainWorkflow.java b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/MainWorkflow.java index 59f108a43dc9..95d5e59eadfa 100644 --- a/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/MainWorkflow.java +++ b/openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/flowable/MainWorkflow.java @@ -19,7 +19,7 @@ import org.openmetadata.schema.governance.workflows.WorkflowDefinition; import org.openmetadata.schema.governance.workflows.elements.EdgeDefinition; import org.openmetadata.schema.governance.workflows.elements.WorkflowNodeDefinitionInterface; -import org.openmetadata.schema.utils.JsonUtils; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; import org.openmetadata.service.governance.workflows.elements.Edge; import org.openmetadata.service.governance.workflows.elements.NodeFactory; import org.openmetadata.service.governance.workflows.elements.NodeInterface; @@ -107,15 +107,9 @@ public WorkflowGraph(WorkflowDefinition workflowDefinition) { } private void validateNode(WorkflowNodeDefinitionInterface nodeDefinition) { - Map inputNamespaceMap = - (Map) - JsonUtils.readOrConvertValue(nodeDefinition.getInputNamespaceMap(), Map.class); + InputNamespaces inputNamespaces = InputNamespaces.read(nodeDefinition.getInputNamespaceMap()); - if (inputNamespaceMap == null) { - return; - } - - for (Map.Entry entry : inputNamespaceMap.entrySet()) { + for (Map.Entry entry : inputNamespaces.namespaces().entrySet()) { String variable = entry.getKey(); String namespace = entry.getValue(); diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandlerInputNamespacesTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandlerInputNamespacesTest.java new file mode 100644 index 000000000000..ed5bcd1639b3 --- /dev/null +++ b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/WorkflowVariableHandlerInputNamespacesTest.java @@ -0,0 +1,76 @@ +/* + * Copyright 2025 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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.HashMap; +import java.util.Map; +import org.flowable.common.engine.api.delegate.Expression; +import org.flowable.engine.delegate.DelegateExecution; +import org.junit.jupiter.api.Test; +import org.openmetadata.schema.utils.JsonUtils; +import org.openmetadata.service.governance.workflows.WorkflowVariableHandler.InputNamespaces; + +class WorkflowVariableHandlerInputNamespacesTest { + + @Test + void readAcceptsJsonFieldExtension() { + InputNamespaces namespaces = + InputNamespaces.read( + JsonUtils.pojoToJson(Map.of("relatedEntity", "global", "updatedBy", "task"))); + + assertEquals("global", namespaces.namespaceFor("relatedEntity")); + assertEquals("task", namespaces.namespaceFor("updatedBy")); + } + + @Test + void readAcceptsConvertedMapFieldExtension() { + InputNamespaces namespaces = InputNamespaces.read(Map.of("relatedEntity", "global")); + + assertEquals("global", namespaces.namespaceFor("relatedEntity")); + assertNull(namespaces.namespaceFor("updatedBy")); + } + + @Test + void readTreatsMissingExtensionAsEmpty() { + InputNamespaces namespaces = InputNamespaces.read(null); + + assertNull(namespaces.namespaceFor("relatedEntity")); + } + + @Test + void readToleratesNullNamespaceValues() { + Map rawNamespaces = new HashMap<>(); + rawNamespaces.put("relatedEntity", null); + + InputNamespaces namespaces = InputNamespaces.read(rawNamespaces); + + assertNull(namespaces.namespaceFor("relatedEntity")); + } + + @Test + void fromReadsExpressionValue() { + Expression expression = mock(Expression.class); + DelegateExecution execution = mock(DelegateExecution.class); + when(expression.getValue(execution)).thenReturn(Map.of("relatedEntity", "global")); + + InputNamespaces namespaces = InputNamespaces.from(expression, execution); + + assertEquals("global", namespaces.namespaceFor("relatedEntity")); + } +} diff --git a/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTaskTest.java b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTaskTest.java index 23b9fe3f0e00..7030300cd0b4 100644 --- a/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTaskTest.java +++ b/openmetadata-service/src/test/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTaskTest.java @@ -19,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -34,9 +35,11 @@ import org.junit.jupiter.api.Test; import org.mockito.Mockito; import org.openmetadata.schema.entity.tasks.Task; +import org.openmetadata.schema.type.DataAccessRequestPayload; import org.openmetadata.schema.type.EntityReference; import org.openmetadata.schema.type.Include; import org.openmetadata.schema.type.TaskEntityStatus; +import org.openmetadata.schema.type.TaskEntityType; import org.openmetadata.service.exception.EntityNotFoundException; import org.openmetadata.service.jdbi3.TaskRepository; @@ -227,31 +230,44 @@ void testResolveEffectiveDueDatePreservesRequestedDueDateForNonGrantedStatus() { assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Approved, Map.of("duration", "P14D"), requested)); + TaskEntityStatus.Approved, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"), + requested)); assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Open, Map.of("duration", "P14D"), requested)); + TaskEntityStatus.Open, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"), + requested)); } @Test void testResolveEffectiveDueDatePreservesRequestedDueDateWhenPayloadIsNull() { Long requested = 999L; assertEquals( - requested, CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, null, requested)); + requested, + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, null, requested)); } @Test - void testResolveEffectiveDueDatePreservesRequestedDueDateForNonMapPayload() { + void testResolveEffectiveDueDatePreservesRequestedDueDateForNonDarTaskPayload() { Long requested = 999L; assertEquals( requested, - CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, "plain-string", requested)); - assertEquals( - requested, - CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, List.of("a"), requested)); - assertEquals( - requested, CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, 42, requested)); + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, TaskEntityType.GlossaryApproval, "plain-string", requested)); + } + + @Test + void testResolveEffectiveDueDateThrowsForInvalidDarPayload() { + assertThrows( + IllegalArgumentException.class, + () -> + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, "plain-string", 999L)); } @Test @@ -260,7 +276,10 @@ void testResolveEffectiveDueDatePreservesRequestedDueDateWhenDurationKeyAbsent() assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("accessType", "FullAccess"), requested)); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("accessType", "FullAccess"), + requested)); } @Test @@ -269,7 +288,10 @@ void testResolveEffectiveDueDatePreservesRequestedDueDateWhenDurationIsNonString assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", 14), requested)); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", 14), + requested)); } @Test @@ -278,7 +300,10 @@ void testResolveEffectiveDueDatePreservesRequestedDueDateWhenDurationIsBlank() { assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", " "), requested)); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", " "), + requested)); } @Test @@ -286,7 +311,26 @@ void testResolveEffectiveDueDateComputesDayDuration() { long before = System.currentTimeMillis(); Long result = CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", "P14D"), 0L); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"), + 0L); + long after = System.currentTimeMillis(); + + long fourteenDays = 14L * 24 * 60 * 60 * 1000; + assertTrue(result >= before + fourteenDays); + assertTrue(result <= after + fourteenDays); + } + + @Test + void testResolveEffectiveDueDateParsesJsonStringPayload() { + long before = System.currentTimeMillis(); + Long result = + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + "{\"accessType\":\"FullAccess\",\"duration\":\"P14D\",\"reason\":\"audit\"}", + 0L); long after = System.currentTimeMillis(); long fourteenDays = 14L * 24 * 60 * 60 * 1000; @@ -299,7 +343,10 @@ void testResolveEffectiveDueDateComputesHourDuration() { long before = System.currentTimeMillis(); Long result = CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", "PT2H"), 0L); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "PT2H"), + 0L); long after = System.currentTimeMillis(); long twoHours = 2L * 60 * 60 * 1000; @@ -311,7 +358,11 @@ void testResolveEffectiveDueDateComputesHourDuration() { void testResolveEffectiveDueDateComputesMonthDuration() { long before = System.currentTimeMillis(); Long result = - CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, Map.of("duration", "P1M"), 0L); + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P1M"), + 0L); long after = System.currentTimeMillis(); long expectedMin = @@ -331,7 +382,11 @@ void testResolveEffectiveDueDateComputesMonthDuration() { void testResolveEffectiveDueDateComputesYearDuration() { long before = System.currentTimeMillis(); Long result = - CreateTask.resolveEffectiveDueDate(TaskEntityStatus.Granted, Map.of("duration", "P1Y"), 0L); + CreateTask.resolveEffectiveDueDate( + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P1Y"), + 0L); long after = System.currentTimeMillis(); long expectedMin = @@ -352,7 +407,10 @@ void testResolveEffectiveDueDateComputesCombinedPeriod() { long before = System.currentTimeMillis(); Long result = CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", "P2Y3M"), 0L); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P2Y3M"), + 0L); long after = System.currentTimeMillis(); long expectedMin = @@ -374,7 +432,10 @@ void testResolveEffectiveDueDateFallsBackForUnparseableDuration() { assertEquals( requested, CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", "not-a-duration"), requested)); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "not-a-duration"), + requested)); } // ---- resolveEffectiveExpirationDate ---- @@ -383,49 +444,68 @@ void testResolveEffectiveDueDateFallsBackForUnparseableDuration() { void testResolveEffectiveExpirationDateReturnsNullForNonGrantedStatus() { assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Approved, Map.of("duration", "P14D"))); + TaskEntityStatus.Approved, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"))); assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Open, Map.of("duration", "P14D"))); + TaskEntityStatus.Open, TaskEntityType.DataAccessRequest, Map.of("duration", "P14D"))); assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Revoked, Map.of("duration", "P14D"))); + TaskEntityStatus.Revoked, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"))); } @Test void testResolveEffectiveExpirationDateReturnsNullForNullPayload() { - assertNull(CreateTask.resolveEffectiveExpirationDate(TaskEntityStatus.Granted, null)); + assertNull( + CreateTask.resolveEffectiveExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, null)); + } + + @Test + void testResolveEffectiveExpirationDateReturnsNullForNonDarTaskPayload() { + assertNull( + CreateTask.resolveEffectiveExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.GlossaryApproval, "plain")); } @Test - void testResolveEffectiveExpirationDateReturnsNullForNonMapPayload() { - assertNull(CreateTask.resolveEffectiveExpirationDate(TaskEntityStatus.Granted, "plain")); - assertNull(CreateTask.resolveEffectiveExpirationDate(TaskEntityStatus.Granted, 42)); - assertNull(CreateTask.resolveEffectiveExpirationDate(TaskEntityStatus.Granted, List.of("a"))); + void testResolveEffectiveExpirationDateThrowsForInvalidDarPayload() { + assertThrows( + IllegalArgumentException.class, + () -> + CreateTask.resolveEffectiveExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, "plain")); } @Test void testResolveEffectiveExpirationDateReturnsNullWhenDurationMissingOrBlank() { assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("accessType", "FullAccess"))); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("accessType", "FullAccess"))); assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", " "))); + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, Map.of("duration", " "))); } @Test void testResolveEffectiveExpirationDateReturnsNullForNonStringDuration() { assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", 14))); + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, Map.of("duration", 14))); } @Test void testResolveEffectiveExpirationDateReturnsNullForUnparseableDuration() { assertNull( CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", "not-a-duration"))); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "not-a-duration"))); } @Test @@ -433,7 +513,7 @@ void testResolveEffectiveExpirationDateComputesDayDuration() { long before = System.currentTimeMillis(); Long result = CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", "P14D")); + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, Map.of("duration", "P14D")); long after = System.currentTimeMillis(); assertNotNull(result); @@ -447,7 +527,7 @@ void testResolveEffectiveExpirationDateComputesHourDuration() { long before = System.currentTimeMillis(); Long result = CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", "PT2H")); + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, Map.of("duration", "PT2H")); long after = System.currentTimeMillis(); assertNotNull(result); @@ -463,7 +543,9 @@ void testResolveEffectiveExpirationDateIsIdempotentWhenPayloadAlreadyHasIt() { Long existing = 1700000000000L; Long result = CreateTask.resolveEffectiveExpirationDate( - TaskEntityStatus.Granted, Map.of("duration", "P14D", "expirationDate", existing)); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D", "expirationDate", existing)); assertEquals(existing, result); } @@ -473,66 +555,123 @@ void testResolveEffectiveExpirationDateIsIdempotentWhenPayloadAlreadyHasIt() { void testWithGrantExpirationDateMergesIntoPayloadOnGranted() { Map payload = Map.of("accessType", "Masked", "duration", "P14D", "reason", "audit"); - Object result = CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, payload); - - assertTrue(result instanceof Map); - Map mergedMap = (Map) result; - assertEquals("Masked", mergedMap.get("accessType")); - assertEquals("P14D", mergedMap.get("duration")); - assertEquals("audit", mergedMap.get("reason")); - Object expiration = mergedMap.get("expirationDate"); + Object result = + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, payload); + + assertTrue(result instanceof DataAccessRequestPayload); + DataAccessRequestPayload mergedPayload = (DataAccessRequestPayload) result; + assertEquals("Masked", mergedPayload.getAccessType().value()); + assertEquals("P14D", mergedPayload.getDuration()); + assertEquals("audit", mergedPayload.getReason()); + Object expiration = mergedPayload.getExpirationDate(); assertNotNull(expiration); assertTrue(expiration instanceof Long); long expirationMillis = (Long) expiration; assertTrue(expirationMillis > System.currentTimeMillis()); } + @Test + void testWithGrantExpirationDateParsesJsonStringPayload() { + Object result = + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + "{\"accessType\":\"FullAccess\",\"duration\":\"P14D\",\"reason\":\"audit\"}"); + + assertTrue(result instanceof DataAccessRequestPayload); + DataAccessRequestPayload mergedPayload = (DataAccessRequestPayload) result; + assertEquals("FullAccess", mergedPayload.getAccessType().value()); + assertEquals("P14D", mergedPayload.getDuration()); + assertNotNull(mergedPayload.getExpirationDate()); + } + @Test void testWithGrantExpirationDateReturnsSameRefWhenNotGranted() { Map payload = Map.of("duration", "P14D"); - assertSame(payload, CreateTask.withGrantExpirationDate(TaskEntityStatus.Approved, payload)); + assertSame( + payload, + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Approved, TaskEntityType.DataAccessRequest, payload)); } @Test void testWithGrantExpirationDateReturnsSameRefWhenNoDuration() { Map payload = Map.of("accessType", "FullAccess"); - assertSame(payload, CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, payload)); + assertSame( + payload, + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, payload)); } @Test void testWithGrantExpirationDateReturnsSameRefWhenPayloadAlreadyHasExpirationDate() { Map payload = Map.of("duration", "P14D", "expirationDate", 1700000000000L); - assertSame(payload, CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, payload)); + assertSame( + payload, + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, payload)); + } + + @Test + void testWithGrantExpirationDateReturnsSameRefForNonDarTaskPayload() { + assertSame( + "string", + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.GlossaryApproval, "string")); + assertNull( + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, null)); } @Test - void testWithGrantExpirationDateReturnsSameRefWhenPayloadNotMap() { - assertSame("string", CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, "string")); - assertNull(CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, null)); + void testWithGrantExpirationDateThrowsForInvalidDarPayload() { + assertThrows( + IllegalArgumentException.class, + () -> + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, "string")); } @Test void testWithGrantExpirationDateDoesNotMutateOriginalPayload() { // The payload comes in from the workflow runtime; mutating it could leak state into // subsequent listeners. Confirm we produce a fresh map and leave the input alone. - Map payload = new java.util.HashMap<>(); - payload.put("duration", "P14D"); - Object result = CreateTask.withGrantExpirationDate(TaskEntityStatus.Granted, payload); + Map payload = Map.of("duration", "P14D"); + Object result = + CreateTask.withGrantExpirationDate( + TaskEntityStatus.Granted, TaskEntityType.DataAccessRequest, payload); assertFalse(payload.containsKey("expirationDate")); - assertTrue(result instanceof Map); - assertTrue(((Map) result).containsKey("expirationDate")); + assertTrue(result instanceof DataAccessRequestPayload); + assertNotNull(((DataAccessRequestPayload) result).getExpirationDate()); } @Test void testResolveEffectiveDueDateWithNullRequestedDueDateAndValidDurationReturnsComputedValue() { Long result = CreateTask.resolveEffectiveDueDate( - TaskEntityStatus.Granted, Map.of("duration", "P14D"), null); + TaskEntityStatus.Granted, + TaskEntityType.DataAccessRequest, + Map.of("duration", "P14D"), + null); assertNotNull(result); assertTrue(result > System.currentTimeMillis()); } + @Test + void testExtractPayloadCreatedByIgnoresNonRecognizerTaskPayload() { + assertNull(CreateTask.extractPayloadCreatedBy("plain", TaskEntityType.GlossaryApproval)); + } + + @Test + void testExtractPayloadCreatedByThrowsForInvalidRecognizerPayload() { + assertThrows( + IllegalArgumentException.class, + () -> + CreateTask.extractPayloadCreatedBy("plain", TaskEntityType.RecognizerFeedbackApproval)); + } + // ---- parseMillisFromIso8601Duration ---- @Test