Feat(Workflows): Auto-inject inclusive gateways for batch routing

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/Workflow.java

@@ -16,6 +16,8 @@ public class Workflow {
   public static final String ENTITY_LIST_VARIABLE = "entityList";
   public static final String TRUE_ENTITY_LIST_VARIABLE = "true_entityList";
   public static final String FALSE_ENTITY_LIST_VARIABLE = "false_entityList";
+  public static final String HAS_TRUE_ENTITIES_VARIABLE = "hasTrueEntities";
+  public static final String HAS_FALSE_ENTITIES_VARIABLE = "hasFalseEntities";
   public static final String BATCH_SINK_PROCESSED_VARIABLE = "batchSinkProcessed";
   public static final String TRIGGERING_OBJECT_ID_VARIABLE = "triggeringObjectId";
   public static final String RECOGNIZER_FEEDBACK = "recognizerFeedback";

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/NodeInterface.java

@@ -5,6 +5,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
 import org.flowable.bpmn.model.Activity;
 import org.flowable.bpmn.model.BoundaryEvent;
 import org.flowable.bpmn.model.BpmnModel;
@@ -22,6 +23,16 @@
 public interface NodeInterface {
   void addToWorkflow(BpmnModel model, Process process);
 
+  /**
+   * Returns the output ports this node can produce. Nodes that split a batch into multiple
+   * entity lists (e.g. CheckEntityAttributes → "true"/"false") declare their ports here.
+   * The MainWorkflow uses this to automatically inject inclusive gateways at split/join points.
+   * The default (empty set) means no inclusive gateway is injected for this node.
+   */
+  default Set<String> getOutputPorts() {
+    return Set.of();
+  }
+
   default BoundaryEvent getRuntimeExceptionBoundaryEvent() {
     return null;
   }

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/CheckChangeDescriptionTask.java

@@ -6,6 +6,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.bpmn.model.BoundaryEvent;
 import org.flowable.bpmn.model.BpmnModel;
@@ -84,6 +85,11 @@ public CheckChangeDescriptionTask(
     this.subProcess = subProcess;
   }
 
+  @Override
+  public Set<String> getOutputPorts() {
+    return Set.of("true", "false");
+  }
+
   @Override
   public BoundaryEvent getRuntimeExceptionBoundaryEvent() {
     return runtimeExceptionBoundaryEvent;

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/CheckEntityAttributesTask.java

@@ -6,6 +6,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.bpmn.model.BoundaryEvent;
 import org.flowable.bpmn.model.BpmnModel;
@@ -77,6 +78,11 @@ public CheckEntityAttributesTask(
     this.subProcess = subProcess;
   }
 
+  @Override
+  public Set<String> getOutputPorts() {
+    return Set.of("true", "false");
+  }
+
   @Override
   public BoundaryEvent getRuntimeExceptionBoundaryEvent() {
     return runtimeExceptionBoundaryEvent;

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/DataCompletenessTask.java

@@ -7,6 +7,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.bpmn.model.BoundaryEvent;
 import org.flowable.bpmn.model.BpmnModel;
@@ -19,6 +22,7 @@
 import org.flowable.bpmn.model.SubProcess;
 import org.openmetadata.schema.governance.workflows.WorkflowConfiguration;
 import org.openmetadata.schema.governance.workflows.elements.nodes.automatedTask.DataCompletenessTaskDefinition;
+import org.openmetadata.schema.governance.workflows.elements.nodes.automatedTask.QualityBand;
 import org.openmetadata.schema.utils.JsonUtils;
 import org.openmetadata.service.governance.workflows.elements.NodeInterface;
 import org.openmetadata.service.governance.workflows.elements.nodes.automatedTask.impl.DataCompletenessImpl;
@@ -32,9 +36,17 @@
 public class DataCompletenessTask implements NodeInterface {
   private final SubProcess subProcess;
   private final BoundaryEvent runtimeExceptionBoundaryEvent;
+  private final Set<String> outputPorts;
 
   public DataCompletenessTask(
       DataCompletenessTaskDefinition nodeDefinition, WorkflowConfiguration workflowConfig) {
+    this.outputPorts =
+        Optional.ofNullable(nodeDefinition.getConfig())
+            .map(c -> c.getQualityBands())
+            .orElse(List.of())
+            .stream()
+            .map(QualityBand::getName)
+            .collect(Collectors.toSet());
     String subProcessId = nodeDefinition.getName();
 
     SubProcess subProcess = new SubProcessBuilder().id(subProcessId).build();
@@ -107,6 +119,11 @@ private ServiceTask getDataCompletenessServiceTask(
     return builder.build();
   }
 
+  @Override
+  public Set<String> getOutputPorts() {
+    return outputPorts;
+  }
+
   @Override
   public void addToWorkflow(BpmnModel model, Process process) {
     process.addFlowElement(subProcess);

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckChangeDescriptionTaskImpl.java

@@ -2,7 +2,8 @@
 
 import static org.openmetadata.service.governance.workflows.Workflow.EXCEPTION_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.FALSE_ENTITY_LIST_VARIABLE;
-import static org.openmetadata.service.governance.workflows.Workflow.RESULT_VARIABLE;
+import static org.openmetadata.service.governance.workflows.Workflow.HAS_FALSE_ENTITIES_VARIABLE;
+import static org.openmetadata.service.governance.workflows.Workflow.HAS_TRUE_ENTITIES_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.TRUE_ENTITY_LIST_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION;
 import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId;
@@ -77,10 +78,11 @@ public void execute(DelegateExecution execution) {
         }
       }
 
-      boolean result = !trueEntityList.isEmpty();
       varHandler.setNodeVariable(TRUE_ENTITY_LIST_VARIABLE, trueEntityList);
       varHandler.setNodeVariable(FALSE_ENTITY_LIST_VARIABLE, falseEntityList);
-      varHandler.setNodeVariable(RESULT_VARIABLE, result);
+      varHandler.setNodeVariable(HAS_TRUE_ENTITIES_VARIABLE, !trueEntityList.isEmpty());
+      varHandler.setNodeVariable(
+          HAS_FALSE_ENTITIES_VARIABLE, !falseEntityList.isEmpty() || entityList.isEmpty());
     } catch (Exception exc) {
       LOG.error(
           "[{}] Failure: ", getProcessDefinitionKeyFromId(execution.getProcessDefinitionId()), exc);

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/CheckEntityAttributesImpl.java

@@ -2,7 +2,8 @@
 
 import static org.openmetadata.service.governance.workflows.Workflow.EXCEPTION_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.FALSE_ENTITY_LIST_VARIABLE;
-import static org.openmetadata.service.governance.workflows.Workflow.RESULT_VARIABLE;
+import static org.openmetadata.service.governance.workflows.Workflow.HAS_FALSE_ENTITIES_VARIABLE;
+import static org.openmetadata.service.governance.workflows.Workflow.HAS_TRUE_ENTITIES_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.TRUE_ENTITY_LIST_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION;
 import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId;
@@ -69,10 +70,11 @@ public void execute(DelegateExecution execution) {
         }
       }
 
-      boolean result = !trueEntityList.isEmpty();
       varHandler.setNodeVariable(TRUE_ENTITY_LIST_VARIABLE, trueEntityList);
       varHandler.setNodeVariable(FALSE_ENTITY_LIST_VARIABLE, falseEntityList);
-      varHandler.setNodeVariable(RESULT_VARIABLE, result);
+      varHandler.setNodeVariable(HAS_TRUE_ENTITIES_VARIABLE, !trueEntityList.isEmpty());
+      varHandler.setNodeVariable(
+          HAS_FALSE_ENTITIES_VARIABLE, !falseEntityList.isEmpty() || entityList.isEmpty());
     } catch (Exception exc) {
       LOG.error(
           "[{}] Failure: ", getProcessDefinitionKeyFromId(execution.getProcessDefinitionId()), exc);

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/automatedTask/impl/DataCompletenessImpl.java

@@ -2,7 +2,6 @@
 
 import static org.openmetadata.service.governance.workflows.Workflow.ENTITY_LIST_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.EXCEPTION_VARIABLE;
-import static org.openmetadata.service.governance.workflows.Workflow.RESULT_VARIABLE;
 import static org.openmetadata.service.governance.workflows.Workflow.WORKFLOW_RUNTIME_EXCEPTION;
 import static org.openmetadata.service.governance.workflows.WorkflowHandler.getProcessDefinitionKeyFromId;
 
@@ -121,10 +120,12 @@ public void execute(DelegateExecution execution) {
             entityList.size());
       }
 
-      // Per-band entity lists — ALL bands stored, empty or not (inclusive gateway ready)
+      // Per-band entity lists — ALL bands stored, empty or not.
+      // When all flags are false the split gateway falls through to its defaultFlow.
       for (QualityBand band : qualityBands) {
         List<String> bandEntities = entitiesByBand.getOrDefault(band.getName(), List.of());
         varHandler.setNodeVariable(band.getName() + "_" + ENTITY_LIST_VARIABLE, bandEntities);
+        varHandler.setNodeVariable(bandFlagVariable(band.getName()), !bandEntities.isEmpty());
       }
 
       // Priority band = highest minimumScore band that has entities
@@ -142,7 +143,6 @@ public void execute(DelegateExecution execution) {
           priorityBand,
           entitiesByBand.keySet());
 
-      varHandler.setNodeVariable(RESULT_VARIABLE, priorityBand);
       varHandler.setNodeVariable("entityResults", entityResults);
 
       // Scalar outputs for backward compat when processing a single entity
@@ -165,6 +165,10 @@ public void execute(DelegateExecution execution) {
     }
   }
 
+  public static String bandFlagVariable(String bandName) {
+    return "has_" + bandName + "_entities";
+  }
+
   private void storeFieldList(
       WorkflowVariableHandler varHandler, String varName, List<String> fields) {
     if (fields.size() <= 50) {