Added ability to simulate activity timeouts in unit tests (#138)

* reduced visibility of DecisionContextImpl

* Added ability to simulate activity timeouts in unit tests

* Fixed serialization of DataConverter field needed for ActivityTimeoutException.

Author: mfateev

src/main/java/com/uber/cadence/converter/JsonDataConverter.java

@@ -74,6 +74,8 @@ public final class JsonDataConverter implements DataConverter {
   private static final DataConverter INSTANCE = new JsonDataConverter();
   private static final byte[] EMPTY_BLOB = new byte[0];
   private static final Object[] EMPTY_OBJECT_ARRAY = new Object[0];
+  public static final String TYPE_FIELD_NAME = "type";
+  public static final String JSON_CONVERTER_TYPE = "JSON";
   private final Gson gson;
   private final JsonParser parser = new JsonParser();
 
@@ -165,6 +167,34 @@ public Object[] fromDataArray(byte[] content, Class<?>... valueType)
   private static class ThrowableTypeAdapterFactory implements TypeAdapterFactory {
     @Override
     public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> typeToken) {
+      // Special handling of fields of DataConverter type.
+      // Needed to serialize exceptions like ActivityTimeoutException.
+      if (DataConverter.class.isAssignableFrom(typeToken.getRawType())) {
+        return new TypeAdapter<T>() {
+          @Override
+          public void write(JsonWriter out, T value) throws IOException {
+            out.beginObject();
+            out.name(TYPE_FIELD_NAME).value(JSON_CONVERTER_TYPE);
+            out.endObject();
+          }
+
+          @Override
+          @SuppressWarnings("unchecked")
+          public T read(JsonReader in) throws IOException {
+            in.beginObject();
+            if (!in.nextName().equals("type")) {
+              throw new IOException("Cannot deserialize DataConverter. Missing type field");
+            }
+            String value = in.nextString();
+            if (!"JSON".equals(value)) {
+              throw new IOException(
+                  "Cannot deserialize DataConverter. Expected type is JSON. " + "Found " + value);
+            }
+            in.endObject();
+            return (T) JsonDataConverter.getInstance();
+          }
+        };
+      }
       if (!Throwable.class.isAssignableFrom(typeToken.getRawType())) {
         return null; // this class only serializes 'Throwable' and its subtypes
       }

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

@@ -30,7 +30,7 @@
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
-public final class DecisionContextImpl implements DecisionContext, HistoryEventHandler {
+final class DecisionContextImpl implements DecisionContext, HistoryEventHandler {
 
   private final ActivityDecisionContext activityClient;
 

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

@@ -527,8 +527,8 @@ private DecisionStateMachine getDecision(DecisionId decisionId) {
       throw new IllegalArgumentException(
           "Unknown "
               + decisionId
-              + ". The possible causes are "
-              + "nondeterministic workflow definition code or incompatible change in the workflow definition.");
+              + ". The possible causes are a nondeterministic workflow definition code or an incompatible change in the workflow definition."
+              + "See the \"Workflow Implementation Constraints\" section from the github.com/uber-java/cadence-client README");
     }
     return result;
   }

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

@@ -31,6 +31,7 @@
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.worker.ActivityTaskHandler;
 import com.uber.cadence.serviceclient.IWorkflowService;
+import com.uber.cadence.testing.TestActivityTimeoutException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
@@ -103,8 +104,16 @@ private ActivityTaskHandler.Result mapToActivityFailure(ActivityTask task, Throw
     if (failure instanceof ActivityCancelledException) {
       throw new CancellationException(failure.getMessage());
     }
+    // Only expected during unit tests.
+    if (failure instanceof TestActivityTimeoutException) {
+      TestActivityTimeoutException timeoutException = (TestActivityTimeoutException) failure;
+      failure =
+          new TestActivityTimeoutExceptionInternal(
+              timeoutException.getTimeoutType(),
+              dataConverter.toData(timeoutException.getDetails()));
+    }
     RespondActivityTaskFailedRequest result = new RespondActivityTaskFailedRequest();
-    failure = CheckedExceptionWrapper.unwrap((Exception) failure);
+    failure = CheckedExceptionWrapper.unwrap(failure);
     result.setReason(failure.getClass().getName());
     result.setDetails(dataConverter.toData(failure));
     return new ActivityTaskHandler.Result(null, result, null, null);

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

@@ -180,6 +180,18 @@ private RuntimeException mapActivityException(Exception failure) {
       } catch (Exception e) {
         cause = e;
       }
+      if (cause instanceof TestActivityTimeoutExceptionInternal) {
+        // This exception is thrown only in unit tests to mock the activity timeouts
+        TestActivityTimeoutExceptionInternal testTimeout =
+            (TestActivityTimeoutExceptionInternal) cause;
+        return new ActivityTimeoutException(
+            taskFailed.getEventId(),
+            taskFailed.getActivityType(),
+            taskFailed.getActivityId(),
+            testTimeout.getTimeoutType(),
+            testTimeout.getDetails(),
+            getDataConverter());
+      }
       return new ActivityFailureException(
           taskFailed.getEventId(), taskFailed.getActivityType(), taskFailed.getActivityId(), cause);
     }

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

@@ -0,0 +1,49 @@
+/*
+ *  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.internal.sync;
+
+import com.uber.cadence.TimeoutType;
+
+/**
+ * TestActivityTimeoutException is created from an TestActivityTimeoutException. The main difference
+ * is that details are in a serialized form.
+ */
+final class TestActivityTimeoutExceptionInternal extends RuntimeException {
+
+  private final TimeoutType timeoutType;
+
+  private final byte[] details;
+
+  TestActivityTimeoutExceptionInternal(TimeoutType timeoutType, byte[] details) {
+    this.timeoutType = timeoutType;
+    this.details = details;
+  }
+
+  TestActivityTimeoutExceptionInternal(TimeoutType timeoutType) {
+    this.timeoutType = timeoutType;
+    this.details = null;
+  }
+
+  TimeoutType getTimeoutType() {
+    return timeoutType;
+  }
+
+  byte[] getDetails() {
+    return details;
+  }
+}

src/main/java/com/uber/cadence/testing/TestActivityTimeoutException.java

@@ -0,0 +1,50 @@
+/*
+ *  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.testing;
+
+import com.uber.cadence.TimeoutType;
+
+/**
+ * TestActivityTimeoutException can be thrown from an activity implementation to simulate an
+ * activity timeout. To be used only in unit tests. The workflow code is going to receive it as
+ * {@link com.uber.cadence.workflow.ActivityTimeoutException}.
+ */
+public final class TestActivityTimeoutException extends RuntimeException {
+
+  private final TimeoutType timeoutType;
+
+  private final Object details;
+
+  public TestActivityTimeoutException(TimeoutType timeoutType, Object details) {
+    this.timeoutType = timeoutType;
+    this.details = details;
+  }
+
+  public TestActivityTimeoutException(TimeoutType timeoutType) {
+    this.timeoutType = timeoutType;
+    this.details = null;
+  }
+
+  public TimeoutType getTimeoutType() {
+    return timeoutType;
+  }
+
+  public Object getDetails() {
+    return details;
+  }
+}

src/main/java/com/uber/cadence/workflow/ActivityTimeoutException.java

@@ -20,19 +20,20 @@
 import com.uber.cadence.ActivityType;
 import com.uber.cadence.TimeoutType;
 import com.uber.cadence.converter.DataConverter;
+import java.util.Objects;
 
 /**
- * Exception that indicates Activity that activity timed out. If timeout type is {@link
- * TimeoutType#HEARTBEAT} then {@link #getDetails(Class)} returns the value passed to the latest
- * successful {@link com.uber.cadence.activity.Activity#heartbeat(Object)} call.
+ * ActivityTimeoutException indicates that an activity has timed out. If the timeout type is a
+ * {@link TimeoutType#HEARTBEAT} then the {@link #getDetails(Class)} returns a value passed to the
+ * latest successful {@link com.uber.cadence.activity.Activity#heartbeat(Object)} call.
  */
 @SuppressWarnings("serial")
 public final class ActivityTimeoutException extends ActivityException {
 
   private final TimeoutType timeoutType;
 
   private final byte[] details;
-  private final transient DataConverter dataConverter;
+  private final DataConverter dataConverter;
 
   public ActivityTimeoutException(
       long eventId,
@@ -42,9 +43,9 @@ public ActivityTimeoutException(
       byte[] details,
       DataConverter dataConverter) {
     super("TimeoutType=" + String.valueOf(timeoutType), eventId, activityType, activityId);
-    this.timeoutType = timeoutType;
+    this.timeoutType = Objects.requireNonNull(timeoutType);
     this.details = details;
-    this.dataConverter = dataConverter;
+    this.dataConverter = Objects.requireNonNull(dataConverter);
   }
 
   public TimeoutType getTimeoutType() {

src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java

@@ -36,6 +36,7 @@
 import com.uber.cadence.client.WorkflowStub;
 import com.uber.cadence.client.WorkflowTimedOutException;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
+import com.uber.cadence.testing.TestActivityTimeoutException;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
 import com.uber.cadence.worker.Worker;
 import com.uber.cadence.workflow.ActivityTimeoutException;
@@ -200,6 +201,33 @@ public void testActivityFailure() {
     }
   }
 
+  private static class SimulatedTimeoutActivityImpl implements TestActivity {
+
+    @Override
+    public String activity1(String input) {
+      throw new TestActivityTimeoutException(TimeoutType.HEARTBEAT, "progress1");
+    }
+  }
+
+  @Test
+  public void testActivitySimulatedTimeout() {
+    Worker worker = testEnvironment.newWorker(TASK_LIST);
+    worker.registerWorkflowImplementationTypes(ActivityWorkflow.class);
+    worker.registerActivitiesImplementations(new SimulatedTimeoutActivityImpl());
+    worker.start();
+    WorkflowClient client = testEnvironment.newWorkflowClient();
+    TestWorkflow workflow = client.newWorkflowStub(TestWorkflow.class);
+    try {
+      workflow.workflow1("input1");
+      fail("unreacheable");
+    } catch (WorkflowException e) {
+      assertTrue(e.getCause() instanceof ActivityTimeoutException);
+      ActivityTimeoutException te = (ActivityTimeoutException) e.getCause();
+      assertEquals(TimeoutType.HEARTBEAT, te.getTimeoutType());
+      assertEquals("progress1", te.getDetails(String.class));
+    }
+  }
+
   public interface TestActivityTimeoutWorkflow {
 
     @WorkflowMethod(executionStartToCloseTimeoutSeconds = 3600 * 24, taskList = TASK_LIST)