feat(engine): Populating process instance id in jobs

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

@@ -54,6 +54,7 @@ protected void postProcessJob(RestartProcessInstancesBatchConfiguration configur
           .findDeployedProcessDefinitionById(configuration.getProcessDefinitionId());
       job.setDeploymentId(processDefinitionEntity.getDeploymentId());
     }
+    job.setProcessDefinitionId(configuration.getProcessDefinitionId());
   }
 
   @Override

engine/src/main/java/org/camunda/bpm/engine/impl/batch/AbstractBatchJobHandler.java

@@ -169,6 +169,11 @@ protected void createJobEntities(BatchEntity batch, T configuration, String depl
       ByteArrayEntity configurationEntity = saveConfiguration(byteArrayManager, jobConfiguration);
 
       JobEntity job = createBatchJob(batch, configurationEntity);
+
+      if (jobConfiguration.getIds() != null && jobConfiguration.getIds().size() == 1) {
+        job.setProcessInstanceId(jobConfiguration.getIds().get(0));
+      }
+
       job.setDeploymentId(deploymentId);
       postProcessJob(configuration, job, jobConfiguration);
       jobManager.insertAndHintJobExecutor(job);

engine/src/main/java/org/camunda/bpm/engine/impl/batch/message/MessageCorrelationBatchJobHandler.java

@@ -66,10 +66,7 @@ protected MessageCorrelationBatchConfiguration createJobConfiguration(MessageCor
 
   @Override
   protected void postProcessJob(MessageCorrelationBatchConfiguration configuration, JobEntity job, MessageCorrelationBatchConfiguration jobConfiguration) {
-    // if there is only one process instance to adjust, set its ID to the job so exclusive scheduling is possible
-    if (jobConfiguration.getIds() != null && jobConfiguration.getIds().size() == 1) {
-      job.setProcessInstanceId(jobConfiguration.getIds().get(0));
-    }
+
   }
 
   @Override

engine/src/main/java/org/camunda/bpm/engine/impl/batch/variables/BatchSetVariablesHandler.java

@@ -78,10 +78,7 @@ public String getType() {
 
   @Override
   protected void postProcessJob(BatchConfiguration configuration, JobEntity job, BatchConfiguration jobConfiguration) {
-    // if there is only one process instance to adjust, set its ID to the job so exclusive scheduling is possible
-    if (jobConfiguration.getIds() != null && jobConfiguration.getIds().size() == 1) {
-      job.setProcessInstanceId(jobConfiguration.getIds().get(0));
-    }
+
   }
 
   protected ByteArrayEntity findByteArrayById(String byteArrayId, CommandContext commandContext) {

engine/src/test/java/org/camunda/bpm/engine/test/api/history/BatchHistoricDecisionInstanceDeletionTest.java

@@ -16,6 +16,7 @@
  */
 package org.camunda.bpm.engine.test.api.history;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -147,6 +148,19 @@ public void removeBatches() {
     ClockUtil.reset();
   }
 
+  @Test
+  public void shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations(){
+    // when
+    Batch batch = historyService.deleteHistoricDecisionInstancesAsync(decisionInstanceIds, null);
+    helper.executeSeedJob(batch);
+
+    //then
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(helper.getExecutionJobs(batch))
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(decisionInstanceIds.toArray());
+  }
+
   @Test
   public void createBatchDeletionByIds() {
     // when

engine/src/test/java/org/camunda/bpm/engine/test/api/history/HistoryServiceAsyncOperationsTest.java

@@ -16,6 +16,8 @@
  */
 package org.camunda.bpm.engine.test.api.history;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.CoreMatchers.anyOf;
 import static org.hamcrest.CoreMatchers.not;
@@ -109,6 +111,21 @@ public void testDeleteHistoryProcessInstancesAsyncWithList() throws Exception {
     assertAllHistoricProcessInstancesAreDeleted();
   }
 
+  @Test
+  public void testDeleteHistoryProcessInstances_shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
+    //when
+    Batch batch = historyService.deleteHistoricProcessInstancesAsync(historicProcessInstances, TEST_REASON);
+
+    completeSeedJobs(batch);
+    List<Job> executionJobs = managementService.createJobQuery().jobDefinitionId(batch.getBatchJobDefinitionId()).list();
+
+    // then
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(historicProcessInstances.toArray());
+  }
+
   @Test
   public void testDeleteHistoryProcessInstancesAsyncWithListForDeletedDeployment() throws Exception {
     // given a second deployment

engine/src/test/java/org/camunda/bpm/engine/test/api/history/removaltime/batch/BatchSetRemovalTimeTest.java

@@ -420,6 +420,34 @@ public void shouldSetRemovalTimeForBatch_MultipleInvocationsPerBatchJob() {
     assertThat(historicBatches.get(1).getRemovalTime()).isEqualTo(REMOVAL_TIME);
   }
 
+  @Test
+  public void testRemovalTimeProcess_shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
+    // given
+    testRule.getProcessEngineConfiguration().setInvocationsPerBatchJob(1);
+
+    String processInstanceIdOne = testRule.process().userTask().deploy().start();
+    String processInstanceIdTwo = testRule.process().userTask().deploy().start();
+
+    // when
+    HistoricProcessInstanceQuery query = historyService.createHistoricProcessInstanceQuery();
+    Batch batch = historyService.setRemovalTimeToHistoricProcessInstances()
+            .absoluteRemovalTime(REMOVAL_TIME)
+            .byQuery(query)
+            .executeAsync();
+
+    testRule.executeSeedJobs(batch);
+
+    // then
+    //Making sure that processInstanceId is set in execution jobs #4205
+    List<Job> executionJobs = testRule.getExecutionJobs(batch);
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processInstanceIdOne, processInstanceIdTwo);
+
+    // clear
+    managementService.deleteBatch(batch.getId(), true);
+  }
+
   @Test
   public void shouldSetRemovalTime_SingleInvocationPerBatchJob() {
     // given
@@ -2835,6 +2863,9 @@ public void shouldSetExecutionStartTimeInBatchAndHistoryForDecisions() {
                 .putValue("temperature", 32)
                 .putValue("dayType", "Weekend")
         ).evaluate();
+
+    HistoricDecisionInstanceQuery query = historyService.createHistoricDecisionInstanceQuery();
+
     Batch batch = historyService.setRemovalTimeToHistoricDecisionInstances()
         .absoluteRemovalTime(CURRENT_DATE)
         .byQuery(historyService.createHistoricDecisionInstanceQuery())
@@ -2851,6 +2882,11 @@ public void shouldSetExecutionStartTimeInBatchAndHistoryForDecisions() {
 
     assertThat(batch.getExecutionStartTime()).isEqualToIgnoringMillis(CURRENT_DATE);
     assertThat(historicBatch.getExecutionStartTime()).isEqualToIgnoringMillis(CURRENT_DATE);
+
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(query.list().stream().map(HistoricDecisionInstance::getId).toArray());
   }
 
 }

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/CorrelateAllMessageBatchTest.java

@@ -580,6 +580,7 @@ public void shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
     rule.executeSeedJobs(batch);
 
     // then
+    //Making sure that processInstanceId is set in execution jobs #4205
     List<Job> executionJobs = rule.getExecutionJobs(batch);
     assertThat(executionJobs)
       .extracting("processInstanceId")

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/ModificationExecutionAsyncTest.java

@@ -167,6 +167,13 @@ public void createBatchModification() {
     Batch batch = runtimeService.createModification(processDefinition.getId()).startAfterActivity("user2").processInstanceIds(processInstanceIds).executeAsync();
 
     assertBatchCreated(batch, 2);
+
+    //Making sure that processInstanceId is set in execution jobs #4205
+    helper.executeSeedJob(batch);
+    List<Job> executionJobs = helper.getExecutionJobs(batch);
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processInstanceIds.toArray());
   }
 
   @Test

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/RestartProcessInstanceAsyncTest.java

@@ -42,6 +42,7 @@
 import org.camunda.bpm.engine.batch.Batch;
 import org.camunda.bpm.engine.batch.history.HistoricBatch;
 import org.camunda.bpm.engine.delegate.ExecutionListener;
+import org.camunda.bpm.engine.history.HistoricJobLog;
 import org.camunda.bpm.engine.history.HistoricProcessInstanceQuery;
 import org.camunda.bpm.engine.impl.cfg.ProcessEngineConfigurationImpl;
 import org.camunda.bpm.engine.impl.cfg.multitenancy.TenantIdProvider;
@@ -1150,6 +1151,35 @@ public void shouldSetExecutionStartTimeInBatchAndHistory() {
     Assertions.assertThat(historicBatch.getExecutionStartTime()).isEqualToIgnoringMillis(TEST_DATE);
   }
 
+  @Test
+  public void shouldSetProcessInstanceAndDefinitionIdInHistoryJobLog() {
+    // given
+
+    ProcessDefinition processDefinition = testRule.deployAndGetDefinition(ProcessModels.TWO_TASKS_PROCESS);
+    ProcessInstance processInstance = runtimeService.createProcessInstanceById(processDefinition.getId())
+            .startBeforeActivity("userTask1")
+            .execute();
+    runtimeService.deleteProcessInstance(processInstance.getId(), "test");
+    Batch batch = runtimeService.restartProcessInstances(processDefinition.getId())
+            .startAfterActivity("userTask2")
+            .processInstanceIds(processInstance.getId())
+            .executeAsync();
+    helper.executeSeedJob(batch);
+
+    Job executionJob = helper.getExecutionJobs(batch).get(0);
+    assertEquals(processInstance.getProcessDefinitionId(), executionJob.getProcessDefinitionId());
+    assertEquals(processInstance.getRootProcessInstanceId(), executionJob.getProcessInstanceId());
+
+    // when
+    helper.executeJob(executionJob);
+
+    // then
+    HistoricJobLog jobLog = historyService.createHistoricJobLogQuery().jobDefinitionType(Batch.TYPE_PROCESS_INSTANCE_RESTART).list().get(0);
+
+    assertEquals(processInstance.getProcessDefinitionId(), jobLog.getProcessDefinitionId());
+    assertEquals(processInstance.getRootProcessInstanceId(), jobLog.getProcessInstanceId());
+  }
+
   protected void assertBatchCreated(Batch batch, int processInstanceCount) {
     assertNotNull(batch);
     assertNotNull(batch.getId());

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/RuntimeServiceAsyncOperationsTest.java

@@ -93,6 +93,27 @@ public void testDeleteProcessInstancesAsyncWithList() throws Exception {
     assertProcessInstancesAreDeleted();
   }
 
+  @Deployment(resources = {
+          "org/camunda/bpm/engine/test/api/oneTaskProcess.bpmn20.xml"})
+  @Test
+  public void testDeleteProcessInstances_shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
+    // given
+    List<String> processIds = startTestProcesses(2);
+
+    // when
+    Batch batch = runtimeService.deleteProcessInstancesAsync(processIds, null, TESTING_INSTANCE_DELETE);
+    completeSeedJobs(batch);
+
+    List<Job> executionJobs = managementService.createJobQuery().jobDefinitionId(batch.getBatchJobDefinitionId()).list();
+
+    // then
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processIds.toArray());
+  }
+
+
   @Deployment(resources = {
       "org/camunda/bpm/engine/test/api/oneTaskProcess.bpmn20.xml"})
   @Test

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/SetVariablesBatchTest.java

@@ -685,6 +685,7 @@ public void shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
     rule.executeSeedJobs(batch);
 
     // then
+    //Making sure that processInstanceId is set in execution jobs #4205
     List<Job> executionJobs = rule.getExecutionJobs(batch);
     assertThat(executionJobs)
         .extracting("processInstanceId")

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/UpdateProcessInstancesSuspendStateAsyncTest.java

@@ -194,9 +194,20 @@ public void testBatchActivationById() {
     // when
     Batch suspendprocess = runtimeService.updateProcessInstanceSuspensionState().byProcessInstanceIds(Arrays.asList(processInstance1.getId(), processInstance2.getId())).suspendAsync();
     helper.completeSeedJobs(suspendprocess);
+
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(helper.getExecutionJobs(suspendprocess))
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processInstance1.getId(), processInstance2.getId());
+
     helper.executeJobs(suspendprocess);
     Batch activateprocess = runtimeService.updateProcessInstanceSuspensionState().byProcessInstanceIds(Arrays.asList(processInstance1.getId(), processInstance2.getId())).activateAsync();
     helper.completeSeedJobs(activateprocess);
+
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(helper.getExecutionJobs(activateprocess))
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processInstance1.getId(), processInstance2.getId());
     helper.executeJobs(activateprocess);
 
 

engine/src/test/java/org/camunda/bpm/engine/test/api/runtime/migration/batch/BatchMigrationTest.java

@@ -944,6 +944,41 @@ public void shouldSetExecutionStartTimeInBatchAndHistory() {
     Assertions.assertThat(historicBatch.getExecutionStartTime()).isEqualToIgnoringMillis(TEST_DATE);
   }
 
+  @Test
+  public void shouldCreateProcessInstanceRelatedBatchJobsForSingleInvocations() {
+    RuntimeService runtimeService = engineRule.getRuntimeService();
+    int processInstanceCount = 2;
+
+    ProcessDefinition sourceProcessDefinition = migrationRule.deployAndGetDefinition(ProcessModels.ONE_TASK_PROCESS);
+    ProcessDefinition targetProcessDefinition = migrationRule.deployAndGetDefinition(ProcessModels.ONE_TASK_PROCESS);
+
+    List<String> processInstanceIds = new ArrayList<>();
+    for (int i = 0; i < processInstanceCount; i++) {
+      processInstanceIds.add(
+              runtimeService.startProcessInstanceById(sourceProcessDefinition.getId()).getId()
+      );
+    }
+
+    MigrationPlan migrationPlan = engineRule.getRuntimeService()
+            .createMigrationPlan(sourceProcessDefinition.getId(), targetProcessDefinition.getId())
+            .mapEqualActivities()
+            .build();
+
+    // when
+    Batch batch = runtimeService.newMigration(migrationPlan)
+            .processInstanceIds(processInstanceIds)
+            .executeAsync();
+
+    helper.executeSeedJob(batch);
+    List<Job> executionJobs = helper.getExecutionJobs(batch);
+
+    // then
+    //Making sure that processInstanceId is set in execution jobs #4205
+    assertThat(executionJobs)
+            .extracting("processInstanceId")
+            .containsExactlyInAnyOrder(processInstanceIds.toArray());
+  }
+
   protected void assertBatchCreated(Batch batch, int processInstanceCount) {
     assertNotNull(batch);
     assertNotNull(batch.getId());