More flexibility for implementors of persistence plugin (#1186)

Operations is now a parent interface of transactions. This allow more
flexibility in commit strategies without having to duplicate default
writer logic.

Test now expect a PersistenceHandlers class rather than a store.

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/AbstractPersistenceInstanceReader.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 io.serverlessworkflow.impl.persistence;
+
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+public abstract class AbstractPersistenceInstanceReader implements PersistenceInstanceReader {
+
+  protected final Stream<WorkflowInstance> scanAll(
+      PersistenceInstanceOperations operations,
+      WorkflowDefinition definition,
+      String applicationId) {
+    return operations
+        .scanAll(applicationId, definition)
+        .map(v -> new WorkflowPersistenceInstance(definition, v));
+  }
+
+  protected final Optional<WorkflowInstance> find(
+      PersistenceInstanceOperations operations, WorkflowDefinition definition, String instanceId) {
+    return operations
+        .readWorkflowInfo(definition, instanceId)
+        .map(i -> new WorkflowPersistenceInstance(definition, i));
+  }
+
+  @Override
+  public void close() {}
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/AbstractPersistenceInstanceWriter.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 io.serverlessworkflow.impl.persistence;
+
+import io.serverlessworkflow.impl.TaskContextData;
+import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowStatus;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+public abstract class AbstractPersistenceInstanceWriter implements PersistenceInstanceWriter {
+
+  @Override
+  public CompletableFuture<Void> started(WorkflowContextData workflowContext) {
+    return doTransaction(t -> t.writeInstanceData(workflowContext), workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> completed(WorkflowContextData workflowContext) {
+    return removeProcessInstance(workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> failed(WorkflowContextData workflowContext, Throwable ex) {
+    return removeProcessInstance(workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> aborted(WorkflowContextData workflowContext) {
+    return removeProcessInstance(workflowContext);
+  }
+
+  protected CompletableFuture<Void> removeProcessInstance(WorkflowContextData workflowContext) {
+    return doTransaction(t -> t.removeProcessInstance(workflowContext), workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> taskStarted(
+      WorkflowContextData workflowContext, TaskContextData taskContext) {
+    return CompletableFuture.completedFuture(null);
+  }
+
+  @Override
+  public CompletableFuture<Void> taskRetried(
+      WorkflowContextData workflowContext, TaskContextData taskContext) {
+    return doTransaction(t -> t.writeRetryTask(workflowContext, taskContext), workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> taskCompleted(
+      WorkflowContextData workflowContext, TaskContextData taskContext) {
+    return doTransaction(t -> t.writeCompletedTask(workflowContext, taskContext), workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> suspended(WorkflowContextData workflowContext) {
+    return doTransaction(
+        t -> t.writeStatus(workflowContext, WorkflowStatus.SUSPENDED), workflowContext);
+  }
+
+  @Override
+  public CompletableFuture<Void> resumed(WorkflowContextData workflowContext) {
+    return doTransaction(t -> t.clearStatus(workflowContext), workflowContext);
+  }
+
+  @Override
+  public void close() throws Exception {}
+
+  protected abstract CompletableFuture<Void> doTransaction(
+      Consumer<PersistenceInstanceOperations> operation, WorkflowContextData context);
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/DefaultPersistenceInstanceReader.java

@@ -20,7 +20,7 @@
 import java.util.Optional;
 import java.util.stream.Stream;
 
-public class DefaultPersistenceInstanceReader implements PersistenceInstanceReader {
+public class DefaultPersistenceInstanceReader extends AbstractPersistenceInstanceReader {
 
   private final PersistenceInstanceStore store;
 
@@ -32,7 +32,7 @@ protected DefaultPersistenceInstanceReader(PersistenceInstanceStore store) {
   public Optional<WorkflowInstance> find(WorkflowDefinition definition, String instanceId) {
     PersistenceInstanceTransaction transaction = store.begin();
     try {
-      Optional<WorkflowInstance> instance = read(transaction, definition, instanceId);
+      Optional<WorkflowInstance> instance = find(transaction, definition, instanceId);
       transaction.commit(definition);
       return instance;
     } catch (Exception ex) {
@@ -41,21 +41,10 @@ public Optional<WorkflowInstance> find(WorkflowDefinition definition, String ins
     }
   }
 
-  private Optional<WorkflowInstance> read(
-      PersistenceInstanceTransaction t, WorkflowDefinition definition, String instanceId) {
-    return t.readWorkflowInfo(definition, instanceId)
-        .map(i -> new WorkflowPersistenceInstance(definition, i));
-  }
-
   @Override
   public Stream<WorkflowInstance> scanAll(WorkflowDefinition definition, String applicationId) {
     PersistenceInstanceTransaction transaction = store.begin();
-    return transaction
-        .scanAll(applicationId, definition)
-        .onClose(() -> transaction.commit(definition))
-        .map(v -> new WorkflowPersistenceInstance(definition, v));
+    return super.scanAll(transaction, definition, applicationId)
+        .onClose(() -> transaction.commit(definition));
   }
-
-  @Override
-  public void close() {}
 }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/DefaultPersistenceInstanceWriter.java

@@ -15,26 +15,28 @@
  */
 package io.serverlessworkflow.impl.persistence;
 
-import io.serverlessworkflow.impl.TaskContextData;
 import io.serverlessworkflow.impl.WorkflowContextData;
 import io.serverlessworkflow.impl.WorkflowDefinitionData;
-import io.serverlessworkflow.impl.WorkflowStatus;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-public class DefaultPersistenceInstanceWriter implements PersistenceInstanceWriter {
+public class DefaultPersistenceInstanceWriter extends AbstractPersistenceInstanceWriter {
 
   private final PersistenceInstanceStore store;
   private final Map<String, CompletableFuture<Void>> futuresMap = new ConcurrentHashMap<>();
   private final Optional<ExecutorService> executorService;
   private final Duration closeTimeout;
 
+  private static final Logger logger =
+      LoggerFactory.getLogger(DefaultPersistenceInstanceWriter.class);
+
   protected DefaultPersistenceInstanceWriter(
       PersistenceInstanceStore store,
       Optional<ExecutorService> executorService,
@@ -45,61 +47,14 @@ protected DefaultPersistenceInstanceWriter(
   }
 
   @Override
-  public CompletableFuture<Void> started(WorkflowContextData workflowContext) {
-    return doTransaction(t -> t.writeInstanceData(workflowContext), workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> completed(WorkflowContextData workflowContext) {
-    return removeProcessInstance(workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> failed(WorkflowContextData workflowContext, Throwable ex) {
-    return removeProcessInstance(workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> aborted(WorkflowContextData workflowContext) {
-    return removeProcessInstance(workflowContext);
-  }
-
   protected CompletableFuture<Void> removeProcessInstance(WorkflowContextData workflowContext) {
-    return doTransaction(t -> t.removeProcessInstance(workflowContext), workflowContext)
+    return super.removeProcessInstance(workflowContext)
         .thenRun(() -> futuresMap.remove(workflowContext.instanceData().id()));
   }
 
   @Override
-  public CompletableFuture<Void> taskStarted(
-      WorkflowContextData workflowContext, TaskContextData taskContext) {
-    return CompletableFuture.completedFuture(null);
-  }
-
-  @Override
-  public CompletableFuture<Void> taskRetried(
-      WorkflowContextData workflowContext, TaskContextData taskContext) {
-    return doTransaction(t -> t.writeRetryTask(workflowContext, taskContext), workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> taskCompleted(
-      WorkflowContextData workflowContext, TaskContextData taskContext) {
-    return doTransaction(t -> t.writeCompletedTask(workflowContext, taskContext), workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> suspended(WorkflowContextData workflowContext) {
-    return doTransaction(
-        t -> t.writeStatus(workflowContext, WorkflowStatus.SUSPENDED), workflowContext);
-  }
-
-  @Override
-  public CompletableFuture<Void> resumed(WorkflowContextData workflowContext) {
-    return doTransaction(t -> t.clearStatus(workflowContext), workflowContext);
-  }
-
-  private CompletableFuture<Void> doTransaction(
-      Consumer<PersistenceInstanceTransaction> operation, WorkflowContextData context) {
+  protected CompletableFuture<Void> doTransaction(
+      Consumer<PersistenceInstanceOperations> operation, WorkflowContextData context) {
     final ExecutorService service =
         this.executorService.orElse(context.definition().application().executorService());
     final Runnable runnable = () -> executeTransaction(operation, context.definition());
@@ -112,28 +67,23 @@ private CompletableFuture<Void> doTransaction(
   }
 
   private void executeTransaction(
-      Consumer<PersistenceInstanceTransaction> operation, WorkflowDefinitionData definition) {
+      Consumer<PersistenceInstanceOperations> operation, WorkflowDefinitionData definition) {
     PersistenceInstanceTransaction transaction = store.begin();
     try {
       operation.accept(transaction);
       transaction.commit(definition);
     } catch (Exception ex) {
-      transaction.rollback(definition);
+      try {
+        transaction.rollback(definition);
+      } catch (Exception rollEx) {
+        logger.warn("Exception during rollback. Ignoring it", ex);
+      }
       throw ex;
     }
   }
 
   @Override
   public void close() {
     futuresMap.clear();
-    executorService.ifPresent(
-        e -> {
-          try {
-            e.awaitTermination(closeTimeout.toMillis(), TimeUnit.MILLISECONDS);
-            e.shutdown();
-          } catch (InterruptedException ex) {
-            Thread.currentThread().interrupt();
-          }
-        });
   }
 }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceOperations.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * 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 io.serverlessworkflow.impl.persistence;
+
+import io.serverlessworkflow.impl.TaskContextData;
+import io.serverlessworkflow.impl.WorkflowContextData;
+import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowStatus;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+public interface PersistenceInstanceOperations {
+  void writeInstanceData(WorkflowContextData workflowContext);
+
+  void writeRetryTask(WorkflowContextData workflowContext, TaskContextData taskContext);
+
+  void writeCompletedTask(WorkflowContextData workflowContext, TaskContextData taskContext);
+
+  void writeStatus(WorkflowContextData workflowContext, WorkflowStatus suspended);
+
+  void removeProcessInstance(WorkflowContextData workflowContext);
+
+  void clearStatus(WorkflowContextData workflowContext);
+
+  Stream<PersistenceWorkflowInfo> scanAll(String applicationId, WorkflowDefinition definition);
+
+  Optional<PersistenceWorkflowInfo> readWorkflowInfo(
+      WorkflowDefinition definition, String instanceId);
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceTransaction.java

@@ -15,34 +15,11 @@
  */
 package io.serverlessworkflow.impl.persistence;
 
-import io.serverlessworkflow.impl.TaskContextData;
-import io.serverlessworkflow.impl.WorkflowContextData;
-import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowDefinitionData;
-import io.serverlessworkflow.impl.WorkflowStatus;
-import java.util.Optional;
-import java.util.stream.Stream;
 
-public interface PersistenceInstanceTransaction {
+public interface PersistenceInstanceTransaction extends PersistenceInstanceOperations {
 
   void commit(WorkflowDefinitionData definition);
 
   void rollback(WorkflowDefinitionData definition);
-
-  void writeInstanceData(WorkflowContextData workflowContext);
-
-  void writeRetryTask(WorkflowContextData workflowContext, TaskContextData taskContext);
-
-  void writeCompletedTask(WorkflowContextData workflowContext, TaskContextData taskContext);
-
-  void writeStatus(WorkflowContextData workflowContext, WorkflowStatus suspended);
-
-  void removeProcessInstance(WorkflowContextData workflowContext);
-
-  void clearStatus(WorkflowContextData workflowContext);
-
-  Stream<PersistenceWorkflowInfo> scanAll(String applicationId, WorkflowDefinition definition);
-
-  Optional<PersistenceWorkflowInfo> readWorkflowInfo(
-      WorkflowDefinition definition, String instanceId);
 }