7373import java .util .function .LongSupplier ;
7474import java .util .stream .Collectors ;
7575import java .util .stream .IntStream ;
76+ import javax .annotation .Nonnull ;
7677import javax .annotation .Nullable ;
7778import org .slf4j .Logger ;
7879import 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