feat(engine): Support Exclusive Jobs across Process Hierarchies

Context: Exclusive jobs that originate from process hierarchies (processes which contain multi-instance subprocesses) can now be executed exclusively.

Why: The exclusive execution would only be applied on tasks that originate at a root level. Any subprocess spawned by root processes would not be correlated and considered for exclusive execution.

Changes: See below the changes of this feature

- New column `ROOT_PROC_INST_ID_ ` is introduced to correlated a process instance with its root parent
    - The column is added to all supported databases & `7.20_to_7.21` migration scripts.
- The query `selectNextJobsToExecute` is enriched to consider the root process instance id when the feature is enabled.
- The feature flag `jobExecutorAcquireExclusiveOverProcessHierarchies` enables / disables the feature.
- The feature is disabled by default for backwards compatibility with the legacy behaviour.

Tests:

- Unit tests under `ExclusiveJobAcquisitionTest` which cover:
    - Legacy behaviour with the feature disabled
    - Feature behaviour with the feature enabled & a 1-level deep process hierarchy (process which spawns a subprocess)
    - Feature behaviour with the feature enabled & a 2-level deep process hierarchy (process which spawns a subprocess - which spawns another subprocess)

- Migration tests under `ExclusiveOverProcessHierarchiesTest` which cover:
    - How the job acquisition and execution (see `AcquireJobsCmd`) behave with existing 7.20 process instances when the feature is disabled (legacy behaviour and backwards compatibility)
    - How the job acquisition and execution (see `AcquireJobsCmd`) behave with existing 7.20 process instances when the feature is enabled (feature behaviour)
    - The tests need to select all the jobs and lock them to perform their logic. After the test execution the jobs are unlocked.

Other Notable Changes: 

- Fixed flakiness of `DecisionDefinitionTest`
    - The Test class was susceptible to timezone changes due its test data using the current date; as a result, depending on the date of execution, changing to DST could fail the tests.
    - This behaviour is fixed by adjusting the test data to use a fixed past date, rendering the test executions immune to date of execution.
- The waiting behaviour on jobs for the tests to rely on the scheduler, shared by `SequentialJobAcquisitionTest`, `ExclusiveJobAcquisitionTest` is extracted into class `JobExecutorWaitUtils`

Co-authored-by: daniel.kelemen
Co-authored-by: yanavasileva
Co-authored-by: petros.savvidis

Related-to: #4004 , #4003

engine/src/main/java/org/camunda/bpm/engine/impl/JobQueryImpl.java

@@ -48,6 +48,7 @@ public class JobQueryImpl extends AbstractQuery<JobQuery, Job> implements JobQue
   protected String id;
   protected Set<String> ids;
   protected String jobDefinitionId;
+  protected String rootProcessInstanceId;
   protected String processInstanceId;
   protected Set<String> processInstanceIds;
   protected String executionId;
@@ -100,6 +101,12 @@ public JobQuery jobDefinitionId(String jobDefinitionId) {
     return this;
   }
 
+  public JobQueryImpl rootProcessInstanceId(String rootProcessInstanceId) {
+    ensureNotNull("Provided root process instance id", rootProcessInstanceId);
+    this.rootProcessInstanceId = rootProcessInstanceId;
+    return this;
+  }
+
   public JobQueryImpl processInstanceId(String processInstanceId) {
     ensureNotNull("Provided process instance id", processInstanceId);
     this.processInstanceId = processInstanceId;
@@ -368,6 +375,9 @@ public List<ImmutablePair<String, String>> executeDeploymentIdMappingsList(Comma
   public Set<String> getIds() {
     return ids;
   }
+  public String getRootProcessInstanceId() {
+    return rootProcessInstanceId;
+  }
   public String getProcessInstanceId() {
     return processInstanceId;
   }

engine/src/main/java/org/camunda/bpm/engine/impl/cfg/ProcessEngineConfigurationImpl.java

@@ -516,6 +516,25 @@ public abstract class ProcessEngineConfigurationImpl extends ProcessEngineConfig
   protected long jobExecutorPriorityRangeMin = Long.MIN_VALUE;
   protected long jobExecutorPriorityRangeMax = Long.MAX_VALUE;
 
+  /**
+   * When set to false, exclusivity (no parallel execution) of tasks is applied per process instance.
+   * When set to true, exclusivity (no parallel execution) of tasks is extended across all hierarchies of each given
+   * process instance.
+   * <p>
+   * The feature is targeting processes which might spawn a hierarchy of subprocesses e.g. via a call activity.
+   * The legacy behaviour does not guarantee sequential execution of any spawned subprocesses. Instead, it works only
+   * at the root level. When the feature is enabled, any spawned subprocess of the root process will be acquired and
+   * executed sequentially by one thread.
+   * <p>
+   * Note that the above configuration might introduce performance implications in complex process modelling that involves
+   * high multi-instance multiplicity and numerous subprocesses.
+   * <p>
+   * Use the feature in combination with awareness of your process modeling.
+   * <p>
+   * Default value: false; to keep the legacy behaviour backwards compatible.
+   */
+  protected boolean jobExecutorAcquireExclusiveOverProcessHierarchies = false;
+
   // EXTERNAL TASK /////////////////////////////////////////////////////////////
   protected PriorityProvider<ExternalTaskActivityBehavior> externalTaskPriorityProvider;
 
@@ -5004,6 +5023,15 @@ public ProcessEngineConfigurationImpl setEnforceHistoryTimeToLive(boolean enforc
     return this;
   }
 
+  public ProcessEngineConfigurationImpl setJobExecutorAcquireExclusiveOverProcessHierarchies(boolean jobExecutorAcquireExclusiveOverProcessHierarchies) {
+    this.jobExecutorAcquireExclusiveOverProcessHierarchies = jobExecutorAcquireExclusiveOverProcessHierarchies;
+    return this;
+  }
+
+  public boolean isJobExecutorAcquireExclusiveOverProcessHierarchies() {
+    return this.jobExecutorAcquireExclusiveOverProcessHierarchies;
+  }
+
   public String getBatchOperationHistoryTimeToLive() {
     return batchOperationHistoryTimeToLive;
   }

engine/src/main/java/org/camunda/bpm/engine/impl/cmd/AcquireJobsCmd.java

@@ -22,7 +22,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-
 import org.camunda.bpm.engine.impl.Page;
 import org.camunda.bpm.engine.impl.db.DbEntity;
 import org.camunda.bpm.engine.impl.db.entitymanager.OptimisticLockingListener;
@@ -65,18 +64,17 @@ public AcquiredJobs execute(CommandContext commandContext) {
       .getJobManager()
       .findNextJobsToExecute(new Page(0, numJobsToAcquire));
 
-    Map<String, List<String>> exclusiveJobsByProcessInstance = new HashMap<String, List<String>>();
+    Map<String, List<String>> exclusiveJobsByProcessInstance = new HashMap<>();
+
+    boolean isAcquireExclusiveOverProcessHierarchies = isAcquireExclusiveOverProcessHierarchies(commandContext);
 
     for (AcquirableJobEntity job : jobs) {
 
       lockJob(job);
 
       if(job.isExclusive()) {
-        List<String> list = exclusiveJobsByProcessInstance.get(job.getProcessInstanceId());
-        if (list == null) {
-          list = new ArrayList<String>();
-          exclusiveJobsByProcessInstance.put(job.getProcessInstanceId(), list);
-        }
+        String processInstanceId = selectProcessInstanceId(job, isAcquireExclusiveOverProcessHierarchies);
+        List<String> list = exclusiveJobsByProcessInstance.computeIfAbsent(processInstanceId, key -> new ArrayList<>());
         list.add(job.getId());
       }
       else {
@@ -140,7 +138,7 @@ public OptimisticLockingResult failedOperation(DbOperation operation) {
     if (operation instanceof DbEntityOperation) {
 
       DbEntityOperation entityOperation = (DbEntityOperation) operation;
-      
+
       // could not lock the job -> remove it from list of acquired jobs
       acquiredJobs.removeJobId(entityOperation.getEntity().getId());
 
@@ -154,4 +152,19 @@ public OptimisticLockingResult failedOperation(DbOperation operation) {
     return OptimisticLockingResult.THROW;
   }
 
+  protected boolean isAcquireExclusiveOverProcessHierarchies(CommandContext context) {
+    var engineConfig = context.getProcessEngineConfiguration();
+
+    return engineConfig != null && engineConfig.isJobExecutorAcquireExclusiveOverProcessHierarchies();
+  }
+
+  protected String selectProcessInstanceId(AcquirableJobEntity job, boolean isAcquireExclusiveOverProcessHierarchies) {
+
+    if (isAcquireExclusiveOverProcessHierarchies && job.getRootProcessInstanceId() != null) {
+      return job.getRootProcessInstanceId();
+    }
+
+    return job.getProcessInstanceId();
+  }
+
 }

engine/src/main/java/org/camunda/bpm/engine/impl/db/sql/DbSqlSessionFactory.java

@@ -601,7 +601,7 @@ public class DbSqlSessionFactory implements SessionFactory {
     databaseSpecificLimitBetweenStatements.put(DB2, db2LimitBetweenWithoutColumns + "RES.* ");
     databaseSpecificLimitBetweenFilterStatements.put(DB2, db2LimitBetweenWithoutColumns + "RES.ID_, RES.REV_, RES.RESOURCE_TYPE_, RES.NAME_, RES.OWNER_ ");
     databaseSpecificLimitBetweenAcquisitionStatements.put(DB2, db2LimitBetweenWithoutColumns
-        + "RES.ID_, RES.REV_, RES.TYPE_, RES.LOCK_EXP_TIME_, RES.LOCK_OWNER_, RES.EXCLUSIVE_, RES.PROCESS_INSTANCE_ID_, RES.DUEDATE_, RES.PRIORITY_ ");
+        + "RES.ID_, RES.REV_, RES.TYPE_, RES.LOCK_EXP_TIME_, RES.LOCK_OWNER_, RES.EXCLUSIVE_, RES.ROOT_PROC_INST_ID_, RES.PROCESS_INSTANCE_ID_, RES.DUEDATE_, RES.PRIORITY_ ");
     databaseSpecificLimitBeforeInUpdate.put(DB2, "");
     databaseSpecificLimitAfterInUpdate.put(DB2, "");
     databaseSpecificLimitBeforeWithoutOffsetStatements.put(DB2, "");
@@ -699,7 +699,7 @@ public class DbSqlSessionFactory implements SessionFactory {
     databaseSpecificLimitBetweenStatements.put(MSSQL, mssqlLimitBetweenWithoutColumns + "RES.* ");
     databaseSpecificLimitBetweenFilterStatements.put(MSSQL, "");
     databaseSpecificLimitBetweenAcquisitionStatements.put(MSSQL, mssqlLimitBetweenWithoutColumns
-        + "RES.ID_, RES.REV_, RES.TYPE_, RES.LOCK_EXP_TIME_, RES.LOCK_OWNER_, RES.EXCLUSIVE_, RES.PROCESS_INSTANCE_ID_, RES.DUEDATE_, RES.PRIORITY_ ");
+        + "RES.ID_, RES.REV_, RES.TYPE_, RES.LOCK_EXP_TIME_, RES.LOCK_OWNER_, RES.EXCLUSIVE_, RES.ROOT_PROC_INST_ID_, RES.PROCESS_INSTANCE_ID_, RES.DUEDATE_, RES.PRIORITY_ ");
     databaseSpecificLimitBeforeInUpdate.put(MSSQL, "");
     databaseSpecificLimitAfterInUpdate.put(MSSQL, "");
     databaseSpecificLimitBeforeWithoutOffsetStatements.put(MSSQL, "TOP (#{maxResults})");

engine/src/main/java/org/camunda/bpm/engine/impl/persistence/entity/AcquirableJobEntity.java

@@ -34,6 +34,7 @@ public class AcquirableJobEntity implements DbEntity, HasDbRevision {
   protected Date lockExpirationTime = null;
   protected Date duedate;
 
+  protected String rootProcessInstanceId = null;
   protected String processInstanceId = null;
 
   protected boolean isExclusive = DEFAULT_EXCLUSIVE;
@@ -99,6 +100,14 @@ public void setLockExpirationTime(Date lockExpirationTime) {
     this.lockExpirationTime = lockExpirationTime;
   }
 
+  public String getRootProcessInstanceId() {
+    return rootProcessInstanceId;
+  }
+
+  public void setRootProcessInstanceId(String rootProcessInstanceId) {
+    this.rootProcessInstanceId = rootProcessInstanceId;
+  }
+
   public String getProcessInstanceId() {
     return processInstanceId;
   }
@@ -148,6 +157,7 @@ public String toString() {
         + ", lockOwner=" + lockOwner
         + ", lockExpirationTime=" + lockExpirationTime
         + ", duedate=" + duedate
+        + ", rootProcessInstanceId=" + rootProcessInstanceId
         + ", processInstanceId=" + processInstanceId
         + ", isExclusive=" + isExclusive
         + "]";

engine/src/main/java/org/camunda/bpm/engine/impl/persistence/entity/JobEntity.java

@@ -247,6 +247,7 @@ public void setExecution(ExecutionEntity execution) {
       this.execution = execution;
       executionId = execution.getId();
       processInstanceId = execution.getProcessInstanceId();
+      rootProcessInstanceId = execution.getRootProcessInstanceId();
       // if the execution is suspended, suspend the job entity as well to prevent unwanted job execution
       if(execution.isSuspended()) {
         suspensionState = execution.getSuspensionState();
@@ -257,6 +258,7 @@ public void setExecution(ExecutionEntity execution) {
       this.execution.removeJob(this);
       this.execution = execution;
       processInstanceId = null;
+      rootProcessInstanceId = null;
       executionId = null;
     }
   }
@@ -714,7 +716,6 @@ public String toString() {
            + ", executionId=" + executionId
            + ", processInstanceId=" + processInstanceId
            + ", isExclusive=" + isExclusive
-           + ", isExclusive=" + isExclusive
            + ", jobDefinitionId=" + jobDefinitionId
            + ", jobHandlerType=" + jobHandlerType
            + ", jobHandlerConfiguration=" + jobHandlerConfiguration

engine/src/main/java/org/camunda/bpm/engine/impl/persistence/entity/JobManager.java

@@ -204,9 +204,11 @@ public List<AcquirableJobEntity> findNextJobsToExecute(Page page) {
 
     Map<String,Object> params = new HashMap<>();
     Date now = ClockUtil.getCurrentTime();
+
     params.put("now", now);
     params.put("alwaysSetDueDate", isEnsureJobDueDateNotNull());
     params.put("deploymentAware", engineConfiguration.isJobExecutorDeploymentAware());
+
     if (engineConfiguration.isJobExecutorDeploymentAware()) {
       Set<String> registeredDeployments = engineConfiguration.getRegisteredDeployments();
       if (!registeredDeployments.isEmpty()) {
@@ -217,12 +219,14 @@ public List<AcquirableJobEntity> findNextJobsToExecute(Page page) {
     boolean jobExecutorAcquireByPriority = engineConfiguration.isJobExecutorAcquireByPriority();
     long jobExecutorPriorityRangeMin = engineConfiguration.getJobExecutorPriorityRangeMin();
     long jobExecutorPriorityRangeMax = engineConfiguration.getJobExecutorPriorityRangeMax();
+
     params.put("jobPriorityMin", jobExecutorAcquireByPriority && jobExecutorPriorityRangeMin != Long.MIN_VALUE ? jobExecutorPriorityRangeMin : null);
     params.put("jobPriorityMax", jobExecutorAcquireByPriority && jobExecutorPriorityRangeMax != Long.MAX_VALUE ? jobExecutorPriorityRangeMax : null);
 
     params.put("historyCleanupEnabled", engineConfiguration.isHistoryCleanupEnabled());
 
     List<QueryOrderingProperty> orderingProperties = new ArrayList<>();
+
     if (engineConfiguration.isJobExecutorAcquireByPriority()) {
       orderingProperties.add(JOB_PRIORITY_ORDERING_PROPERTY);
     }
@@ -236,6 +240,7 @@ public List<AcquirableJobEntity> findNextJobsToExecute(Page page) {
     params.put("orderingProperties", orderingProperties);
     // don't apply default sorting
     params.put("applyOrdering", !orderingProperties.isEmpty());
+    params.put("applyExclusiveOverProcessHierarchies", engineConfiguration.isJobExecutorAcquireExclusiveOverProcessHierarchies());
 
     return getDbEntityManager().selectList("selectNextJobsToExecute", params, page);
   }

engine/src/main/java/org/camunda/bpm/engine/runtime/Job.java

@@ -40,6 +40,11 @@ public interface Job {
    */
   Date getDuedate();
 
+  /**
+   * Returns the id of the root process instance which execution created the job.
+   */
+  String getRootProcessInstanceId();
+
   /**
    * Returns the id of the process instance which execution created the job.
    */

engine/src/main/java/org/camunda/bpm/engine/runtime/JobQuery.java

@@ -39,6 +39,9 @@ public interface JobQuery extends Query<JobQuery, Job> {
   /** Only select jobs which exist for the given job definition id. **/
   JobQuery jobDefinitionId(String jobDefinitionId);
 
+  /** Only select jobs which exist for the given root process instance. **/
+  JobQuery rootProcessInstanceId(String rootProcessInstanceId);
+
   /** Only select jobs which exist for the given process instance. **/
   JobQuery processInstanceId(String processInstanceId);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.cockroachdb.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ boolean,
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -384,6 +385,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_);
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_,HANDLER_CFG_);
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.db2.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ smallint check(EXCLUSIVE_ in (1,0)),
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -387,6 +388,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_);
 -- CAM-5914
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.h2.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ boolean,
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -385,6 +386,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_);
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_,HANDLER_CFG_);
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.mariadb.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ boolean,
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -386,6 +387,7 @@ create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 -- this index needs to be limited in mariadb see CAM-6938
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_(100),HANDLER_CFG_(155));
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.mssql.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ nvarchar(255),
     EXCLUSIVE_ bit,
     EXECUTION_ID_ nvarchar(64),
+    ROOT_PROC_INST_ID_ nvarchar(64),
     PROCESS_INSTANCE_ID_ nvarchar(64),
     PROCESS_DEF_ID_ nvarchar(64),
     PROCESS_DEF_KEY_ nvarchar(255),
@@ -385,6 +386,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_);
 -- CAM-5914
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 create unique index ACT_UNIQ_AUTH_USER on ACT_RU_AUTHORIZATION (TYPE_,USER_ID_,RESOURCE_TYPE_,RESOURCE_ID_) where USER_ID_ is not null;

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.mysql.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ boolean,
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -386,6 +387,7 @@ create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 -- this index needs to be limited in mysql see CAM-6938
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_(100),HANDLER_CFG_(155));
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.oracle.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ NVARCHAR2(255),
     EXCLUSIVE_ NUMBER(1,0) CHECK (EXCLUSIVE_ IN (1,0)),
     EXECUTION_ID_ NVARCHAR2(64),
+    ROOT_PROC_INST_ID_ NVARCHAR2(64),
     PROCESS_INSTANCE_ID_ NVARCHAR2(64),
     PROCESS_DEF_ID_ NVARCHAR2(64),
     PROCESS_DEF_KEY_ NVARCHAR2(255),
@@ -384,6 +385,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_, 0);
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_, SUBSTR(HANDLER_CFG_, 1, 1850));
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_, 0);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_, 0);
 

engine/src/main/resources/org/camunda/bpm/engine/db/create/activiti.postgres.create.engine.sql

@@ -112,6 +112,7 @@ create table ACT_RU_JOB (
     LOCK_OWNER_ varchar(255),
     EXCLUSIVE_ boolean,
     EXECUTION_ID_ varchar(64),
+    ROOT_PROC_INST_ID_ varchar(64),
     PROCESS_INSTANCE_ID_ varchar(64),
     PROCESS_DEF_ID_ varchar(64),
     PROCESS_DEF_KEY_ varchar(255),
@@ -384,6 +385,7 @@ create index ACT_IDX_INC_TENANT_ID on ACT_RU_INCIDENT(TENANT_ID_);
 create index ACT_IDX_JOB_EXECUTION_ID on ACT_RU_JOB(EXECUTION_ID_);
 create index ACT_IDX_JOB_HANDLER on ACT_RU_JOB(HANDLER_TYPE_,HANDLER_CFG_);
 create index ACT_IDX_JOB_PROCINST on ACT_RU_JOB(PROCESS_INSTANCE_ID_);
+create index ACT_IDX_JOB_ROOT_PROCINST on ACT_RU_JOB(ROOT_PROC_INST_ID_);
 create index ACT_IDX_JOB_TENANT_ID on ACT_RU_JOB(TENANT_ID_);
 create index ACT_IDX_JOBDEF_TENANT_ID on ACT_RU_JOBDEF(TENANT_ID_);
 

engine/src/main/resources/org/camunda/bpm/engine/db/drop/activiti.cockroachdb.drop.engine.sql

@@ -121,6 +121,7 @@ drop index ACT_IDX_VAR_PROCINST;
 drop index ACT_IDX_VAR_BYTEARRAY;
 drop index ACT_IDX_JOB_EXCEPTION;
 drop index ACT_IDX_JOB_PROCINST;
+drop index ACT_IDX_JOB_ROOT_PROCINST;
 drop index ACT_IDX_INC_CONFIGURATION;
 drop index ACT_IDX_AUTH_GROUP_ID;