Skip to content

Commit

Permalink
Fix flakiness in ManualActivityCompletionWorkflowTest (#937)
Browse files Browse the repository at this point in the history
The operations here are flaky because we're starting an activity and then attempting to cancel/complete/fail it in parallel. We need the second activity to block until the first one has started, and there's no way to orchestrate that within the workflow. The best we can do is make the acitivites block until the ActivityTask is available.
  • Loading branch information
natemort authored Nov 4, 2024
1 parent d4b2d0b commit 9746755
Showing 1 changed file with 38 additions and 13 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package com.uber.cadence.workflow;

import com.google.common.base.Preconditions;
import com.uber.cadence.WorkflowExecution;
import com.uber.cadence.activity.Activity;
import com.uber.cadence.activity.ActivityMethod;
Expand Down Expand Up @@ -48,6 +47,9 @@ public interface ManualCompletionActivities {
@ActivityMethod
String asyncActivity();

@ActivityMethod
void reset();

@ActivityMethod
void completeAsyncActivity(String result);

Expand All @@ -73,48 +75,61 @@ private class ManualCompletionActivitiesImpl implements ManualCompletionActiviti
@Override
public synchronized String asyncActivity() {
openTask = Activity.getTask();
notifyAll();

Activity.doNotCompleteOnReturn();
return null;
}

@Override
public synchronized void reset() {
openTask = null;
}

@Override
public synchronized void completeAsyncActivity(String details) {
Preconditions.checkState(openTask != null);
getClient().complete(openTask.getTaskToken(), details);
getClient().complete(getOpenTask().getTaskToken(), details);
}

@Override
public synchronized void completeAsyncActivityById(String details) {
Preconditions.checkState(openTask != null);
getClient().complete(getCurrentWorkflow(), openTask.getActivityId(), details);
getClient().complete(getCurrentWorkflow(), getOpenTask().getActivityId(), details);
}

@Override
public synchronized void failAsyncActivity(String details) {
Preconditions.checkState(openTask != null);
getClient()
.completeExceptionally(openTask.getTaskToken(), new ExceptionWithDetaills(details));
.completeExceptionally(getOpenTask().getTaskToken(), new ExceptionWithDetaills(details));
}

@Override
public synchronized void failAsyncActivityById(String details) {
Preconditions.checkState(openTask != null);
getClient()
.completeExceptionally(
getCurrentWorkflow(), openTask.getActivityId(), new ExceptionWithDetaills(details));
getCurrentWorkflow(),
getOpenTask().getActivityId(),
new ExceptionWithDetaills(details));
}

@Override
public synchronized void cancelAsyncActivity(String details) {
Preconditions.checkState(openTask != null);
getClient().reportCancellation(openTask.getTaskToken(), details);
getClient().reportCancellation(getOpenTask().getTaskToken(), details);
}

@Override
public synchronized void cancelAsyncActivityById(String details) {
Preconditions.checkState(openTask != null);
getClient().reportCancellation(getCurrentWorkflow(), openTask.getActivityId(), details);
getClient().reportCancellation(getCurrentWorkflow(), getOpenTask().getActivityId(), details);
}

private synchronized ActivityTask getOpenTask() {
while (openTask == null) {
try {
wait();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
return openTask;
}

private WorkflowExecution getCurrentWorkflow() {
Expand Down Expand Up @@ -146,21 +161,29 @@ public void run() {
expectSuccess("1", result);
expectFailure(() -> activities.completeAsyncActivity("again"));

activities.reset();

result = Async.function(activities::asyncActivity);
activities.completeAsyncActivityById("2");
expectSuccess("2", result);
expectFailure(() -> activities.completeAsyncActivityById("again"));

activities.reset();

result = Async.function(activities::asyncActivity);
activities.failAsyncActivity("3");
expectFailureWithDetails(result, "3");
expectFailure(() -> activities.failAsyncActivity("again"));

activities.reset();

result = Async.function(activities::asyncActivity);
activities.failAsyncActivityById("4");
expectFailureWithDetails(result, "4");
expectFailure(() -> activities.failAsyncActivityById("again"));

activities.reset();

// Need to request cancellation, then the activity can respond with the cancel
CompletablePromise<String> completablePromise = Workflow.newPromise();
CancellationScope scope =
Expand All @@ -178,6 +201,8 @@ public void run() {
activities.cancelAsyncActivity("5");
expectCancelled(result);

activities.reset();

// Need to request cancellation, then the activity can respond with the cancel
CompletablePromise<String> completablePromise2 = Workflow.newPromise();
scope =
Expand Down

0 comments on commit 9746755

Please sign in to comment.