Fix flaky test failures on Kafka, DB2 and SQL Server

- TestEventConsumer: use CopyOnWriteArrayList instead of ArrayList to
  fix ConcurrentModificationException when Kafka listener threads add
  events while test assertions stream over the list

- ProcessDefinitionSuspensionTest: use
  waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs instead of
  waitForJobExecutorToProcessAllJobs so the wait does not exit before
  timer jobs are acquired on slow databases like DB2

- ExternalWorkerServiceTaskTest (BPMN + CMMN): replace
  isEqualToIgnoringMillis with isCloseTo(date, 1000) to handle SQL
  Server datetime rounding across second boundaries

Author: filiphr

modules/flowable-cmmn-engine/src/test/java/org/flowable/cmmn/test/externalworker/ExternalWorkerServiceTaskTest.java

@@ -977,7 +977,7 @@ public void testCaseInstanceIsCorrectlyLocked() {
                 .singleResult();
 
         assertThat(caseInstance.getLockOwner()).isEqualTo("worker1");
-        assertThat(caseInstance.getLockTime()).isEqualToIgnoringMillis(acquiredJob.getLockExpirationTime());
+        assertThat(caseInstance.getLockTime()).isCloseTo(acquiredJob.getLockExpirationTime(), 1000);
 
         cmmnEngineConfiguration.getCommandExecutor().execute(new ClearCaseInstanceLockTimesCmd(
                 cmmnEngineConfiguration.getAsyncExecutor().getLockOwner(), cmmnEngineConfiguration));
@@ -988,7 +988,7 @@ public void testCaseInstanceIsCorrectlyLocked() {
                 .singleResult();
 
         assertThat(caseInstance.getLockOwner()).isEqualTo("worker1");
-        assertThat(caseInstance.getLockTime()).isEqualToIgnoringMillis(acquiredJob.getLockExpirationTime());
+        assertThat(caseInstance.getLockTime()).isCloseTo(acquiredJob.getLockExpirationTime(), 1000);
 
         cmmnEngineConfiguration.getCommandExecutor().execute(new ClearCaseInstanceLockTimesCmd("worker1", cmmnEngineConfiguration));
 

modules/flowable-engine/src/test/java/org/flowable/engine/test/api/repository/ProcessDefinitionSuspensionTest.java

@@ -244,7 +244,7 @@ public void testJobIsExecutedOnProcessDefinitionSuspend() {
 
         // The jobs should simply be executed
         processEngineConfiguration.getClock().setCurrentTime(new Date(now.getTime() + (60 * 60 * 1000))); // Timer is set to fire on 5 minutes
-        waitForJobExecutorToProcessAllJobs(2000L, 100L);
+        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(2000L, 100L);
         assertThat(managementService.createJobQuery().count()).isZero();
         assertThat(managementService.createTimerJobQuery().count()).isZero();
     }
@@ -273,7 +273,7 @@ public void testDelayedSuspendProcessDefinition() {
         // Move clock 8 days further and let job executor run
         long eightDaysSinceStartTime = oneWeekFromStartTime + (24 * 60 * 60 * 1000);
         processEngineConfiguration.getClock().setCurrentTime(new Date(eightDaysSinceStartTime));
-        waitForJobExecutorToProcessAllJobs(7000L, 200L);
+        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(7000L, 200L);
 
         // verify job is now removed
         assertThat(managementService.createJobQuery().processDefinitionId(processDefinition.getId()).count()).isZero();
@@ -329,7 +329,7 @@ public void testDelayedSuspendProcessDefinitionIncludingProcessInstances() {
         // Move clock 9 days further and let job executor run
         long eightDaysSinceStartTime = oneWeekFromStartTime + (2 * 24 * 60 * 60 * 1000);
         processEngineConfiguration.getClock().setCurrentTime(new Date(eightDaysSinceStartTime));
-        waitForJobExecutorToProcessAllJobs(7000L, 50L);
+        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(7000L, 50L);
 
         // Try to start process instance. It should fail now.
         assertThatThrownBy(() -> runtimeService.startProcessInstanceById(processDefinition.getId()))
@@ -380,7 +380,7 @@ public void testDelayedActivateProcessDefinition() {
         // Move clock two days and let job executor run
         long twoDaysFromStart = startTime.getTime() + (2 * 24 * 60 * 60 * 1000);
         processEngineConfiguration.getClock().setCurrentTime(new Date(twoDaysFromStart));
-        waitForJobExecutorToProcessAllJobs(7000L, 50L);
+        waitForJobExecutorToProcessAllJobsAndExecutableTimerJobs(7000L, 50L);
 
         // Starting a process instance should now succeed
         runtimeService.startProcessInstanceById(processDefinition.getId());

modules/flowable-engine/src/test/java/org/flowable/engine/test/externalworker/ExternalWorkerServiceTaskTest.java

@@ -994,7 +994,7 @@ void testProcessInstanceIsCorrectlyLocked() {
                 .singleResult();
 
         assertThat(processInstance.getLockOwner()).isEqualTo("worker1");
-        assertThat(processInstance.getLockTime()).isEqualToIgnoringMillis(acquiredJob.getLockExpirationTime());
+        assertThat(processInstance.getLockTime()).isCloseTo(acquiredJob.getLockExpirationTime(), 1000);
 
         managementService.executeCommand(new ClearProcessInstanceLockTimesCmd(processEngineConfiguration.getAsyncExecutor().getLockOwner()));
 
@@ -1004,7 +1004,7 @@ void testProcessInstanceIsCorrectlyLocked() {
                 .singleResult();
 
         assertThat(processInstance.getLockOwner()).isEqualTo("worker1");
-        assertThat(processInstance.getLockTime()).isEqualToIgnoringMillis(acquiredJob.getLockExpirationTime());
+        assertThat(processInstance.getLockTime()).isCloseTo(acquiredJob.getLockExpirationTime(), 1000);
 
         managementService.executeCommand(new ClearProcessInstanceLockTimesCmd("worker1"));
 

modules/flowable-event-registry-spring/src/test/java/org/flowable/eventregistry/spring/test/TestEventConsumer.java

@@ -12,8 +12,8 @@
  */
 package org.flowable.eventregistry.spring.test;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.Objects;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -31,7 +31,7 @@
  */
 public class TestEventConsumer implements EventRegistryEventConsumer {
 
-    protected final List<EventRegistryEvent> events = new ArrayList<>();
+    protected final List<EventRegistryEvent> events = new CopyOnWriteArrayList<>();
     protected Consumer<EventRegistryEvent> eventConsumer = event -> {};
 
     @Override