Added support for NonDeterministicWorkflowPolicy.FailWorkflow (#236)

* Added nonDeterministicWorkflowPolicy to WorkflowMethod

* Added WorkflowImplementationOptions

* Added support for NonDeterministicWorkflowPolicyFailWorkflow

Author: mfateev

src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.internal.replay;
 
+import static com.uber.cadence.worker.NonDeterministicWorkflowPolicy.FailWorkflow;
+
 import com.uber.cadence.Decision;
 import com.uber.cadence.EventType;
 import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
@@ -246,7 +248,7 @@ private void eventLoop() {
     try {
       completed = workflow.eventLoop();
     } catch (Error e) {
-      throw e; // errors fail decision, not a workflow
+      throw e;
     } catch (WorkflowExecutionException e) {
       failure = e;
       completed = true;
@@ -418,8 +420,17 @@ private void decideImpl(PollForDecisionTaskResponse decisionTask, Functions.Proc
         decisionsHelper.handleDecisionTaskStartedEvent(decision);
       }
     } catch (Error e) {
-      metricsScope.counter(MetricsType.DECISION_TASK_ERROR_COUNTER).inc(1);
-      throw e;
+      if (this.workflow.getWorkflowImplementationOptions().getNonDeterministicWorkflowPolicy()
+          == FailWorkflow) {
+        // fail workflow
+        failure = workflow.mapError(e);
+        completed = true;
+        completeWorkflow();
+      } else {
+        metricsScope.counter(MetricsType.DECISION_TASK_ERROR_COUNTER).inc(1);
+        // fail decision, not a workflow
+        throw e;
+      }
     } finally {
       if (query != null) {
         query.apply();

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -109,7 +109,6 @@ private Result handleDecisionTaskImpl(PollForDecisionTaskResponse decisionTask)
     if (decisionTask.isSetQuery()) {
       return processQuery(decisionTask);
     } else {
-
       return processDecision(decisionTask);
     }
   }

src/main/java/com/uber/cadence/internal/replay/ReplayWorkflow.java

@@ -20,6 +20,7 @@
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.WorkflowQuery;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
+import com.uber.cadence.worker.WorkflowImplementationOptions;
 
 public interface ReplayWorkflow {
 
@@ -60,4 +61,8 @@ public interface ReplayWorkflow {
    * @return Serialized failure
    */
   WorkflowExecutionException mapUnexpectedException(Exception failure);
+
+  WorkflowExecutionException mapError(Error failure);
+
+  WorkflowImplementationOptions getWorkflowImplementationOptions();
 }

src/main/java/com/uber/cadence/internal/sync/POJOWorkflowImplementationFactory.java

@@ -17,6 +17,8 @@
 
 package com.uber.cadence.internal.sync;
 
+import static com.uber.cadence.worker.NonDeterministicWorkflowPolicy.FailWorkflow;
+
 import com.google.common.reflect.TypeToken;
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.converter.DataConverter;
@@ -29,6 +31,7 @@
 import com.uber.cadence.internal.replay.ReplayWorkflowFactory;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import com.uber.cadence.testing.SimulatedTimeoutException;
+import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.Functions;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.QueryMethod;
@@ -62,6 +65,9 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
   private final Map<String, Functions.Func<SyncWorkflowDefinition>> workflowDefinitions =
       Collections.synchronizedMap(new HashMap<>());
 
+  private Map<String, WorkflowImplementationOptions> implementationOptions =
+      Collections.synchronizedMap(new HashMap<>());
+
   private final Map<Class<?>, Functions.Func<?>> workflowImplementationFactories =
       Collections.synchronizedMap(new HashMap<>());
 
@@ -79,19 +85,25 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
     this.cache = cache;
   }
 
-  void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {
+  void setWorkflowImplementationTypes(
+      WorkflowImplementationOptions options, Class<?>[] workflowImplementationTypes) {
     workflowDefinitions.clear();
     for (Class<?> type : workflowImplementationTypes) {
-      addWorkflowImplementationType(type);
+      addWorkflowImplementationType(options, type);
     }
   }
 
   <R> void addWorkflowImplementationFactory(Class<R> clazz, Functions.Func<R> factory) {
     workflowImplementationFactories.put(clazz, factory);
-    addWorkflowImplementationType(clazz);
+    WorkflowImplementationOptions unitTestingOptions =
+        new WorkflowImplementationOptions.Builder()
+            .setNonDeterministicWorkflowPolicy(FailWorkflow)
+            .build();
+    addWorkflowImplementationType(unitTestingOptions, clazz);
   }
 
-  private void addWorkflowImplementationType(Class<?> workflowImplementationClass) {
+  private void addWorkflowImplementationType(
+      WorkflowImplementationOptions options, Class<?> workflowImplementationClass) {
     TypeToken<?>.TypeSet interfaces =
         TypeToken.of(workflowImplementationClass).getTypes().interfaces();
     if (interfaces.isEmpty()) {
@@ -129,6 +141,7 @@ private void addWorkflowImplementationType(Class<?> workflowImplementationClass)
                 workflowName + " workflow type is already registered with the worker");
           }
           workflowDefinitions.put(workflowName, factory);
+          implementationOptions.put(workflowName, options);
           hasWorkflowMethod = true;
         }
         if (signalMethod != null) {
@@ -183,7 +196,9 @@ public void setDataConverter(DataConverter dataConverter) {
   @Override
   public ReplayWorkflow getWorkflow(WorkflowType workflowType) {
     SyncWorkflowDefinition workflow = getWorkflowDefinition(workflowType);
-    return new SyncWorkflow(workflow, dataConverter, threadPool, interceptorFactory, cache);
+    WorkflowImplementationOptions options = implementationOptions.get(workflowType.getName());
+    return new SyncWorkflow(
+        workflow, options, dataConverter, threadPool, interceptorFactory, cache);
   }
 
   @Override
@@ -238,6 +253,7 @@ public byte[] execute(byte[] input) throws CancellationException, WorkflowExecut
                   + context.getWorkflowType(),
               targetException);
         }
+        // Cast to Exception is safe as Error is handled above.
         throw mapToWorkflowExecutionException((Exception) targetException, dataConverter);
       }
     }
@@ -335,6 +351,11 @@ static WorkflowExecutionException mapToWorkflowExecutionException(
         failure.getClass().getName(), dataConverter.toData(failure));
   }
 
+  static WorkflowExecutionException mapError(Error failure, DataConverter dataConverter) {
+    return new WorkflowExecutionException(
+        failure.getClass().getName(), dataConverter.toData(failure));
+  }
+
   @Override
   public String toString() {
     return "POJOWorkflowImplementationFactory{"

src/main/java/com/uber/cadence/internal/sync/SyncWorkflow.java

@@ -27,6 +27,7 @@
 import com.uber.cadence.internal.replay.DecisionContext;
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
+import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
@@ -41,24 +42,35 @@ class SyncWorkflow implements ReplayWorkflow {
   private final DataConverter dataConverter;
   private final ExecutorService threadPool;
   private final SyncWorkflowDefinition workflow;
+  WorkflowImplementationOptions workflowImplementationOptions;
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
   private DeciderCache cache;
   private WorkflowRunnable workflowProc;
   private DeterministicRunner runner;
 
   public SyncWorkflow(
       SyncWorkflowDefinition workflow,
+      WorkflowImplementationOptions workflowImplementationOptions,
       DataConverter dataConverter,
       ExecutorService threadPool,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
       DeciderCache cache) {
     this.workflow = Objects.requireNonNull(workflow);
+    this.workflowImplementationOptions =
+        workflowImplementationOptions == null
+            ? new WorkflowImplementationOptions.Builder().build()
+            : workflowImplementationOptions;
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
     this.cache = cache;
   }
 
+  @Override
+  public WorkflowImplementationOptions getWorkflowImplementationOptions() {
+    return workflowImplementationOptions;
+  }
+
   @Override
   public void start(HistoryEvent event, DecisionContext context) {
     WorkflowType workflowType =
@@ -137,4 +149,9 @@ public WorkflowExecutionException mapUnexpectedException(Exception failure) {
     return POJOWorkflowImplementationFactory.mapToWorkflowExecutionException(
         failure, dataConverter);
   }
+
+  @Override
+  public WorkflowExecutionException mapError(Error failure) {
+    return POJOWorkflowImplementationFactory.mapError(failure, dataConverter);
+  }
 }

src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java

@@ -27,6 +27,7 @@
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
 import com.uber.cadence.internal.worker.WorkflowWorker;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.cadence.worker.WorkflowImplementationOptions;
 import com.uber.cadence.workflow.Functions.Func;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import java.lang.reflect.Type;
@@ -74,8 +75,9 @@ public SyncWorkflowWorker(
     worker = new WorkflowWorker(service, domain, taskList, this.options, taskHandler);
   }
 
-  public void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {
-    factory.setWorkflowImplementationTypes(workflowImplementationTypes);
+  public void setWorkflowImplementationTypes(
+      WorkflowImplementationOptions options, Class<?>[] workflowImplementationTypes) {
+    factory.setWorkflowImplementationTypes(options, workflowImplementationTypes);
   }
 
   public <R> void addWorkflowImplementationFactory(Class<R> clazz, Func<R> factory) {

src/main/java/com/uber/cadence/worker/NonDeterministicWorkflowPolicy.java

@@ -0,0 +1,33 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.worker;
+
+public enum NonDeterministicWorkflowPolicy {
+
+  /**
+   * Fails decision tasks, blocking workflow progress until problem is fixed usually by rollback.
+   */
+  BlockWorkflow,
+
+  /**
+   * Fails a workflow instance on the first non deterministic error. Useful when workflow doesn't
+   * have any important state (like cron) and is restarted automatically through {@link
+   * com.uber.cadence.common.RetryOptions}.
+   */
+  FailWorkflow,
+}

src/main/java/com/uber/cadence/worker/Worker.java

@@ -185,7 +185,31 @@ public void registerWorkflowImplementationTypes(Class<?>... workflowImplementati
         !started.get(),
         "registerWorkflowImplementationTypes is not allowed after worker has started");
 
-    workflowWorker.setWorkflowImplementationTypes(workflowImplementationClasses);
+    workflowWorker.setWorkflowImplementationTypes(
+        new WorkflowImplementationOptions.Builder().build(), workflowImplementationClasses);
+  }
+
+  /**
+   * Register workflow implementation classes with a worker. Overwrites previously registered types.
+   * A workflow implementation class must implement at least one interface with a method annotated
+   * with {@link WorkflowMethod}. That method becomes a workflow type that this worker supports.
+   *
+   * <p>Implementations that share a worker must implement different interfaces as a workflow type
+   * is identified by the workflow interface, not by the implementation.
+   *
+   * <p>The reason for registration accepting workflow class, but not the workflow instance is that
+   * workflows are stateful and a new instance is created for each workflow execution.
+   */
+  public void registerWorkflowImplementationTypes(
+      WorkflowImplementationOptions options, Class<?>... workflowImplementationClasses) {
+    Preconditions.checkState(
+        workflowWorker != null,
+        "registerWorkflowImplementationTypes is not allowed when disableWorkflowWorker is set in worker options");
+    Preconditions.checkState(
+        !started.get(),
+        "registerWorkflowImplementationTypes is not allowed after worker has started");
+
+    workflowWorker.setWorkflowImplementationTypes(options, workflowImplementationClasses);
   }
 
   /**

src/main/java/com/uber/cadence/worker/WorkflowImplementationOptions.java

@@ -0,0 +1,78 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.worker;
+
+import static com.uber.cadence.worker.NonDeterministicWorkflowPolicy.BlockWorkflow;
+
+import java.util.Objects;
+
+public final class WorkflowImplementationOptions {
+
+  public static final class Builder {
+
+    private NonDeterministicWorkflowPolicy nonDeterministicWorkflowPolicy = BlockWorkflow;
+
+    /**
+     * Optional: Sets how decision worker deals with non-deterministic history events (presumably
+     * arising from non-deterministic workflow definitions or non-backward compatible workflow
+     * definition changes). default: BlockWorkflow, which just logs error but reply nothing back to
+     * server.
+     */
+    public Builder setNonDeterministicWorkflowPolicy(
+        NonDeterministicWorkflowPolicy nonDeterministicWorkflowPolicy) {
+      this.nonDeterministicWorkflowPolicy = Objects.requireNonNull(nonDeterministicWorkflowPolicy);
+      return this;
+    }
+
+    public WorkflowImplementationOptions build() {
+      return new WorkflowImplementationOptions(nonDeterministicWorkflowPolicy);
+    }
+  }
+
+  private final NonDeterministicWorkflowPolicy nonDeterministicWorkflowPolicy;
+
+  public WorkflowImplementationOptions(
+      NonDeterministicWorkflowPolicy nonDeterministicWorkflowPolicy) {
+    this.nonDeterministicWorkflowPolicy = nonDeterministicWorkflowPolicy;
+  }
+
+  public NonDeterministicWorkflowPolicy getNonDeterministicWorkflowPolicy() {
+    return nonDeterministicWorkflowPolicy;
+  }
+
+  @Override
+  public String toString() {
+    return "WorkflowImplementationOptions{"
+        + "nonDeterministicWorkflowPolicy="
+        + nonDeterministicWorkflowPolicy
+        + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    WorkflowImplementationOptions that = (WorkflowImplementationOptions) o;
+    return nonDeterministicWorkflowPolicy == that.nonDeterministicWorkflowPolicy;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(nonDeterministicWorkflowPolicy);
+  }
+}