feat(engine): set removal time in chunks (camunda#3687)

* Makes the set removal time batch job repeatable, so it can run multiple
  times to update in chunks, keeping updates to a defined limit of rows.
* This has to be enabled specifically per batch execution to avoid more
  complex update queries running by default on all databases.
* The new flag is `updateInChunks` which can be set via Java and REST API.
  In combination with it, the `chunkSize` can also be defined dynamically,
  falling back to a defined default that is also configurable in the engine.

related to camunda#3064

engine-rest/engine-rest-openapi/src/main/templates/models/org/camunda/bpm/engine/rest/dto/history/SetRemovalTimeToHistoricProcessInstancesDto.ftl

@@ -16,9 +16,28 @@
   <@lib.property
       name = "hierarchical"
       type = "boolean"
-      last = true
       desc = "Sets the removal time to all historic process instances in the hierarchy.
               Value may only be `true`, as `false` is the default behavior."/>
 
+  <@lib.property
+      name = "updateInChunks"
+      type = "boolean"
+      desc = "Handles removal time updates in chunks, taking into account the defined size in
+              `removalTimeUpdateChunkSize` in the process engine configuration. The size of the 
+              chunks can also be overridden per call with the `updateChunkSize` parameter.
+              Enabling this option can lead to multiple executions of the resulting jobs, preventing
+              the database transaction from timing out by limiting the number of rows to update.
+              Value may only be `true`, as `false` is the default behavior."/>
+
+  <@lib.property
+      name = "updateChunkSize"
+      type = "integer"
+      format = "int32"
+      last = true
+      desc = "Defines the size of the chunks in which removal time updates are processed.
+              The value must be a positive integer between `1` and `500`. This only has an 
+              effect if `updateInChunks` is set to `true`. If undefined, the operation uses the 
+              `removalTimeUpdateChunkSize` defined in the process engine configuration."/>
+
 </@lib.dto>
 </#macro>

engine-rest/engine-rest-openapi/src/main/templates/paths/history/process-instance/set-removal-time/post.ftl

@@ -26,6 +26,22 @@
                                   "b4d2ad94-7240-11e9-98b7-be5e0f7575b7"
                                 ]
                               }
+                   }',
+                   '"example-2": {
+                     "summary": "POST `/history/process-instance/set-removal-time`",
+                     "value": {
+                                "absoluteRemovalTime": "2019-05-05T11:56:24.725+0200",
+                                "hierarchical": true,
+                                "updateInChunks": true,
+                                "updateChunkSize": 300,
+                                "historicProcessInstanceQuery": {
+                                  "unfinished": true
+                                },
+                                "historicProcessInstanceIds": [
+                                  "b4d2ad98-7240-11e9-98b7-be5e0f7575b7",
+                                  "b4d2ad94-7240-11e9-98b7-be5e0f7575b7"
+                                ]
+                              }
                    }'
                 ] />
   "responses" : {

engine-rest/engine-rest/src/main/java/org/camunda/bpm/engine/rest/dto/history/batch/removaltime/SetRemovalTimeToHistoricProcessInstancesDto.java

@@ -26,6 +26,8 @@ public class SetRemovalTimeToHistoricProcessInstancesDto extends AbstractSetRemo
   protected String[] historicProcessInstanceIds;
   protected HistoricProcessInstanceQueryDto historicProcessInstanceQuery;
   protected boolean hierarchical;
+  protected boolean updateInChunks;
+  protected Integer updateChunkSize;
 
   public String[] getHistoricProcessInstanceIds() {
     return historicProcessInstanceIds;
@@ -51,4 +53,19 @@ public void setHierarchical(boolean hierarchical) {
     this.hierarchical = hierarchical;
   }
 
+  public boolean isUpdateInChunks() {
+    return updateInChunks;
+  }
+
+  public void setUpdateInChunks(boolean updateInChunks) {
+    this.updateInChunks = updateInChunks;
+  }
+
+  public Integer getUpdateChunkSize() {
+    return updateChunkSize;
+  }
+
+  public void setUpdateChunkSize(Integer updateChunkSize) {
+    this.updateChunkSize = updateChunkSize;
+  }
 }

engine-rest/engine-rest/src/main/java/org/camunda/bpm/engine/rest/impl/history/HistoricProcessInstanceRestServiceImpl.java

@@ -194,6 +194,15 @@ public BatchDto setRemovalTimeAsync(SetRemovalTimeToHistoricProcessInstancesDto
 
     }
 
+    if (dto.isUpdateInChunks()) {
+      builder.updateInChunks();
+    }
+
+    Integer chunkSize = dto.getUpdateChunkSize();
+    if (chunkSize != null) {
+      builder.chunkSize(chunkSize);
+    }
+
     Batch batch = builder.executeAsync();
     return BatchDto.fromBatch(batch);
   }

engine-rest/engine-rest/src/test/java/org/camunda/bpm/engine/rest/history/HistoricProcessInstanceRestServiceInteractionTest.java

@@ -16,8 +16,35 @@
  */
 package org.camunda.bpm.engine.rest.history;
 
+import static io.restassured.RestAssured.given;
+import static io.restassured.path.json.JsonPath.from;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
 import io.restassured.http.ContentType;
 import io.restassured.response.Response;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.ws.rs.core.Response.Status;
 import org.camunda.bpm.engine.AuthorizationException;
 import org.camunda.bpm.engine.BadUserRequestException;
 import org.camunda.bpm.engine.HistoryService;
@@ -42,39 +69,11 @@
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
-import javax.ws.rs.core.Response.Status;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import static io.restassured.RestAssured.given;
-import static io.restassured.path.json.JsonPath.from;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.mockito.Mockito.*;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyList;
-import static org.mockito.Mockito.anyString;
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
-
 public class HistoricProcessInstanceRestServiceInteractionTest extends AbstractRestServiceTest {
 
   @ClassRule
   public static TestContainerRule rule = new TestContainerRule();
-  
+
   protected static final String DELETE_REASON = "deleteReason";
   protected static final String TEST_DELETE_REASON = "test";
   protected static final String FAIL_IF_NOT_EXISTS = "failIfNotExists";
@@ -183,7 +182,7 @@ public void testDeleteNonExistingProcessInstanceIfExists() {
     given().pathParam("id", MockProvider.EXAMPLE_PROCESS_INSTANCE_ID).queryParam("failIfNotExists", false)
     .then().expect().statusCode(Status.NO_CONTENT.getStatusCode())
     .when().delete(HISTORIC_SINGLE_PROCESS_INSTANCE_URL);
-    
+
     verify(historyServiceMock).deleteHistoricProcessInstanceIfExists(MockProvider.EXAMPLE_PROCESS_INSTANCE_ID);
   }
 
@@ -263,7 +262,7 @@ public void testDeleteAsyncWithBadRequestQuery() {
         .statusCode(Status.BAD_REQUEST.getStatusCode())
         .when().post(DELETE_HISTORIC_PROCESS_INSTANCES_ASYNC_URL);
   }
-  
+
   @Test
   public void testDeleteAllVariablesByProcessInstanceId() {
     given()
@@ -275,12 +274,12 @@ public void testDeleteAllVariablesByProcessInstanceId() {
 
     verify(historyServiceMock).deleteHistoricVariableInstancesByProcessInstanceId(EXAMPLE_PROCESS_INSTANCE_ID);
   }
-  
+
   @Test
   public void testDeleteAllVariablesForNonExistingProcessInstance() {
     doThrow(new NotFoundException("No historic process instance found with id: 'NON_EXISTING_ID'"))
     .when(historyServiceMock).deleteHistoricVariableInstancesByProcessInstanceId("NON_EXISTING_ID");
-    
+
     given()
       .pathParam("id", "NON_EXISTING_ID")
     .expect()
@@ -385,7 +384,7 @@ public void shouldSetRemovalTime_Absolute() {
   }
 
   @Test
-  public void shouldNotSetRemovalTime_Absolute() {
+  public void shouldSetRemovalTime_AbsoluteNoTime() {
     SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builderMock =
       mock(SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder.class, RETURNS_DEEP_STUBS);
 
@@ -413,7 +412,7 @@ public void shouldNotSetRemovalTime_Absolute() {
   }
 
   @Test
-  public void shouldClearRemovalTime() {
+  public void shouldSetRemovalTime_ClearTime() {
     SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builderMock =
       mock(SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder.class, RETURNS_DEEP_STUBS);
 
@@ -464,7 +463,7 @@ public void shouldSetRemovalTime_Response() {
   }
 
   @Test
-  public void shouldSetRemovalTime_ThrowBadUserException() {
+  public void shouldSetRemovalTime_FailBadUserRequest() {
     SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builderMock =
       mock(SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder.class, RETURNS_DEEP_STUBS);
 
@@ -481,6 +480,70 @@ public void shouldSetRemovalTime_ThrowBadUserException() {
       .post(SET_REMOVAL_TIME_HISTORIC_PROCESS_INSTANCES_ASYNC_URL);
   }
 
+  @Test
+  public void shouldSetRemovalTime_InChunks() {
+    SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builderMock =
+        mock(SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder.class, RETURNS_DEEP_STUBS);
+
+      when(historyServiceMock.setRemovalTimeToHistoricProcessInstances())
+        .thenReturn(builderMock);
+
+      Map<String, Object> payload = new HashMap<>();
+      payload.put("historicProcessInstanceIds", Collections.singletonList(EXAMPLE_PROCESS_INSTANCE_ID));
+      payload.put("clearedRemovalTime", true);
+      payload.put("updateInChunks", true);
+
+      given()
+        .contentType(ContentType.JSON)
+        .body(payload)
+      .then()
+        .expect().statusCode(Status.OK.getStatusCode())
+      .when()
+        .post(SET_REMOVAL_TIME_HISTORIC_PROCESS_INSTANCES_ASYNC_URL);
+
+      SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builder =
+        historyServiceMock.setRemovalTimeToHistoricProcessInstances();
+
+      verify(builder).clearedRemovalTime();
+      verify(builder).byIds(EXAMPLE_PROCESS_INSTANCE_ID);
+      verify(builder).byQuery(null);
+      verify(builder).updateInChunks();
+      verify(builder).executeAsync();
+      verifyNoMoreInteractions(builder);
+  }
+
+  @Test
+  public void shouldSetRemovalTime_ChunkSize() {
+    SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builderMock =
+        mock(SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder.class, RETURNS_DEEP_STUBS);
+
+      when(historyServiceMock.setRemovalTimeToHistoricProcessInstances())
+        .thenReturn(builderMock);
+
+      Map<String, Object> payload = new HashMap<>();
+      payload.put("historicProcessInstanceIds", Collections.singletonList(EXAMPLE_PROCESS_INSTANCE_ID));
+      payload.put("clearedRemovalTime", true);
+      payload.put("updateChunkSize", 20);
+
+      given()
+        .contentType(ContentType.JSON)
+        .body(payload)
+      .then()
+        .expect().statusCode(Status.OK.getStatusCode())
+      .when()
+        .post(SET_REMOVAL_TIME_HISTORIC_PROCESS_INSTANCES_ASYNC_URL);
+
+      SetRemovalTimeSelectModeForHistoricProcessInstancesBuilder builder =
+        historyServiceMock.setRemovalTimeToHistoricProcessInstances();
+
+      verify(builder).clearedRemovalTime();
+      verify(builder).byIds(EXAMPLE_PROCESS_INSTANCE_ID);
+      verify(builder).byQuery(null);
+      verify(builder).chunkSize(20);
+      verify(builder).executeAsync();
+      verifyNoMoreInteractions(builder);
+  }
+
   @Test
   public void testOrQuery() {
     // given

engine/src/main/java/org/camunda/bpm/engine/history/SetRemovalTimeToHistoricProcessInstancesBuilder.java

@@ -22,6 +22,7 @@
 import org.camunda.bpm.engine.authorization.Permissions;
 import org.camunda.bpm.engine.authorization.Resources;
 import org.camunda.bpm.engine.batch.Batch;
+import org.camunda.bpm.engine.impl.batch.removaltime.ProcessSetRemovalTimeJobHandler;
 
 /**
  * Fluent builder to set the removal time to historic process instances and
@@ -59,6 +60,36 @@ public interface SetRemovalTimeToHistoricProcessInstancesBuilder {
    */
   SetRemovalTimeToHistoricProcessInstancesBuilder hierarchical();
 
+  /**
+   * Handles removal time updates in chunks, taking into account the defined
+   * size in {@code removalTimeUpdateChunkSize} in the process engine
+   * configuration. The size of the chunks can also be overridden per call with
+   * the {@link #chunkSize(int)} option. Enabling this option can lead to
+   * multiple executions of the resulting jobs, preventing the database
+   * transaction from timing out by limiting the number of rows to update.
+   *
+   * @since 7.20
+   *
+   * @return the builder.
+   */
+  SetRemovalTimeToHistoricProcessInstancesBuilder updateInChunks();
+
+  /**
+   * Defines the size of the chunks in which removal time updates are processed.
+   * The value must be a positive integer value that doesn't exceed the
+   * {@link ProcessSetRemovalTimeJobHandler#MAX_CHUNK_SIZE}.
+   * 
+   * Only has an effect if {@link #updateInChunks()} is invoked as well.
+   * 
+   * If undefined, the operation uses the `removalTimeUpdateChunkSize` defined
+   * in the process engine configuration.
+   *
+   * @since 7.20
+   *
+   * @return the builder.
+   */
+  SetRemovalTimeToHistoricProcessInstancesBuilder chunkSize(int chunkSize);
+
   /**
    * Sets the removal time asynchronously as batch. The returned batch can be used to
    * track the progress of setting a removal time.