/*
 *  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.workflow;

import static com.uber.cadence.testUtils.TestEnvironment.DOMAIN;
import static com.uber.cadence.worker.NonDeterministicWorkflowPolicy.FailWorkflow;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeTrue;

import com.google.common.base.Strings;
import com.google.common.util.concurrent.UncheckedExecutionException;
import com.uber.cadence.BadRequestError;
import com.uber.cadence.CancellationAlreadyRequestedError;
import com.uber.cadence.DomainAlreadyExistsError;
import com.uber.cadence.DomainNotActiveError;
import com.uber.cadence.EntityNotExistsError;
import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
import com.uber.cadence.HistoryEvent;
import com.uber.cadence.Memo;
import com.uber.cadence.QueryConsistencyLevel;
import com.uber.cadence.QueryFailedError;
import com.uber.cadence.QueryRejectCondition;
import com.uber.cadence.SearchAttributes;
import com.uber.cadence.SignalExternalWorkflowExecutionFailedCause;
import com.uber.cadence.TimeoutType;
import com.uber.cadence.WorkflowExecution;
import com.uber.cadence.WorkflowExecutionAlreadyCompletedError;
import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
import com.uber.cadence.WorkflowExecutionCloseStatus;
import com.uber.cadence.WorkflowIdReusePolicy;
import com.uber.cadence.activity.Activity;
import com.uber.cadence.activity.ActivityMethod;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.activity.ActivityTask;
import com.uber.cadence.activity.LocalActivityOptions;
import com.uber.cadence.client.*;
import com.uber.cadence.common.CronSchedule;
import com.uber.cadence.common.MethodRetry;
import com.uber.cadence.common.RetryOptions;
import com.uber.cadence.converter.JsonDataConverter;
import com.uber.cadence.internal.common.WorkflowExecutionUtils;
import com.uber.cadence.internal.sync.DeterministicRunnerTest;
import com.uber.cadence.testUtils.CadenceTestRule;
import com.uber.cadence.testUtils.RequiresDockerService;
import com.uber.cadence.testUtils.RequiresTestService;
import com.uber.cadence.testUtils.TestEnvironment;
import com.uber.cadence.worker.*;
import com.uber.cadence.workflow.interceptors.TracingWorkflowInterceptorFactory;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestName;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@RunWith(Parameterized.class)
public class WorkflowTest {

  private static final Logger log = LoggerFactory.getLogger(WorkflowTest.class);

  private static final String ANNOTATION_TASK_LIST = "WorkflowTest-testExecute[Docker]";
  private static final String UUID_REGEXP =
      "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
  private static final boolean stickyOff = Boolean.parseBoolean(System.getenv("STICKY_OFF"));

  @Rule public final CadenceTestRule cadenceTestRule;
  @Rule public final TestName testName = new TestName();

  @Parameters(name = "{0}")
  public static Object[] data() {
    if (TestEnvironment.isUseDockerService()) {
      return new Object[][] {
        {"Docker Sticky " + (stickyOff ? "OFF" : "ON"), stickyOff},
      };
    } else {
      return new Object[][] {{"TestService Sticky OFF", true}, {"TestService Sticky ON", false}};
    }
  }

  private final boolean disableStickyExecution;
  private String taskList;
  private TestActivitiesImpl activitiesImpl;
  private WorkflowClient workflowClient;
  private TracingWorkflowInterceptorFactory tracer;

  public WorkflowTest(String ignored, boolean disableStickyExecution) {
    this.disableStickyExecution = disableStickyExecution;
    this.cadenceTestRule =
        CadenceTestRule.builder()
            .withWorkerFactoryOptions(
                WorkerFactoryOptions.newBuilder()
                    .setDisableStickyExecution(disableStickyExecution)
                    .build())
            .build();
  }

  private static WorkflowOptions.Builder newWorkflowOptionsBuilder(String taskList) {
    if (TestEnvironment.isDebuggerTimeouts()) {
      return new WorkflowOptions.Builder()
          .setExecutionStartToCloseTimeout(Duration.ofSeconds(1000))
          .setTaskStartToCloseTimeout(Duration.ofSeconds(60))
          .setTaskList(taskList);
    } else {
      return new WorkflowOptions.Builder()
          .setExecutionStartToCloseTimeout(Duration.ofSeconds(30))
          .setTaskStartToCloseTimeout(Duration.ofSeconds(5))
          .setTaskList(taskList);
    }
  }

  private static ActivityOptions newActivityOptions1(String taskList) {
    if (TestEnvironment.isDebuggerTimeouts()) {
      return new ActivityOptions.Builder()
          .setTaskList(taskList)
          .setScheduleToCloseTimeout(Duration.ofSeconds(1000))
          .setHeartbeatTimeout(Duration.ofSeconds(1000))
          .setScheduleToStartTimeout(Duration.ofSeconds(1000))
          .setStartToCloseTimeout(Duration.ofSeconds(10000))
          .build();
    } else {
      return new ActivityOptions.Builder()
          .setTaskList(taskList)
          .setScheduleToCloseTimeout(Duration.ofSeconds(5))
          .setHeartbeatTimeout(Duration.ofSeconds(5))
          .setScheduleToStartTimeout(Duration.ofSeconds(5))
          .setStartToCloseTimeout(Duration.ofSeconds(10))
          .build();
    }
  }

  private static LocalActivityOptions newLocalActivityOptions1() {
    if (TestEnvironment.isDebuggerTimeouts()) {
      return new LocalActivityOptions.Builder()
          .setScheduleToCloseTimeout(Duration.ofSeconds(1000))
          .build();
    } else {
      return new LocalActivityOptions.Builder()
          .setScheduleToCloseTimeout(Duration.ofSeconds(5))
          .build();
    }
  }

  private static ActivityOptions newActivityOptions2() {
    return new ActivityOptions.Builder().setScheduleToCloseTimeout(Duration.ofSeconds(20)).build();
  }

  @Before
  public void setUp() {
    String testMethod = testName.getMethodName();
    if (testMethod.startsWith("testExecute") || testMethod.startsWith("testStart")) {
      taskList = ANNOTATION_TASK_LIST;
    } else {
      taskList = cadenceTestRule.getDefaultTaskList();
    }

    workflowClient = cadenceTestRule.getWorkflowClient();
    tracer = cadenceTestRule.getTracer();
    ActivityCompletionClient completionClient = workflowClient.newActivityCompletionClient();
    activitiesImpl = new TestActivitiesImpl(completionClient);
    cadenceTestRule.getWorker(taskList).registerActivitiesImplementations(activitiesImpl);

    activitiesImpl.invocations.clear();
    activitiesImpl.procResult.clear();
  }

  @After
  public void tearDown() throws Throwable {
    if (activitiesImpl != null) {
      activitiesImpl.close();
    }
    cadenceTestRule.stop();
  }

  private void startWorkerFor(Class<?>... workflowTypes) {
    Worker worker = cadenceTestRule.getWorker(taskList);
    worker.registerWorkflowImplementationTypes(workflowTypes);
    cadenceTestRule.start();
  }

  private void startWorkerFor(WorkflowImplementationOptions options, Class<?>... workflowTypes) {
    Worker worker = cadenceTestRule.getWorker(taskList);
    worker.registerWorkflowImplementationTypes(options, workflowTypes);
    cadenceTestRule.start();
  }

  // TODO: Refactor testEnvironment to support testing through real service to avoid this
  // conditional switches
  void registerDelayedCallback(Duration delay, Runnable r) {
    cadenceTestRule.registerDelayedCallback(delay, r);
  }

  void sleep(Duration d) {
    cadenceTestRule.sleep(d);
  }

  long currentTimeMillis() {
    return cadenceTestRule.currentTimeMillis();
  }

  public interface TestWorkflow1 {

    @WorkflowMethod
    String execute(String taskList);
  }

  public interface TestWorkflowSignaled {

    @WorkflowMethod
    String execute();

    @SignalMethod(name = "testSignal")
    void signal1(String arg);
  }

  public interface TestWorkflow2 {

    @WorkflowMethod(name = "testActivity")
    String execute(boolean useExternalService);

    @QueryMethod(name = "getTrace")
    List<String> getTrace();
  }

  public interface TestWorkflow3 {

    @WorkflowMethod
    String execute(String taskList);

    @SignalMethod(name = "testSignal")
    void signal1(String arg);

    @QueryMethod(name = "getState")
    String getState();
  }

  public interface TestWorkflowQuery {

    @WorkflowMethod()
    String execute(String taskList);

    @QueryMethod()
    String query();
  }

  public static class TestWorkflowSignaledSimple implements TestWorkflowSignaled {

    List<String> messageQueue = new ArrayList<>(10);

    @Override
    @WorkflowMethod
    public String execute() {
      Workflow.await(() -> !messageQueue.isEmpty());
      messageQueue.remove(0);
      return "Simple workflow signaled";
    }

    @Override
    @SignalMethod
    public void signal1(String arg) {
      messageQueue.add("arg");
    }
  }

  public static class TestWorkflowActivityOptionOverride implements TestWorkflow1 {

    @Override
    public String execute(String taskList) {
      TestActivities activities = Workflow.newActivityStub(TestActivities.class);

      return activities.timeOutActivity();
    }
  }

  public static class TestSyncWorkflowImpl implements TestWorkflow1 {

    @Override
    public String execute(String taskList) {
      TestActivities activities =
          Workflow.newActivityStub(TestActivities.class, newActivityOptions1(taskList));
      // Invoke synchronously in a separate thread for testing purposes only.
      // In real workflows use
      // Async.procedure(activities::activityWithDelay, 1000, true)
      Promise<String> a1 = Async.function(() -> activities.activityWithDelay(1000, true));
      Workflow.sleep(2000);
      return activities.activity2(a1.get(), 10);
    }
  }

  @Test
  public void testSync() {
    startWorkerFor(TestSyncWorkflowImpl.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    String result = workflowStub.execute(taskList);
    assertEquals("activity10", result);
    tracer.setExpected(
        "executeWorkflow: TestWorkflow1::execute",
        "sleep PT2S",
        "executeActivity TestActivities::activityWithDelay",
        "executeActivity TestActivities::activity2");
  }

  public interface TestMultipleTimers {
    @WorkflowMethod
    long execute();
  }

  public static class TestMultipleTimersImpl implements TestMultipleTimers {

    @Override
    public long execute() {
      Promise<Void> t1 = Async.procedure(() -> Workflow.sleep(Duration.ofSeconds(1)));
      Promise<Void> t2 = Async.procedure(() -> Workflow.sleep(Duration.ofSeconds(2)));
      long start = Workflow.currentTimeMillis();
      Promise.anyOf(t1, t2).get();
      long elapsed = Workflow.currentTimeMillis() - start;
      return elapsed;
    }
  }

  @Test
  public void testMultipleTimers() {
    startWorkerFor(TestMultipleTimersImpl.class);
    TestMultipleTimers workflowStub =
        workflowClient.newWorkflowStub(
            TestMultipleTimers.class, newWorkflowOptionsBuilder(taskList).build());
    long result = workflowStub.execute();
    assertTrue("should be around 1 second: " + result, result < 2000);
  }

  public static class TestActivityRetryWithMaxAttempts implements TestWorkflow1 {

    @Override
    @SuppressWarnings("Finally")
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(3))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setMaximumInterval(Duration.ofSeconds(1))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setMaximumAttempts(3)
                      .setDoNotRetry(AssertionError.class)
                      .build())
              .build();
      TestActivities activities = Workflow.newActivityStub(TestActivities.class, options);
      long start = Workflow.currentTimeMillis();
      try {
        activities.heartbeatAndThrowIO();
      } finally {
        if (Workflow.currentTimeMillis() - start < 2000) {
          fail("Activity retried without delay");
        }
      }
      return "ignored";
    }
  }

  @Test
  public void testActivityRetryWithMaxAttempts() {
    startWorkerFor(TestActivityRetryWithMaxAttempts.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
  }

  public static class TestActivityRetryWithExpiration implements TestWorkflow1 {

    @Override
    @SuppressWarnings("Finally")
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(3))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setExpiration(Duration.ofSeconds(3))
                      .setMaximumInterval(Duration.ofSeconds(1))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setDoNotRetry(AssertionError.class)
                      .build())
              .build();
      TestActivities activities = Workflow.newActivityStub(TestActivities.class, options);
      long start = Workflow.currentTimeMillis();
      try {
        activities.heartbeatAndThrowIO();
      } finally {
        if (Workflow.currentTimeMillis() - start < 2000) {
          fail("Activity retried without delay");
        }
      }
      return "ignored";
    }
  }

  @Test
  public void testActivityRetryWithExpiration() {
    startWorkerFor(TestActivityRetryWithExpiration.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
  }

  public static class TestLocalActivityRetry implements TestWorkflow1 {

    @Override
    @SuppressWarnings("Finally")
    public String execute(String taskList) {
      LocalActivityOptions options =
          new LocalActivityOptions.Builder()
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setExpiration(Duration.ofSeconds(100))
                      .setMaximumInterval(Duration.ofSeconds(20))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setMaximumAttempts(5)
                      .setDoNotRetry(AssertionError.class)
                      .build())
              .build();
      TestActivities activities = Workflow.newLocalActivityStub(TestActivities.class, options);
      activities.throwIO();

      return "ignored";
    }
  }

  @Test
  public void testLocalActivityRetry() {
    startWorkerFor(TestLocalActivityRetry.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class,
            newWorkflowOptionsBuilder(taskList)
                .setTaskStartToCloseTimeout(Duration.ofSeconds(5))
                .build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 5, activitiesImpl.invocations.size());
    assertEquals("last attempt", 5, activitiesImpl.getLastAttempt());
  }

  public static class TestActivityRetryOnTimeout implements TestWorkflow1 {

    @Override
    @SuppressWarnings("Finally")
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setScheduleToCloseTimeout(Duration.ofSeconds(1))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setExpiration(Duration.ofSeconds(100))
                      .setMaximumInterval(Duration.ofSeconds(1))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setMaximumAttempts(3)
                      .setDoNotRetry(AssertionError.class)
                      .build())
              .build();
      TestActivities activities = Workflow.newActivityStub(TestActivities.class, options);
      long start = Workflow.currentTimeMillis();
      try {
        activities.neverComplete(); // should timeout as scheduleToClose is 1 second
        throw new IllegalStateException("unreachable");
      } catch (ActivityTimeoutException e) {
        long elapsed = Workflow.currentTimeMillis() - start;
        if (elapsed < 5000) {
          throw new RuntimeException("Activity retried without delay: " + elapsed);
        }
        throw e;
      }
    }
  }

  @Test
  public void testActivityRetryOnTimeout() {
    startWorkerFor(TestActivityRetryOnTimeout.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    // Wall time on purpose
    long start = System.currentTimeMillis();
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause() instanceof ActivityTimeoutException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
    long elapsed = System.currentTimeMillis() - start;
    if (testName.toString().contains("TestService")) {
      assertTrue("retry timer skips time", elapsed < 5000);
    }
  }

  public static class TestActivityRetryOptionsChange implements TestWorkflow1 {

    @Override
    public String execute(String taskList) {
      ActivityOptions.Builder options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .setScheduleToStartTimeout(Duration.ofSeconds(5))
              .setStartToCloseTimeout(Duration.ofSeconds(10));
      RetryOptions retryOptions;
      if (Workflow.isReplaying()) {
        retryOptions =
            new RetryOptions.Builder()
                .setMaximumInterval(Duration.ofSeconds(1))
                .setInitialInterval(Duration.ofSeconds(1))
                .setExpiration(Duration.ofDays(1))
                .setMaximumAttempts(3)
                .build();
      } else {
        retryOptions =
            new RetryOptions.Builder()
                .setMaximumInterval(Duration.ofSeconds(1))
                .setInitialInterval(Duration.ofSeconds(1))
                .setExpiration(Duration.ofDays(1))
                .setMaximumAttempts(2)
                .build();
      }
      TestActivities activities = Workflow.newActivityStub(TestActivities.class, options.build());
      Workflow.retry(retryOptions, () -> activities.throwIO());
      return "ignored";
    }
  }

  @Test
  public void testActivityRetryOptionsChange() {
    startWorkerFor(TestActivityRetryOptionsChange.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 2, activitiesImpl.invocations.size());
  }

  public static class TestUntypedActivityRetry implements TestWorkflow1 {

    @Override
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .setScheduleToStartTimeout(Duration.ofSeconds(5))
              .setStartToCloseTimeout(Duration.ofSeconds(10))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setExpiration(Duration.ofSeconds(100))
                      .setMaximumInterval(Duration.ofSeconds(1))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setMaximumAttempts(3)
                      .build())
              .build();
      ActivityStub activities = Workflow.newUntypedActivityStub(options);
      activities.execute("TestActivities::throwIO", Void.class);
      return "ignored";
    }
  }

  @Test
  public void testUntypedActivityRetry() {
    startWorkerFor(TestUntypedActivityRetry.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
  }

  public static class TestActivityRetryAnnotated implements TestWorkflow1 {

    private final TestActivities activities;

    public TestActivityRetryAnnotated() {
      this.activities = Workflow.newActivityStub(TestActivities.class);
    }

    @Override
    public String execute(String taskList) {
      activities.throwIOAnnotated();
      return "ignored";
    }
  }

  @Test
  public void testActivityRetryAnnotated() {
    startWorkerFor(TestActivityRetryAnnotated.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
  }

  public static class TestAsyncActivityRetry implements TestWorkflow1 {

    private TestActivities activities;

    @Override
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .setScheduleToStartTimeout(Duration.ofSeconds(5))
              .setStartToCloseTimeout(Duration.ofSeconds(10))
              .setRetryOptions(
                  new RetryOptions.Builder()
                      .setExpiration(Duration.ofSeconds(100))
                      .setMaximumInterval(Duration.ofSeconds(1))
                      .setInitialInterval(Duration.ofSeconds(1))
                      .setMaximumAttempts(3)
                      .build())
              .build();
      this.activities = Workflow.newActivityStub(TestActivities.class, options);
      Async.procedure(activities::heartbeatAndThrowIO).get();
      return "ignored";
    }
  }

  @Test
  public void testAsyncActivityRetry() {
    startWorkerFor(TestAsyncActivityRetry.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 3, activitiesImpl.invocations.size());
  }

  public static class TestAsyncActivityRetryOptionsChange implements TestWorkflow1 {

    private TestActivities activities;

    @Override
    public String execute(String taskList) {
      ActivityOptions.Builder options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(5))
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .setScheduleToStartTimeout(Duration.ofSeconds(5))
              .setStartToCloseTimeout(Duration.ofSeconds(10));
      if (Workflow.isReplaying()) {
        options.setRetryOptions(
            new RetryOptions.Builder()
                .setExpiration(Duration.ofSeconds(100))
                .setMaximumInterval(Duration.ofSeconds(1))
                .setInitialInterval(Duration.ofSeconds(1))
                .setDoNotRetry(NullPointerException.class)
                .setMaximumAttempts(3)
                .build());
      } else {
        options.setRetryOptions(
            new RetryOptions.Builder()
                .setExpiration(Duration.ofSeconds(10))
                .setMaximumInterval(Duration.ofSeconds(1))
                .setInitialInterval(Duration.ofSeconds(1))
                .setMaximumAttempts(2)
                .setDoNotRetry(NullPointerException.class)
                .build());
      }
      this.activities = Workflow.newActivityStub(TestActivities.class, options.build());
      Async.procedure(activities::throwIO).get();
      return "ignored";
    }
  }

  @Test
  public void testAsyncActivityRetryOptionsChange() {
    startWorkerFor(TestAsyncActivityRetryOptionsChange.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    try {
      workflowStub.execute(taskList);
      fail("unreachable");
    } catch (WorkflowException e) {
      assertTrue(e.getCause().getCause() instanceof IOException);
    }
    assertEquals(activitiesImpl.toString(), 2, activitiesImpl.invocations.size());
  }

  public static class TestHeartbeatTimeoutDetails implements TestWorkflow1 {

    @Override
    public String execute(String taskList) {
      ActivityOptions options =
          new ActivityOptions.Builder()
              .setTaskList(taskList)
              .setHeartbeatTimeout(Duration.ofSeconds(1)) // short heartbeat timeout;
              .setScheduleToCloseTimeout(Duration.ofSeconds(5))
              .build();

      TestActivities activities = Workflow.newActivityStub(TestActivities.class, options);
      try {
        // false for second argument means to heartbeat once to set details and then stop.
        activities.activityWithDelay(5000, false);
      } catch (ActivityTimeoutException e) {
        assertEquals(TimeoutType.HEARTBEAT, e.getTimeoutType());
        return e.getDetails(String.class);
      }
      throw new RuntimeException("unreachable");
    }
  }

  @Test
  public void testHeartbeatTimeoutDetails() {
    startWorkerFor(TestHeartbeatTimeoutDetails.class);
    TestWorkflow1 workflowStub =
        workflowClient.newWorkflowStub(
            TestWorkflow1.class, newWorkflowOptionsBuilder(taskList).build());
    String result = workflowStub.execute(taskList);
    assertEquals("heartbeatValue", result);
  }

  @Test
  public void testSyncUntypedAndStackTrace() {
    startWorkerFor(TestSyncWorkflowImpl.class);
    WorkflowStub workflowStub =
        workflowClient.newUntypedWorkflowStub(
            "TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
    WorkflowExecution execution = workflowStub.start(taskList);
    testUntypedAndStackTraceHelper(workflowStub, execution);
  }

  @Test
  public void testUntypedAsyncStart() throws Exception {
    startWorkerFor(TestSyncWorkflowImpl.class);
    WorkflowStub workflowStub =
        workflowClient.newUntypedWorkflowStub(
            "TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
    CompletableFuture<WorkflowExecution> future = workflowStub.startAsync(taskList);
    testUntypedAndStackTraceHelper(workflowStub, future.get());
  }

  @Test
  public void testUntypedAsyncStartWithTimeout() throws Exception {
    startWorkerFor(TestSyncWorkflowImpl.class);
    WorkflowStub workflowStub =
        workflowClient.newUntypedWorkflowStub(
            "TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
    Long timeout = Long.valueOf(200);
    CompletableFuture<WorkflowExecution> future =
        workflowStub.startAsyncWithTimeout(timeout, TimeUnit.MILLISECONDS, taskList);
    testUntypedAndStackTraceHelper(workflowStub, future.get());
  }

  @Test
  public void testUntypedAsyncStartFailed() throws InterruptedException {
    startWorkerFor(TestSyncWorkflowImpl.class);
    String testWorkflowID = "test-untyped-async-failed";
    WorkflowStub workflowStub =
        workflowClient.newUntypedWorkflowStub(
            "TestWorkflow1::execute",
            newWorkflowOptionsBuilder(taskList).setWorkflowId(testWorkflowID).build());
    workflowStub.start(taskList);
    try {
      workflowStub.startAsync(taskList);
      fail("unreachable");
    } catch (DuplicateWorkflowException e) {
      // expected error from stub reuse
    }

    WorkflowStub newWorkflowStub =
        workflowClient.newUntypedWorkflowStub(
            "TestWorkflow1::execute",
            newWorkflowOptionsBuilder(taskList).setWorkflowId(testWorkflowID).build());
    CompletableFuture<WorkflowExecution> future = newWorkflowStub.startAsync(taskList);
    try {
      future.get();
    } catch (ExecutionException e) {
      assertTrue(e.getCause() instanceof WorkflowExecutionAlreadyStartedError);
    }
  }

  @Test
  public void testUntypedAsyncStartAndGetResult() throws InterruptedException {
    startWorkerFor(TestSyncWorkflowImpl.class);
    String wfType = "TestWorkflow1::execute";
    long timeoutMillis = 5000;
    long startTime = System.currentTimeMillis();
    WorkflowStub startStub =
        workflowClient.newUntypedWorkflowStub(wfType, newWorkflowOptionsBuilder(taskList).build());
