fix ConcurrentModificationException in DeterministicRunnerImpl (#194)

* fix ConcurrentModificationException in deterministic runner implementation

Author: halakaraki

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

@@ -282,7 +282,12 @@ public void close() {
         c.stop();
       }
       threads.clear();
-      for (Promise<?> f : failedPromises) {
+
+      // We cannot use an iterator to unregister failed Promises since f.get()
+      // will remove the promise directly from failedPromises. This causes an ConcurrentModificationException
+      // For this reason we will loop over a copy of failedPromises.
+      Set<Promise> failedPromisesLoop = new HashSet<>(failedPromises);
+      for (Promise f : failedPromisesLoop) {
         if (!f.isCompleted()) {
           throw new Error("expected failed");
         }

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -25,6 +25,7 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+import com.google.common.util.concurrent.UncheckedExecutionException;
 import com.uber.cadence.SignalExternalWorkflowExecutionFailedCause;
 import com.uber.cadence.TimeoutType;
 import com.uber.cadence.WorkflowExecution;
@@ -634,6 +635,44 @@ public void testSyncUntypedAndStackTrace() throws InterruptedException {
     assertEquals("activity10", result);
   }
 
+  public static class TestCancellationForWorkflowsWithFailedPromises implements TestWorkflow1 {
+
+    @Override
+    public String execute(String taskList) {
+      Promise<String> failedPromise =
+          Async.function(
+              () -> {
+                throw new UncheckedExecutionException(new Exception("Oh noo!"));
+              });
+      Promise<String> failedPromise2 =
+          Async.function(
+              () -> {
+                throw new UncheckedExecutionException(new Exception("Oh noo again!"));
+              });
+      Workflow.await(() -> false);
+      fail("unreachable");
+      return "done";
+    }
+  }
+
+  @Test
+  public void WorkflowsWithFailedPromisesCanBeCancelled() {
+    worker.registerWorkflowImplementationTypes(
+        TestCancellationForWorkflowsWithFailedPromises.class);
+    testEnvironment.start();
+    WorkflowStub client =
+        workflowClient.newUntypedWorkflowStub(
+            "TestWorkflow1::execute", newWorkflowOptionsBuilder(taskList).build());
+    client.start(taskList);
+    client.cancel();
+
+    try {
+      client.getResult(String.class);
+      fail("unreachable");
+    } catch (CancellationException ignored) {
+    }
+  }
+
   @Test
   public void testWorkflowCancellation() {
     startWorkerFor(TestSyncWorkflowImpl.class);