Skip to content

Commit e3fbdfb

Browse files
justinp-ttrodrigozhou
authored and
Quinn-With-Two-Ns
committed
Add OnConflictOptions Support (#2415)
* Add OnConflictOptions support --------- Co-authored-by: Rodrigo Zhou <2068124+rodrigozhou@users.noreply.github.com> Co-authored-by: rodrigozhou <rodrigo.zhou@temporal.io>
1 parent ff94971 commit e3fbdfb

File tree

10 files changed

+464
-94
lines changed

10 files changed

+464
-94
lines changed

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/WorkflowHandleFailOnConflictTest.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,6 @@
2020

2121
package io.temporal.workflow.nexus;
2222

23-
import static org.junit.Assume.assumeTrue;
24-
2523
import io.nexusrpc.handler.HandlerException;
2624
import io.nexusrpc.handler.OperationHandler;
2725
import io.nexusrpc.handler.OperationImpl;
@@ -40,7 +38,6 @@
4038
import java.util.List;
4139
import java.util.UUID;
4240
import org.junit.Assert;
43-
import org.junit.Before;
4441
import org.junit.Rule;
4542
import org.junit.Test;
4643

@@ -52,12 +49,6 @@ public class WorkflowHandleFailOnConflictTest {
5249
.setNexusServiceImplementation(new TestNexusServiceImpl())
5350
.build();
5451

55-
@Before
56-
public void checkRealServer() {
57-
assumeTrue(
58-
"Test Server doesn't support OnConflictOption yet", SDKTestWorkflowRule.useExternalService);
59-
}
60-
6152
@Test
6253
public void testOnConflictFail() {
6354
TestWorkflows.TestWorkflow1 workflowStub =

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/WorkflowHandleUseExistingOnConflictCancelTest.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,6 @@
2020

2121
package io.temporal.workflow.nexus;
2222

23-
import static org.junit.Assume.assumeTrue;
24-
2523
import io.nexusrpc.handler.OperationHandler;
2624
import io.nexusrpc.handler.OperationImpl;
2725
import io.nexusrpc.handler.ServiceImpl;
@@ -49,12 +47,6 @@ public class WorkflowHandleUseExistingOnConflictCancelTest {
4947
.setNexusServiceImplementation(new TestNexusServiceImpl())
5048
.build();
5149

52-
@Before
53-
public void checkRealServer() {
54-
assumeTrue(
55-
"Test Server doesn't support OnConflictOption yet", SDKTestWorkflowRule.useExternalService);
56-
}
57-
5850
@Test
5951
public void testUseExistingCancel() {
6052
TestWorkflows.TestWorkflow1 workflowStub =

‎temporal-sdk/src/test/java/io/temporal/workflow/nexus/WorkflowHandleUseExistingOnConflictTest.java‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,6 @@
2020

2121
package io.temporal.workflow.nexus;
2222

23-
import static org.junit.Assume.assumeTrue;
24-
2523
import io.nexusrpc.handler.OperationHandler;
2624
import io.nexusrpc.handler.OperationImpl;
2725
import io.nexusrpc.handler.ServiceImpl;
@@ -47,12 +45,6 @@ public class WorkflowHandleUseExistingOnConflictTest {
4745
.setNexusServiceImplementation(new TestNexusServiceImpl())
4846
.build();
4947

50-
@Before
51-
public void checkRealServer() {
52-
assumeTrue(
53-
"Test Server doesn't support OnConflictOption yet", SDKTestWorkflowRule.useExternalService);
54-
}
55-
5648
@Test
5749
public void testOnConflictUseExisting() {
5850
TestWorkflows.TestWorkflow1 workflowStub =

‎temporal-test-server/src/main/java/io/temporal/internal/testservice/ExecutionId.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,17 +30,17 @@
3030
import java.io.IOException;
3131
import java.util.Objects;
3232

33-
final class ExecutionId {
33+
public final class ExecutionId {
3434

3535
private final String namespace;
3636
private final WorkflowExecution execution;
3737

38-
ExecutionId(String namespace, WorkflowExecution execution) {
38+
public ExecutionId(String namespace, WorkflowExecution execution) {
3939
this.namespace = Objects.requireNonNull(namespace);
4040
this.execution = Objects.requireNonNull(execution);
4141
}
4242

43-
ExecutionId(String namespace, String workflowId, String runId) {
43+
public ExecutionId(String namespace, String workflowId, String runId) {
4444
this(
4545
namespace,
4646
WorkflowExecution.newBuilder()

‎temporal-test-server/src/main/java/io/temporal/internal/testservice/StateMachines.java‎

Lines changed: 10 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -20,33 +20,17 @@
2020

2121
package io.temporal.internal.testservice;
2222

23-
import static io.temporal.internal.common.LinkConverter.*;
24-
import static io.temporal.internal.testservice.StateMachines.Action.CANCEL;
25-
import static io.temporal.internal.testservice.StateMachines.Action.COMPLETE;
26-
import static io.temporal.internal.testservice.StateMachines.Action.CONTINUE_AS_NEW;
27-
import static io.temporal.internal.testservice.StateMachines.Action.FAIL;
28-
import static io.temporal.internal.testservice.StateMachines.Action.INITIATE;
29-
import static io.temporal.internal.testservice.StateMachines.Action.QUERY;
30-
import static io.temporal.internal.testservice.StateMachines.Action.REQUEST_CANCELLATION;
31-
import static io.temporal.internal.testservice.StateMachines.Action.START;
32-
import static io.temporal.internal.testservice.StateMachines.Action.TERMINATE;
33-
import static io.temporal.internal.testservice.StateMachines.Action.TIME_OUT;
34-
import static io.temporal.internal.testservice.StateMachines.Action.UPDATE;
35-
import static io.temporal.internal.testservice.StateMachines.Action.UPDATE_WORKFLOW_EXECUTION;
36-
import static io.temporal.internal.testservice.StateMachines.State.CANCELED;
37-
import static io.temporal.internal.testservice.StateMachines.State.CANCELLATION_REQUESTED;
38-
import static io.temporal.internal.testservice.StateMachines.State.COMPLETED;
39-
import static io.temporal.internal.testservice.StateMachines.State.CONTINUED_AS_NEW;
40-
import static io.temporal.internal.testservice.StateMachines.State.FAILED;
41-
import static io.temporal.internal.testservice.StateMachines.State.INITIATED;
42-
import static io.temporal.internal.testservice.StateMachines.State.NONE;
43-
import static io.temporal.internal.testservice.StateMachines.State.STARTED;
44-
import static io.temporal.internal.testservice.StateMachines.State.TERMINATED;
45-
import static io.temporal.internal.testservice.StateMachines.State.TIMED_OUT;
23+
import static io.temporal.internal.common.LinkConverter.nexusLinkToWorkflowEvent;
24+
import static io.temporal.internal.common.LinkConverter.workflowEventToNexusLink;
25+
import static io.temporal.internal.testservice.StateMachines.Action.*;
26+
import static io.temporal.internal.testservice.StateMachines.State.*;
4627

4728
import com.google.common.base.Preconditions;
4829
import com.google.common.base.Strings;
49-
import com.google.protobuf.*;
30+
import com.google.protobuf.Any;
31+
import com.google.protobuf.Duration;
32+
import com.google.protobuf.InvalidProtocolBufferException;
33+
import com.google.protobuf.Timestamp;
5034
import com.google.protobuf.util.Durations;
5135
import com.google.protobuf.util.Timestamps;
5236
import io.grpc.Status;
@@ -65,7 +49,8 @@
6549
import io.temporal.api.query.v1.WorkflowQueryResult;
6650
import io.temporal.api.taskqueue.v1.StickyExecutionAttributes;
6751
import io.temporal.api.taskqueue.v1.TaskQueue;
68-
import io.temporal.api.update.v1.*;
52+
import io.temporal.api.update.v1.Acceptance;
53+
import io.temporal.api.update.v1.Outcome;
6954
import io.temporal.api.update.v1.Request;
7055
import io.temporal.api.update.v1.Response;
7156
import io.temporal.api.workflowservice.v1.*;

‎temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableState.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222

2323
import io.grpc.Deadline;
2424
import io.temporal.api.command.v1.SignalExternalWorkflowExecutionCommandAttributes;
25+
import io.temporal.api.common.v1.Callback;
2526
import io.temporal.api.common.v1.Payload;
2627
import io.temporal.api.common.v1.Payloads;
2728
import io.temporal.api.enums.v1.SignalExternalWorkflowExecutionFailedCause;
@@ -32,6 +33,7 @@
3233
import io.temporal.api.nexus.v1.StartOperationResponse;
3334
import io.temporal.api.taskqueue.v1.StickyExecutionAttributes;
3435
import io.temporal.api.workflowservice.v1.*;
36+
import java.util.List;
3537
import java.util.Optional;
3638
import java.util.function.Consumer;
3739
import javax.annotation.Nullable;
@@ -49,6 +51,8 @@ void startWorkflowTask(
4951

5052
void completeWorkflowTask(int historySize, RespondWorkflowTaskCompletedRequest request);
5153

54+
void applyOnConflictOptions(StartWorkflowExecutionRequest request);
55+
5256
void reportCancelRequested(ExternalWorkflowExecutionCancelRequestedEventAttributes a);
5357

5458
void completeSignalExternalWorkflowExecution(String signalId, String runId);
@@ -143,4 +147,8 @@ PollWorkflowExecutionUpdateResponse pollUpdateWorkflowExecution(
143147
Optional<TestWorkflowMutableState> getParent();
144148

145149
boolean isTerminalState();
150+
151+
boolean isRequestIdAttached(String requestId);
152+
153+
List<Callback> getCompletionCallbacks();
146154
}

‎temporal-test-server/src/main/java/io/temporal/internal/testservice/TestWorkflowMutableStateImpl.java‎

Lines changed: 77 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
import java.util.function.LongSupplier;
7474
import java.util.stream.Collectors;
7575
import java.util.stream.IntStream;
76+
import javax.annotation.Nonnull;
7677
import javax.annotation.Nullable;
7778
import org.slf4j.Logger;
7879
import org.slf4j.LoggerFactory;
@@ -136,6 +137,8 @@ private interface UpdateProcedure {
136137
new ConcurrentHashMap<>();
137138
public StickyExecutionAttributes stickyExecutionAttributes;
138139
private Map<String, Payload> currentMemo;
140+
private final Set<String> attachedRequestIds = new HashSet<>();
141+
private final List<Callback> completionCallbacks = new ArrayList<>();
139142

140143
/**
141144
* @param retryState present if workflow is a retry
@@ -184,6 +187,7 @@ private interface UpdateProcedure {
184187
this.workflow = StateMachines.newWorkflowStateMachine(data);
185188
this.workflowTaskStateMachine = StateMachines.newWorkflowTaskStateMachine(store, startRequest);
186189
this.currentMemo = new HashMap(startRequest.getMemo().getFieldsMap());
190+
this.completionCallbacks.addAll(startRequest.getCompletionCallbacksList());
187191
}
188192

189193
/** Based on overrideStartWorkflowExecutionRequest from historyEngine.go */
@@ -613,6 +617,29 @@ public void completeWorkflowTask(
613617
request.hasStickyAttributes() ? request.getStickyAttributes() : null);
614618
}
615619

620+
@Override
621+
public void applyOnConflictOptions(@Nonnull StartWorkflowExecutionRequest request) {
622+
update(
623+
ctx -> {
624+
OnConflictOptions options = request.getOnConflictOptions();
625+
String requestId = null;
626+
List<Callback> completionCallbacks = null;
627+
List<Link> links = null;
628+
629+
if (options.getAttachRequestId()) {
630+
requestId = request.getRequestId();
631+
}
632+
if (options.getAttachCompletionCallbacks()) {
633+
completionCallbacks = request.getCompletionCallbacksList();
634+
}
635+
if (options.getAttachLinks()) {
636+
links = request.getLinksList();
637+
}
638+
639+
addWorkflowExecutionOptionsUpdatedEvent(ctx, requestId, completionCallbacks, links);
640+
});
641+
}
642+
616643
private void failWorkflowTaskWithAReason(
617644
WorkflowTaskFailedCause failedCause,
618645
ServerFailure eventAttributesFailure,
@@ -1476,6 +1503,7 @@ private void processFailWorkflowExecution(
14761503
identity,
14771504
getExecutionId(),
14781505
workflow.getData().firstExecutionRunId,
1506+
this,
14791507
parent,
14801508
parentChildInitiatedEventId);
14811509
return;
@@ -1608,6 +1636,7 @@ private void startNewCronRun(
16081636
identity,
16091637
getExecutionId(),
16101638
workflow.getData().firstExecutionRunId,
1639+
this,
16111640
parent,
16121641
parentChildInitiatedEventId);
16131642
}
@@ -1665,6 +1694,7 @@ private void processContinueAsNewWorkflowExecution(
16651694
identity,
16661695
getExecutionId(),
16671696
workflow.getData().firstExecutionRunId,
1697+
this,
16681698
parent,
16691699
parentChildInitiatedEventId);
16701700
}
@@ -1696,7 +1726,7 @@ private void processWorkflowCompletionCallbacks(RequestContext ctx) {
16961726
}
16971727
});
16981728

1699-
for (Callback cb : startRequest.getCompletionCallbacksList()) {
1729+
for (Callback cb : completionCallbacks) {
17001730
if (!cb.hasNexus()) {
17011731
// test server only supports nexus callbacks currently
17021732
log.warn("skipping non-nexus completion callback");
@@ -1718,8 +1748,16 @@ private void processWorkflowCompletionCallbacks(RequestContext ctx) {
17181748
.build())
17191749
.build());
17201750

1721-
service.completeNexusOperation(
1722-
ref, ctx.getExecution().getWorkflowId(), startLink, completionEvent.get());
1751+
try {
1752+
service.completeNexusOperation(
1753+
ref, ctx.getExecution().getWorkflowId(), startLink, completionEvent.get());
1754+
} catch (StatusRuntimeException e) {
1755+
// Callback destination not found should not block processing the callbacks nor
1756+
// completing the workflow.
1757+
if (e.getStatus().getCode() != Status.Code.NOT_FOUND) {
1758+
throw e;
1759+
}
1760+
}
17231761
}
17241762
}
17251763

@@ -1985,6 +2023,11 @@ public boolean isTerminalState() {
19852023
return isTerminalState(workflowState);
19862024
}
19872025

2026+
@Override
2027+
public boolean isRequestIdAttached(@Nonnull String requestId) {
2028+
return attachedRequestIds.contains(requestId);
2029+
}
2030+
19882031
private void updateHeartbeatTimer(
19892032
RequestContext ctx,
19902033
long activityId,
@@ -3121,7 +3164,7 @@ private DescribeWorkflowExecutionResponse describeWorkflowExecutionInsideLock()
31213164
.setParentExecution(p.getExecutionId().getExecution()));
31223165

31233166
List<CallbackInfo> callbacks =
3124-
this.startRequest.getCompletionCallbacksList().stream()
3167+
this.completionCallbacks.stream()
31253168
.map(TestWorkflowMutableStateImpl::constructCallbackInfo)
31263169
.collect(Collectors.toList());
31273170

@@ -3429,6 +3472,31 @@ private void addExecutionSignaledByExternalEvent(
34293472
ctx.addEvent(executionSignaled);
34303473
}
34313474

3475+
private void addWorkflowExecutionOptionsUpdatedEvent(
3476+
RequestContext ctx, String requestId, List<Callback> completionCallbacks, List<Link> links) {
3477+
WorkflowExecutionOptionsUpdatedEventAttributes.Builder attrs =
3478+
WorkflowExecutionOptionsUpdatedEventAttributes.newBuilder();
3479+
if (requestId != null) {
3480+
attrs.setAttachedRequestId(requestId);
3481+
this.attachedRequestIds.add(requestId);
3482+
}
3483+
if (completionCallbacks != null) {
3484+
attrs.addAllAttachedCompletionCallbacks(completionCallbacks);
3485+
this.completionCallbacks.addAll(completionCallbacks);
3486+
}
3487+
3488+
HistoryEvent.Builder event =
3489+
HistoryEvent.newBuilder()
3490+
.setWorkerMayIgnore(true)
3491+
.setEventType(EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED)
3492+
.setWorkflowExecutionOptionsUpdatedEventAttributes(attrs);
3493+
if (links != null) {
3494+
event.addAllLinks(links);
3495+
}
3496+
3497+
ctx.addEvent(event.build());
3498+
}
3499+
34323500
private StateMachine<ActivityTaskData> getPendingActivityById(String activityId) {
34333501
Long scheduledEventId = activityById.get(activityId);
34343502
if (scheduledEventId == null) {
@@ -3555,4 +3623,9 @@ private boolean isTerminalState(State workflowState) {
35553623
|| workflowState == State.TERMINATED
35563624
|| workflowState == State.CONTINUED_AS_NEW;
35573625
}
3626+
3627+
@Override
3628+
public List<Callback> getCompletionCallbacks() {
3629+
return completionCallbacks;
3630+
}
35583631
}

0 commit comments

Comments
 (0)