fix(engine): Pass-Through RemovalTime to DefaultHistoryEventProducer

- removalTime of a failing clean-up job's bytearray was null. As a result, that entry was not getting cleaned up. This fix ensures passing the value through.

Related to: camunda#3288

engine/src/main/java/org/camunda/bpm/engine/impl/history/producer/DefaultHistoryEventProducer.java

@@ -1090,11 +1090,9 @@ public HistoryEvent createHistoricJobLogFailedEvt(Job job, Throwable exception)
       byte[] exceptionBytes = toByteArray(exceptionStacktrace);
 
       ByteArrayEntity byteArray = createJobExceptionByteArray(exceptionBytes, ResourceTypes.HISTORY);
-      byteArray.setRootProcessInstanceId(event.getRootProcessInstanceId());
 
-      if (isHistoryRemovalTimeStrategyStart()) {
-        byteArray.setRemovalTime(event.getRemovalTime());
-      }
+      byteArray.setRootProcessInstanceId(event.getRootProcessInstanceId());
+      byteArray.setRemovalTime(event.getRemovalTime());
 
       event.setExceptionByteArrayId(byteArray.getId());
     }

engine/src/main/java/org/camunda/bpm/engine/impl/jobexecutor/historycleanup/HistoryCleanupJobHandler.java

@@ -49,11 +49,15 @@ public void execute(HistoryCleanupJobHandlerConfiguration configuration, Executi
 
   }
 
+  protected HistoryCleanupRemovalTime getTimeBasedHandler() {
+    return new HistoryCleanupRemovalTime();
+  }
+
   protected HistoryCleanupHandler initCleanupHandler(HistoryCleanupJobHandlerConfiguration configuration, CommandContext commandContext) {
     HistoryCleanupHandler cleanupHandler = null;
 
     if (isHistoryCleanupStrategyRemovalTimeBased(commandContext)) {
-      cleanupHandler = new HistoryCleanupRemovalTime();
+      cleanupHandler = getTimeBasedHandler();
     } else {
       cleanupHandler = new HistoryCleanupBatch();
     }

engine/src/test/java/org/camunda/bpm/engine/test/api/history/removaltime/cleanup/HistoryCleanupByteArrayRemovalTest.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH
+ * under one or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. Camunda licenses this file to you under the Apache License,
+ * Version 2.0; 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.camunda.bpm.engine.test.api.history.removaltime.cleanup;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.camunda.bpm.engine.ProcessEngineConfiguration.HISTORY_CLEANUP_STRATEGY_REMOVAL_TIME_BASED;
+import static org.camunda.bpm.engine.ProcessEngineConfiguration.HISTORY_FULL;
+import static org.camunda.bpm.engine.ProcessEngineConfiguration.HISTORY_REMOVAL_TIME_STRATEGY_END;
+import static org.camunda.bpm.engine.impl.jobexecutor.historycleanup.HistoryCleanupHandler.MAX_BATCH_SIZE;
+
+import java.util.Map;
+import org.camunda.bpm.engine.HistoryService;
+import org.camunda.bpm.engine.ManagementService;
+import org.camunda.bpm.engine.ProcessEngine;
+import org.camunda.bpm.engine.impl.cfg.ProcessEngineConfigurationImpl;
+import org.camunda.bpm.engine.impl.db.DbEntity;
+import org.camunda.bpm.engine.impl.db.entitymanager.operation.DbOperation;
+import org.camunda.bpm.engine.impl.history.DefaultHistoryRemovalTimeProvider;
+import org.camunda.bpm.engine.impl.history.event.HistoricJobLogEvent;
+import org.camunda.bpm.engine.impl.jobexecutor.historycleanup.HistoryCleanupJobHandler;
+import org.camunda.bpm.engine.impl.jobexecutor.historycleanup.HistoryCleanupRemovalTime;
+import org.camunda.bpm.engine.impl.persistence.entity.ByteArrayEntity;
+import org.camunda.bpm.engine.test.ProcessEngineRule;
+import org.camunda.bpm.engine.test.RequiredHistoryLevel;
+import org.camunda.bpm.engine.test.api.resources.GetByteArrayCommand;
+import org.camunda.bpm.engine.test.util.EntityRemoveRule;
+import org.camunda.bpm.engine.test.util.ProcessEngineBootstrapRule;
+import org.camunda.bpm.engine.test.util.ProcessEngineTestRule;
+import org.camunda.bpm.engine.test.util.ProvidedProcessEngineRule;
+import org.camunda.bpm.engine.test.util.RemoveAfter;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.RuleChain;
+
+@RequiredHistoryLevel(HISTORY_FULL)
+public class HistoryCleanupByteArrayRemovalTest {
+
+  private ProcessEngineBootstrapRule bootstrapRule = new ProcessEngineBootstrapRule(config -> {
+
+    config.setHistoryRemovalTimeStrategy(HISTORY_REMOVAL_TIME_STRATEGY_END)
+        .setHistoryRemovalTimeProvider(new DefaultHistoryRemovalTimeProvider())
+        .initHistoryRemovalTime();
+
+    config.setHistoryCleanupStrategy(HISTORY_CLEANUP_STRATEGY_REMOVAL_TIME_BASED);
+
+    config.setHistoryCleanupBatchSize(MAX_BATCH_SIZE);
+    config.setHistoryCleanupBatchWindowStartTime(null);
+    config.setHistoryCleanupDegreeOfParallelism(1);
+
+    config.setBatchOperationHistoryTimeToLive(null);
+    config.setBatchOperationsForHistoryCleanup(null);
+
+    config.setHistoryTimeToLive(null);
+
+    config.setTaskMetricsEnabled(false);
+    config.setTaskMetricsTimeToLive(null);
+
+    config.initHistoryCleanup();
+  });
+
+  protected ProcessEngineRule engineRule = new ProvidedProcessEngineRule(bootstrapRule);
+  protected ProcessEngineTestRule testRule = new ProcessEngineTestRule(engineRule);
+  protected EntityRemoveRule entityRemoveRule = EntityRemoveRule.ofLazyRule(() -> testRule);
+
+  @Rule
+  public RuleChain ruleChain = RuleChain.outerRule(bootstrapRule)
+      .around(engineRule)
+      .around(testRule)
+      .around(entityRemoveRule);
+
+  private ManagementService managementService;
+  private HistoryService historyService;
+  private ProcessEngineConfigurationImpl engineConfiguration;
+
+  @Before
+  public void init() {
+    ProcessEngine processEngine = bootstrapRule.getProcessEngine();
+
+    managementService = processEngine.getManagementService();
+    historyService = processEngine.getHistoryService();
+    engineConfiguration = (ProcessEngineConfigurationImpl) processEngine.getProcessEngineConfiguration();
+  }
+
+  @After
+  public void tearDown() {
+    restoreCleanupJobHandler();
+    testRule.deleteHistoryCleanupJobs();
+  }
+
+  @Test
+  @RemoveAfter
+  public void shouldHaveRemovalTimeOnFailingHistoryCleanupJob() {
+    // given
+    engineConfiguration.setHistoryCleanupJobLogTimeToLive("1");
+    overrideFailingCleanupJobHandler();
+
+    try {
+      // when
+      runHistoryCleanup();
+      fail("This test should fail during history cleanup and not reach this point");
+    } catch (Exception e) {
+      HistoricJobLogEvent event = getLastFailingHistoryCleanupJobEvent();
+      String exceptionByteArrayId = event.getExceptionByteArrayId();
+      ByteArrayEntity byteArray = findByteArrayById(exceptionByteArrayId);
+
+      // then
+      assertThat(byteArray).isNotNull();
+      assertThat(byteArray.getRemovalTime()).isNotNull();
+    }
+  }
+
+  protected ByteArrayEntity findByteArrayById(String byteArrayId) {
+    return engineConfiguration.getCommandExecutorTxRequired().execute(new GetByteArrayCommand(byteArrayId));
+  }
+
+  protected void restoreCleanupJobHandler() {
+    engineConfiguration.getJobHandlers().put(HistoryCleanupJobHandler.TYPE, new HistoryCleanupJobHandler());
+  }
+
+  protected void overrideFailingCleanupJobHandler() {
+    engineConfiguration.getJobHandlers().put(HistoryCleanupJobHandler.TYPE, new FailingHistoryCleanupJobHandler());
+  }
+
+  protected void runHistoryCleanup() {
+    historyService.cleanUpHistoryAsync(true);
+
+    historyService.findHistoryCleanupJobs().forEach(job -> managementService.executeJob(job.getId()));
+  }
+
+  protected HistoricJobLogEvent getLastFailingHistoryCleanupJobEvent() {
+    return (HistoricJobLogEvent) historyService.createHistoricJobLogQuery()
+        .failureLog()
+        .jobDefinitionType("history-cleanup")
+        .singleResult();
+  }
+
+  /* History Cleanup Job Handler that fails during process cleanup */
+  static class FailingHistoryCleanupJobHandler extends HistoryCleanupJobHandler {
+
+    @Override
+    protected HistoryCleanupRemovalTime getTimeBasedHandler() {
+      return new FailingProcessCleanupRemovalTime();
+    }
+
+    static class FailingProcessCleanupRemovalTime extends HistoryCleanupRemovalTime {
+      @Override
+      protected Map<Class<? extends DbEntity>, DbOperation> performProcessCleanup() {
+        throw new RuntimeException("This operation is always failing!");
+      }
+    }
+  }
+}

engine/src/test/java/org/camunda/bpm/engine/test/util/EntityRemoveRule.java

@@ -19,6 +19,8 @@
 
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
+import java.util.function.Supplier;
+import org.camunda.bpm.engine.ProcessEngine;
 import org.camunda.bpm.engine.task.Task;
 import org.junit.rules.TestWatcher;
 import org.junit.runner.Description;
@@ -35,7 +37,11 @@ public class EntityRemoveRule extends TestWatcher {
 
   private static final Logger LOG = LoggerFactory.getLogger(EntityRemoveRule.class);
 
-  private final Removable removable;
+  protected Removable removable;
+
+  private EntityRemoveRule() {
+    this.removable = Removable.of((ProcessEngine) null);
+  }
 
   private EntityRemoveRule(ProcessEngineTestRule engineTestRule) {
     this.removable = Removable.of(engineTestRule);
@@ -45,6 +51,10 @@ public static EntityRemoveRule of(ProcessEngineTestRule rule) {
     return new EntityRemoveRule(rule);
   }
 
+  public static EntityRemoveRule ofLazyRule(Supplier<ProcessEngineTestRule> ruleSupplier) {
+    return new LazyEntityRemoveRuleProxy(ruleSupplier);
+  }
+
   @Override
   public Statement apply(Statement base, Description description) {
     RemoveAfter removeAfterAnnotation = getAnnotation(description, RemoveAfter.class);
@@ -54,26 +64,46 @@ public Statement apply(Statement base, Description description) {
       return new Statement() {
         @Override
         public void evaluate() throws Throwable {
-
           base.evaluate();
-
-          if (!methodHasRemoveAfterAnnotation) {
-            return;
-          }
-
-          if (hasZeroArguments(removeAfterAnnotation)) {
-            removable.removeAll();
-            return;
-          }
-
-          removable.remove(removeAfterAnnotation.value());
+          executePostEvaluate(removeAfterAnnotation, methodHasRemoveAfterAnnotation);
         }
       };
     } finally {
       LOG.debug("deleteTasks: {}", methodHasRemoveAfterAnnotation);
     }
   }
 
+  protected void executePostEvaluate(RemoveAfter removeAfterAnnotation, boolean methodHasRemoveAfterAnnotation) {
+
+    if (!methodHasRemoveAfterAnnotation) {
+      return;
+    }
+
+    executePreRemoval();
+    executeRemoval(removeAfterAnnotation);
+  }
+
+  /**
+   * Hook method to supp
+   */
+  protected void executePreRemoval() {
+  }
+
+  /**
+   * Hook method for executing removal.
+   *
+   * @param removeAfterAnnotation the remove after annotation parameter of the executing method.
+   */
+  protected void executeRemoval(RemoveAfter removeAfterAnnotation) {
+
+    if (hasZeroArguments(removeAfterAnnotation)) {
+      removable.removeAll();
+      return;
+    }
+
+    removable.remove(removeAfterAnnotation.value());
+  }
+
   private boolean hasZeroArguments(RemoveAfter annotation) {
     return annotation.value() == null || annotation.value().length == 0;
   }
@@ -93,4 +123,21 @@ private <T extends Annotation> T getAnnotation(Description description, Class<T>
     }
   }
 
+  /* Proxy that enables EntityRemoveRule to support lazy initialization by initializing the rule using a supplier &
+   *  after the execution of the method, before removal. */
+  private static class LazyEntityRemoveRuleProxy extends EntityRemoveRule {
+
+    private Supplier<ProcessEngineTestRule> supplier;
+
+    public LazyEntityRemoveRuleProxy(Supplier<ProcessEngineTestRule> supplier) {
+      super();
+      this.supplier = supplier;
+    }
+
+    @Override
+    protected void executePreRemoval() {
+      removable = Removable.of(supplier.get());
+    }
+  }
+
 }

engine/src/test/java/org/camunda/bpm/engine/test/util/Removable.java

@@ -98,16 +98,14 @@ public static Removable of(ProcessEngineTestRule rule) {
   }
 
   public static Removable of(ProcessEngine engine) {
-    Objects.requireNonNull(engine);
-
     return new Removable(engine);
   }
 
   /**
    * Removes the associated mapped entities from the db for the given class.
    *
    * @param clazz the given class to delete associated entities for
-   * @throws Exception in case anything fails during the process of deletion
+   * @throws EntityRemoveException in case anything fails during the process of deletion
    */
   public void remove(Class<?> clazz) throws EntityRemoveException {
     Objects.requireNonNull(clazz, "remove does not accept null arguments");
@@ -118,6 +116,10 @@ public void remove(Class<?> clazz) throws EntityRemoveException {
       throw new UnsupportedOperationException("class " + clazz.getName() + " is not supported yet for Removal");
     }
 
+    if (!isInitialized()) {
+      throw new EntityRemoveException("Removable is not initialized");
+    }
+
     try {
       runnable.execute();
     } catch (Exception e) {
@@ -129,7 +131,7 @@ public void remove(Class<?> clazz) throws EntityRemoveException {
    * Removes the associated mapped entities from the db for the given classes.
    *
    * @param classes the given classes to delete associated entities for
-   * @throws Exception in case anything fails during the process of deletion for any of the classes
+   * @throws EntityRemoveException in case anything fails during the process of deletion for any of the classes
    */
   public void remove(Class<?>[] classes) throws EntityRemoveException {
     Objects.requireNonNull(classes, "remove does not accept null arguments");
@@ -142,7 +144,7 @@ public void remove(Class<?>[] classes) throws EntityRemoveException {
   /**
    * Removes associated mapped entities for all known classes.
    *
-   * @throws Exception in case anything fails during the process of deletion for any of the classes
+   * @throws EntityRemoveException in case anything fails during the process of deletion for any of the classes
    */
   public void removeAll() throws EntityRemoveException {
     try {
@@ -288,6 +290,10 @@ private void removeAllIncidents() {
     });
   }
 
+  public boolean isInitialized() {
+    return engine != null;
+  }
+
 }
 
 /**
@@ -297,6 +303,10 @@ class EntityRemoveException extends RuntimeException {
   public EntityRemoveException(Exception e) {
     super(e);
   }
+
+  public EntityRemoveException(String message) {
+    super(message);
+  }
 }
 
 /**