Bugfix/npe task enrichment (#485)

* run in context, and NPE handling

* Move task enrichment into Camunda Command and fix #473

digiwf-integrations/digiwf-cosys-integration/digiwf-cosys-integration-client/.openapi-generator/VERSION

@@ -1 +1 @@
-6.5.0
+6.6.0

digiwf-integrations/digiwf-s3-integration/digiwf-s3-integration-client/.openapi-generator/VERSION

@@ -1 +1 @@
-6.5.0
+6.6.0

digiwf-schema-registry/digiwf-schema-registry-client/.openapi-generator/VERSION

@@ -1 +1 @@
-6.5.0
+6.6.0

digiwf-schema-registry/digiwf-schema-registry-client/src/main/java/io/muenchendigital/digiwf/schema/registry/gen/ApiClient.java

@@ -561,7 +561,7 @@ public <T> ResponseEntity<T> invokeAPI(String path, HttpMethod method, Map<Strin
             throw new RestClientException("Could not build URL: " + builder.toUriString(), ex);
         }
 
-        final BodyBuilder requestBuilder = RequestEntity.method(method, uri);
+        final BodyBuilder requestBuilder = RequestEntity.method(method, UriComponentsBuilder.fromHttpUrl(basePath).toUriString() + finalUri, uriParams);
         if (accept != null) {
             requestBuilder.accept(accept.toArray(new MediaType[accept.size()]));
         }

digiwf-task/digiwf-polyflow-connector-starter/src/main/java/io/muenchendigital/digiwf/task/importer/TaskImporterService.java

@@ -6,31 +6,39 @@
 import io.muenchendigital.digiwf.task.listener.TaskDescriptionCreateTaskListener;
 import io.muenchendigital.digiwf.task.listener.TaskSchemaTypeCreateTaskListener;
 import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
+import org.camunda.bpm.engine.ProcessEngineConfiguration;
 import org.camunda.bpm.engine.TaskService;
+import org.camunda.bpm.engine.context.ProcessEngineContext;
+import org.camunda.bpm.engine.impl.cfg.ProcessEngineConfigurationImpl;
 import org.camunda.bpm.engine.impl.persistence.entity.TaskEntity;
 import org.springframework.http.ResponseEntity;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.security.RolesAllowed;
-
 import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 
+import static org.springframework.http.ResponseEntity.internalServerError;
 import static org.springframework.http.ResponseEntity.noContent;
 
 @RequiredArgsConstructor
 @RestController
 @Slf4j
+@Transactional
 public class TaskImporterService {
 
     public static final String CLIENT_IMPORT_TASKS = "clientrole_task_importer";
     private final TaskServiceCollectorService taskServiceCollectorService;
     private final TaskService taskService;
+    private final ProcessEngineConfiguration processEngineConfiguration;
 
     private final AssignmentCreateTaskListener assignmentCreateTaskListener;
     private final CancelableTaskStatusCreateTaskListener cancelableTaskStatusCreateTaskListener;
@@ -45,33 +53,45 @@ void inform() {
     @PostMapping("/rest/admin/tasks/enrich")
     @RolesAllowed(CLIENT_IMPORT_TASKS)
     public ResponseEntity<Void> enrichExistingTasks() {
-
-        log.info("Selecting candidates for task enrichment from " + taskService.createTaskQuery().active().count() + " tasks.");
+        log.info("Starting task enrichment");
         val tasks = new HashSet<TaskEntity>();
         tasks.addAll(taskService.createTaskQuery()
             .active()
-            .withCandidateUsers()
-            .list().stream().map(task -> ((TaskEntity)task)).collect(Collectors.toList()));
+            .taskAssigned()
+            .list()
+            .stream().map(task -> ((TaskEntity) task))
+            .collect(Collectors.toList()));
         tasks.addAll(taskService.createTaskQuery()
-          .active()
-          .withCandidateGroups()
-          .list().stream().map(task -> ((TaskEntity)task)).collect(Collectors.toList()));
+            .active()
+            .withCandidateGroups()
+            .list()
+            .stream().map(task -> ((TaskEntity) task))
+            .collect(Collectors.toList()));
         tasks.addAll(taskService.createTaskQuery()
-          .active()
-          .taskAssigned()
-          .list().stream().map(task -> ((TaskEntity)task)).collect(Collectors.toList()));
-
-        log.info("Selected for enrichment " + tasks.size() + " tasks");
+            .active()
+            .withCandidateUsers()
+            .list()
+            .stream().map(task -> ((TaskEntity) task))
+            .collect(Collectors.toList()));
+        enrichTasks(tasks);
+        return noContent().build();
+    }
 
-        tasks.forEach((task) -> {
+  @SneakyThrows
+  private void enrichTasks(Set<TaskEntity> tasks) {
+      log.info("Selected for enrichment {} tasks", tasks.size());
+    ((ProcessEngineConfigurationImpl)processEngineConfiguration).getCommandExecutorTxRequired().execute(
+        (context) -> {
+          tasks.forEach((task) -> {
             assignmentCreateTaskListener.taskCreated(task);
             cancelableTaskStatusCreateTaskListener.taskCreated(task);
             taskSchemaTypeCreateTaskListener.taskCreated(task);
             taskDescriptionCreateTaskListener.taskCreated(task);
-        });
-
-        log.info("Enrichment of " + tasks.size() + " tasks finished");
-        return noContent().build();
+          });
+          log.info("Enrichment of {} tasks finished", tasks.size());
+          return null;
+        }
+    );
     }
 
 

digiwf-task/digiwf-polyflow-connector-starter/src/test/java/io/muenchendigital/digiwf/task/importer/TaskImporterServiceTest.java

@@ -25,7 +25,7 @@
 import static org.mockito.Mockito.mock;
 
 @Disabled("This test is disabled from CI, since it uses the same in-mem H2 DB as CancelableTaskStatusCreateTaskListenerTest and interfers with it")
-// FIXME - make sure we can run multiple tests with the engine -> maybe a dedicted ITEst-module for this is required.
+// FIXME - make sure we can run multiple tests with the engine -> maybe a dedicated ITEst-module for this is required.
 class TaskImporterServiceTest {
 
   @RegisterExtension
@@ -51,6 +51,7 @@ class TaskImporterServiceTest {
   private final TaskImporterService service = new TaskImporterService(
       mock(TaskServiceCollectorService.class),
       taskService,
+      extension.getProcessEngineConfiguration(),
       listenerMock,
       mock(CancelableTaskStatusCreateTaskListener.class),
       mock(TaskSchemaTypeCreateTaskListener.class),

digiwf-task/digiwf-polyflow-connector-starter/src/test/java/io/muenchendigital/digiwf/task/listener/AssignmentCreateTaskListenerTest.java

@@ -2,13 +2,16 @@
 
 import io.muenchendigital.digiwf.task.TaskManagementProperties;
 import org.assertj.core.util.Lists;
+import org.camunda.bpm.engine.impl.persistence.entity.ExecutionEntity;
+import org.camunda.bpm.engine.impl.persistence.entity.TaskEntity;
 import org.camunda.community.mockito.delegate.DelegateTaskFake;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 import static io.muenchendigital.digiwf.task.TaskVariables.*;
 import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 class AssignmentCreateTaskListenerTest {
@@ -27,7 +30,6 @@ void setup_task() {
     delegateTask.addCandidateUser("candidateUser2");
   }
 
-
   @Test
   public void is_disabled_by_properties() {
     when(properties.isShadow()).thenReturn(false);