Release

.github/workflows/maven.yml

@@ -5,9 +5,9 @@ name: Java CI with Maven
 
 on:
   push:
-    branches: [ main ]
+    branches: [ main, release ]
   pull_request:
-    branches: [ main ]
+    branches: [ main, release ]
 
 jobs:
   build:
@@ -22,11 +22,6 @@ jobs:
           java-version: '8'
           distribution: 'adopt'
           cache: maven
-          server-id: ossrh
-          server-username: MAVEN_USERNAME
-          server-password: MAVEN_PASSWORD
-          gpg-private-key: ${{ secrets.MAVEN_GPG_PRIVATE_KEY }}
-          gpg-passphrase: MAVEN_GPG_PASSPHRASE
       - name: Build with Maven
         run: mvn -B install -DskipTests --file pom.xml
       - name: Checkstyle with Maven
@@ -35,10 +30,32 @@ jobs:
         run: mvn spotbugs:check --file pom.xml
       - name: Unit Test with Maven
         run: mvn test --file pom.xml
+
+  deploy:
+    if: github.ref == 'refs/heads/main'
+    needs:
+      - build
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up JDK 8
+        uses: actions/setup-java@v2
+        with:
+          java-version: '8'
+          distribution: 'adopt'
+          cache: maven
+          server-id: ossrh
+          server-username: MAVEN_USERNAME
+          server-password: MAVEN_PASSWORD
+          gpg-private-key: ${{ secrets.MAVEN_GPG_PRIVATE_KEY }}
+          gpg-passphrase: MAVEN_GPG_PASSPHRASE
+
       - name: Deploy with Maven
         run: mvn -B clean deploy -DskipTests -Pci-cd
+
         env:
           MAVEN_USERNAME: ${{ secrets.OSSRH_USERNAME }}
           MAVEN_PASSWORD: ${{ secrets.OSSRH_TOKEN }}
           MAVEN_GPG_PASSPHRASE: ${{ secrets.MAVEN_GPG_PASSPHRASE }}
 
+

pom.xml

@@ -9,19 +9,22 @@
     <url>http://maven.apache.org</url>
     <description>Workflow - A Distributed Dag Processing Library</description>
 
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-dependencies</artifactId>
+        <version>2.7.18</version>
+    </parent>
+
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <java.version>1.8</java.version>
         <maven.compiler.source>${java.version}</maven.compiler.source>
         <maven.compiler.target>${java.version}</maven.compiler.target>
-        <jackson.version>2.13.4</jackson.version>
         <jgrapht.version>1.3.1</jgrapht.version>
-        <lombok.version>1.18.24</lombok.version>
-        <slf4j.version>2.0.1</slf4j.version>
 
         <!-- TEST props -->
         <junit.version>4.13.2</junit.version>
-        <test.containers.version>1.17.3</test.containers.version>
+        <test.containers.version>1.19.3</test.containers.version>
 
         <!-- PlugIns version -->
         <maven-site-plugin>3.12.1</maven-site-plugin>
@@ -35,6 +38,8 @@
         <spotbugs.version>4.7.0</spotbugs.version>
         <checkstyle.version>9.3</checkstyle.version>
         <doxia.module.apt.version>2.0.0-M3</doxia.module.apt.version>
+
+        <!-- impl dependency versions -->
     </properties>
 
     <dependencies>
@@ -255,5 +260,8 @@
         <module>workflow-api</module>
         <module>workflow-jedis</module>
         <module>workflow-examples</module>
+        <module>workflow-jdbc</module>
+        <module>workflow-mongodb</module>
+        <module>workflow-inmemory</module>
     </modules>
 </project>

workflow-api/pom.xml

@@ -14,12 +14,10 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
-            <version>${jackson.version}</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>${jackson.version}</version>
         </dependency>
         <dependency>
             <groupId>org.jgrapht</groupId>

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/adapter/base/BaseAdapter.java

@@ -13,7 +13,12 @@
 public abstract class BaseAdapter implements Adapter {
 
   protected final String namespace;
+
+  /**
+   * delay generator can have use according to type of adapter.
+   * eg. poll delay for queue and scheduler.
+   *     heartbeat for persistence adapter.
+   */
   protected final PollDelayGenerator pollDelayGenerator;
-  protected final Serde serde;
 
 }

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/adapter/base/BasePersistenceAdapter.java

@@ -1,24 +1,43 @@
 package io.github.pavansharma36.workflow.api.adapter.base;
 
 import io.github.pavansharma36.workflow.api.adapter.PersistenceAdapter;
+import io.github.pavansharma36.workflow.api.adapter.WorkflowAdapter;
+import io.github.pavansharma36.workflow.api.model.ManagerInfo;
 import io.github.pavansharma36.workflow.api.serde.Deserializer;
 import io.github.pavansharma36.workflow.api.serde.Serde;
 import io.github.pavansharma36.workflow.api.serde.Serializer;
 import io.github.pavansharma36.workflow.api.util.PollDelayGenerator;
+import java.util.Date;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * Base class for all {@link PersistenceAdapter}.
  */
+@Slf4j
 public abstract class BasePersistenceAdapter extends BaseAdapter implements PersistenceAdapter {
 
-  protected final Serializer serializer;
-  protected final Deserializer deserializer;
-
   protected BasePersistenceAdapter(String namespace,
-                                PollDelayGenerator pollDelayGenerator,
-                                Serde serde) {
-    super(namespace, pollDelayGenerator, serde);
-    this.serializer = serde.serializer();
-    this.deserializer = serde.deserializer();
+                                PollDelayGenerator heartbeatDelayGenerator) {
+    super(namespace, heartbeatDelayGenerator);
+  }
+
+  @Override
+  public void maintenance(WorkflowAdapter adapter) {
+    List<ManagerInfo> managerInfos = getAllManagerInfos();
+    long minHeartbeatTimestamp = System.currentTimeMillis()
+        - (heartbeatDelayGenerator().delay(false).toMillis() * 15);
+    managerInfos.forEach(m -> {
+      if (m.getHeartbeatEpoch() < minHeartbeatTimestamp) {
+        log.info("WorkflowManager's heartbeat is not updated since {}, purging",
+            new Date(m.getHeartbeatEpoch()));
+        removeManagerInfo(m.getManagerId());
+      }
+    });
+  }
+
+  @Override
+  public PollDelayGenerator heartbeatDelayGenerator() {
+    return pollDelayGenerator;
   }
 }

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/adapter/builder/BasePersistenceAdapterBuilder.java

@@ -0,0 +1,13 @@
+package io.github.pavansharma36.workflow.api.adapter.builder;
+
+import io.github.pavansharma36.workflow.api.adapter.PersistenceAdapter;
+import io.github.pavansharma36.workflow.api.util.PollDelayGenerator;
+
+public abstract class BasePersistenceAdapterBuilder<T extends BasePersistenceAdapterBuilder<T>>
+    extends BaseAdapterBuilder<T, PersistenceAdapter> {
+
+  public T withHeartbeatDelayGenerator(PollDelayGenerator heartbeatDelayGenerator) {
+    this.pollDelayGenerator = heartbeatDelayGenerator;
+    return (T) this;
+  }
+}

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/adapter/builder/BaseScheduleAdapterBuilder.java

@@ -18,15 +18,15 @@ public abstract class BaseScheduleAdapterBuilder<S extends BaseScheduleAdapterBu
       Duration.ofHours(1L));
   protected Duration maxRunDuration = Duration.ofDays(7L);
 
-  public BaseScheduleAdapterBuilder<S> withMaintenanceDelayGenerator(
+  public S withMaintenanceDelayGenerator(
       @NonNull final PollDelayGenerator maintenanceDelayGenerator) {
     this.maintenanceDelayGenerator = maintenanceDelayGenerator;
-    return this;
+    return (S) this;
   }
 
-  public BaseScheduleAdapterBuilder<S> maxRunDuration(@NonNull Duration maxRunDuration) {
+  public S maxRunDuration(@NonNull Duration maxRunDuration) {
     this.maxRunDuration = maxRunDuration;
-    return this;
+    return (S) this;
   }
 
 }

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/adapter/builder/WorkflowAdapterBuilder.java

@@ -10,45 +10,66 @@
 /**
  * Base class for all adapter builder.
  */
-public class WorkflowAdapterBuilder<S extends WorkflowAdapterBuilder<S, S1, S2, S3>,
-    S1 extends BaseScheduleAdapterBuilder<S1>,
-    S2 extends BaseAdapterBuilder<S2, PersistenceAdapter>,
-    S3 extends BaseAdapterBuilder<S3, QueueAdapter>> {
+public class WorkflowAdapterBuilder {
 
-  protected S1 scheduleAdapterBuilder;
-  protected S2 persistenceAdapterBuilder;
-  protected S3 queueAdapterBuilder;
+  protected BaseScheduleAdapterBuilder<?> scheduleAdapterBuilder;
+  protected BasePersistenceAdapterBuilder<?> persistenceAdapterBuilder;
+  protected BaseAdapterBuilder<?, ? extends QueueAdapter> queueAdapterBuilder;
 
-  public WorkflowAdapterBuilder<S, S1, S2, S3> withQueuePollDelayGenerator(
+  public WorkflowAdapterBuilder withQueuePollDelayGenerator(
       final PollDelayGenerator pollDelayGenerator) {
     this.queueAdapterBuilder.withPollDelayGenerator(pollDelayGenerator);
     return this;
   }
 
-  public WorkflowAdapterBuilder<S, S1, S2, S3> withSchedulePollDelayGenerator(
+  public WorkflowAdapterBuilder withSchedulePollDelayGenerator(
       final PollDelayGenerator pollDelayGenerator) {
     this.scheduleAdapterBuilder.withPollDelayGenerator(pollDelayGenerator);
     return this;
   }
 
-  public WorkflowAdapterBuilder<S, S1, S2, S3> withMaintenancePollDelayGenerator(
+  public WorkflowAdapterBuilder withHeartbeatDelayGenerator(PollDelayGenerator heartbeatDelayGenerator) {
+    this.persistenceAdapterBuilder.withHeartbeatDelayGenerator(heartbeatDelayGenerator);
+    return this;
+  }
+
+  public WorkflowAdapterBuilder withMaintenancePollDelayGenerator(
       final PollDelayGenerator pollDelayGenerator
   ) {
     this.scheduleAdapterBuilder.withMaintenanceDelayGenerator(pollDelayGenerator);
     return this;
   }
 
-  public WorkflowAdapterBuilder<S, S1, S2, S3> withMaxRunDuration(Duration duration) {
+  public WorkflowAdapterBuilder withMaxRunDuration(Duration duration) {
     this.scheduleAdapterBuilder.maxRunDuration(duration);
     return this;
   }
 
+  public WorkflowAdapterBuilder withScheduleAdapterBuilder(BaseScheduleAdapterBuilder<?> scheduleAdapterBuilder) {
+    this.scheduleAdapterBuilder = scheduleAdapterBuilder;
+    return this;
+  }
+
+  public WorkflowAdapterBuilder withPersistenceAdapterBuilder(BasePersistenceAdapterBuilder<?> persistenceAdapterBuilder) {
+    this.persistenceAdapterBuilder = persistenceAdapterBuilder;
+    return this;
+  }
+
+  public WorkflowAdapterBuilder withQueueAdapterBuilder(BaseAdapterBuilder<?, ? extends QueueAdapter> queueAdapterBuilder) {
+    this.queueAdapterBuilder = queueAdapterBuilder;
+    return this;
+  }
+
   /**
    * build {@link WorkflowAdapter}.
    *
    * @return - instance of workflowadaper.
    */
   public WorkflowAdapter build() {
+    scheduleAdapterBuilder.validate();
+    queueAdapterBuilder.validate();
+    persistenceAdapterBuilder.validate();
+
     return new WorkflowAdapterImpl(scheduleAdapterBuilder.build(),
         queueAdapterBuilder.build(),
         persistenceAdapterBuilder.build());

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/bean/id/Id.java

@@ -1,5 +1,7 @@
 package io.github.pavansharma36.workflow.api.bean.id;
 
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.github.pavansharma36.workflow.api.serde.IdSerializer;
 import lombok.AllArgsConstructor;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -12,6 +14,7 @@
 @ToString
 @EqualsAndHashCode
 @AllArgsConstructor
+@JsonSerialize(using = IdSerializer.class)
 public class Id {
   /**
    * string representation of this id.

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/bean/id/ManagerId.java

@@ -9,4 +9,8 @@ public class ManagerId extends Id {
   public ManagerId() {
     super(Utils.random());
   }
+
+  public ManagerId(String id) {
+    super(id);
+  }
 }

workflow-api/src/main/java/io/github/pavansharma36/workflow/api/bean/task/TaskType.java

@@ -1,18 +1,22 @@
 package io.github.pavansharma36.workflow.api.bean.task;
 
 import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.NonNull;
+import lombok.Setter;
 import lombok.ToString;
 
 /**
  * Task type to differentiate queue and executor for submitted tasks.
  */
 @Getter
+@Setter
 @ToString
 @NoArgsConstructor
 @AllArgsConstructor
+@EqualsAndHashCode
 public class TaskType {
   private int version = 1;
   private @NonNull String type;