Skip to content

Commit 441ee33

Browse files
Quinn-With-Two-NsQuinn-With-Two-Ns
committed
Align with Update approach
1 parent 16f16b3 commit 441ee33

7 files changed

Lines changed: 269 additions & 157 deletions

File tree

‎temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java‎

Lines changed: 0 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,6 @@
99
import io.temporal.common.Experimental;
1010
import java.lang.reflect.Type;
1111
import java.util.List;
12-
import java.util.Map;
13-
import java.util.Objects;
1412
import java.util.concurrent.CompletableFuture;
1513
import java.util.concurrent.TimeUnit;
1614
import java.util.concurrent.TimeoutException;
@@ -118,56 +116,19 @@ <R> GetActivityResultOutput<R> getActivityResult(GetActivityResultInput<R> input
118116
<R> CompletableFuture<GetActivityResultOutput<R>> getActivityResultAsync(
119117
GetActivityResultInput<R> input);
120118

121-
/**
122-
* Nexus completion-callback metadata propagated through {@link StartActivityInput} when an
123-
* activity is scheduled from a Nexus operation handler.
124-
*/
125-
@Experimental
126-
final class CompletionCallback {
127-
private final String url;
128-
private final Map<String, String> headers;
129-
130-
public CompletionCallback(String url, Map<String, String> headers) {
131-
this.url = Objects.requireNonNull(url);
132-
this.headers = Objects.requireNonNull(headers);
133-
}
134-
135-
public String getUrl() {
136-
return url;
137-
}
138-
139-
public Map<String, String> getHeaders() {
140-
return headers;
141-
}
142-
}
143-
144119
@Experimental
145120
final class StartActivityInput {
146121
private final String activityType;
147122
private final List<Object> args;
148123
private final StartActivityOptions options;
149124
private final Header header;
150-
private final @Nullable CompletionCallback completionCallback;
151-
private final @Nullable List<io.nexusrpc.Link> links;
152125

153126
public StartActivityInput(
154127
String activityType, List<Object> args, StartActivityOptions options, Header header) {
155-
this(activityType, args, options, header, null, null);
156-
}
157-
158-
public StartActivityInput(
159-
String activityType,
160-
List<Object> args,
161-
StartActivityOptions options,
162-
Header header,
163-
@Nullable CompletionCallback completionCallback,
164-
@Nullable List<io.nexusrpc.Link> links) {
165128
this.activityType = activityType;
166129
this.args = args;
167130
this.options = options;
168131
this.header = header;
169-
this.completionCallback = completionCallback;
170-
this.links = links;
171132
}
172133

173134
public String getActivityType() {
@@ -185,38 +146,16 @@ public StartActivityOptions getOptions() {
185146
public Header getHeader() {
186147
return header;
187148
}
188-
189-
@Nullable
190-
public CompletionCallback getCompletionCallback() {
191-
return completionCallback;
192-
}
193-
194-
@Nullable
195-
public List<io.nexusrpc.Link> getLinks() {
196-
return links;
197-
}
198149
}
199150

200151
@Experimental
201152
final class StartActivityOutput {
202153
private final String activityId;
203154
private final @Nullable String activityRunId;
204155

205-
/**
206-
* Set by the invoker when the start request included a Nexus completion callback; null
207-
* otherwise. Internal use; do not depend on this in user-facing code.
208-
*/
209-
private final @Nullable String nexusOperationToken;
210-
211156
public StartActivityOutput(String activityId, @Nullable String activityRunId) {
212-
this(activityId, activityRunId, null);
213-
}
214-
215-
public StartActivityOutput(
216-
String activityId, @Nullable String activityRunId, @Nullable String nexusOperationToken) {
217157
this.activityId = activityId;
218158
this.activityRunId = activityRunId;
219-
this.nexusOperationToken = nexusOperationToken;
220159
}
221160

222161
public String getActivityId() {
@@ -227,11 +166,6 @@ public String getActivityId() {
227166
public String getActivityRunId() {
228167
return activityRunId;
229168
}
230-
231-
@Nullable
232-
public String getNexusOperationToken() {
233-
return nexusOperationToken;
234-
}
235169
}
236170

237171
@Experimental

‎temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java‎

Lines changed: 19 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -22,29 +22,26 @@
2222
import io.temporal.internal.client.external.GenericWorkflowClient;
2323
import io.temporal.internal.common.HeaderUtils;
2424
import io.temporal.internal.common.InternalUtils;
25-
import io.temporal.internal.common.LinkConverter;
2625
import io.temporal.internal.common.ProtoConverters;
2726
import io.temporal.internal.common.ProtobufTimeUtils;
2827
import io.temporal.internal.common.SearchAttributesUtil;
28+
import io.temporal.internal.nexus.CurrentNexusOperationContext;
29+
import io.temporal.internal.nexus.InternalNexusOperationContext;
30+
import io.temporal.internal.nexus.NexusOperationMetadata;
2931
import io.temporal.serviceclient.StatusUtils;
3032
import java.lang.reflect.Type;
3133
import java.util.*;
3234
import java.util.concurrent.CompletableFuture;
3335
import java.util.concurrent.CompletionException;
3436
import java.util.concurrent.TimeoutException;
35-
import java.util.stream.Collectors;
3637
import java.util.stream.StreamSupport;
37-
import org.slf4j.Logger;
38-
import org.slf4j.LoggerFactory;
3938

4039
/**
4140
* Terminus of the activity interceptor chain. Implements all activity RPCs against the Temporal
4241
* service.
4342
*/
4443
public class RootActivityClientInvoker implements ActivityClientCallsInterceptor {
4544

46-
private static final Logger log = LoggerFactory.getLogger(RootActivityClientInvoker.class);
47-
4845
private final GenericWorkflowClient genericClient;
4946
private final ActivityClientOptions clientOptions;
5047

@@ -58,12 +55,19 @@ public RootActivityClientInvoker(
5855
public StartActivityOutput startActivity(StartActivityInput input) {
5956
StartActivityOptions options = input.getOptions();
6057
DataConverter dc = clientOptions.getDataConverter();
58+
InternalNexusOperationContext nexusContext =
59+
CurrentNexusOperationContext.isNexusContext() ? CurrentNexusOperationContext.get() : null;
60+
NexusOperationMetadata nexusOperationMetadata =
61+
nexusContext == null ? null : nexusContext.getNexusOperationMetadata();
6162

6263
StartActivityExecutionRequest.Builder request =
6364
StartActivityExecutionRequest.newBuilder()
6465
.setNamespace(clientOptions.getNamespace())
6566
.setIdentity(clientOptions.getIdentity())
66-
.setRequestId(UUID.randomUUID().toString())
67+
.setRequestId(
68+
nexusOperationMetadata == null
69+
? UUID.randomUUID().toString()
70+
: nexusOperationMetadata.requestId)
6771
.setActivityId(options.getId())
6872
.setActivityType(ActivityType.newBuilder().setName(input.getActivityType()).build())
6973
.setTaskQueue(TaskQueue.newBuilder().setName(options.getTaskQueue()).build())
@@ -113,41 +117,13 @@ public StartActivityOutput startActivity(StartActivityInput input) {
113117
io.temporal.api.common.v1.Header grpcHeader = HeaderUtils.toHeaderGrpc(input.getHeader(), null);
114118
request.setHeader(grpcHeader);
115119

116-
// Hoisted so it can be returned in StartActivityOutput when a Nexus callback is present.
117-
String nexusOperationToken = null;
118-
if (input.getCompletionCallback() != null) {
119-
List<Link> protoLinks = null;
120-
if (input.getLinks() != null) {
121-
protoLinks =
122-
input.getLinks().stream()
123-
.map(
124-
link -> {
125-
if (io.temporal.api.common.v1.Link.WorkflowEvent.getDescriptor()
126-
.getFullName()
127-
.equals(link.getType())) {
128-
io.temporal.api.nexus.v1.Link nexusLink =
129-
io.temporal.api.nexus.v1.Link.newBuilder()
130-
.setType(link.getType())
131-
.setUrl(link.getUri().toString())
132-
.build();
133-
return LinkConverter.nexusLinkToWorkflowEvent(nexusLink);
134-
} else {
135-
log.warn("ignoring unsupported link data type: {}", link.getType());
136-
return null;
137-
}
138-
})
139-
.filter(Objects::nonNull)
140-
.collect(Collectors.toList());
141-
if (!protoLinks.isEmpty()) {
142-
request.addAllLinks(protoLinks);
143-
} else {
144-
protoLinks = null;
145-
}
146-
}
120+
if (nexusOperationMetadata != null) {
121+
List<Link> protoLinks = nexusContext.getRequestLinks();
122+
request.addAllLinks(protoLinks);
147123
// Generate the operation token from the user-supplied activity ID and namespace so the
148124
// dual OPERATION_ID + OPERATION_TOKEN headers can be injected before the start RPC fires.
149125
try {
150-
nexusOperationToken =
126+
nexusOperationMetadata.operationToken =
151127
io.temporal.internal.nexus.OperationTokenUtil.generateActivityExecutionOperationToken(
152128
options.getId(), clientOptions.getNamespace());
153129
} catch (com.fasterxml.jackson.core.JsonProcessingException e) {
@@ -158,9 +134,9 @@ public StartActivityOutput startActivity(StartActivityInput input) {
158134
}
159135
Callback cb =
160136
InternalUtils.buildNexusCallback(
161-
input.getCompletionCallback().getUrl(),
162-
input.getCompletionCallback().getHeaders(),
163-
nexusOperationToken,
137+
nexusOperationMetadata.callbackUrl,
138+
nexusOperationMetadata.callbackHeaders,
139+
nexusOperationMetadata.operationToken,
164140
protoLinks);
165141
request.addCompletionCallbacks(cb);
166142
}
@@ -182,7 +158,7 @@ public StartActivityOutput startActivity(StartActivityInput input) {
182158
}
183159

184160
String runId = response.getRunId().isEmpty() ? null : response.getRunId();
185-
return new StartActivityOutput(options.getId(), runId, nexusOperationToken);
161+
return new StartActivityOutput(options.getId(), runId);
186162
}
187163

188164
@Override

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/InternalNexusOperationContext.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ public class InternalNexusOperationContext {
3737
// by responseLinksLock and getResponseLinks() returns a defensive copy taken under the lock.
3838
private final Object responseLinksLock = new Object();
3939
private final List<Link> responseLinks = new ArrayList<>();
40+
private NexusOperationMetadata nexusOperationMetadata;
4041

4142
public InternalNexusOperationContext(
4243
String namespace,
@@ -82,6 +83,15 @@ public NexusOperationContext getUserFacingContext() {
8283
return new NexusOperationContextImpl();
8384
}
8485

86+
/** Sets metadata for the Temporal primitive backing the current Nexus operation. */
87+
public void setNexusOperationMetadata(NexusOperationMetadata metadata) {
88+
this.nexusOperationMetadata = metadata;
89+
}
90+
91+
public NexusOperationMetadata getNexusOperationMetadata() {
92+
return nexusOperationMetadata;
93+
}
94+
8595
/**
8696
* Set the {@code common.v1.Link}s extracted from the inbound Nexus task so they can be attached
8797
* to RPCs issued by the operation handler.

‎temporal-sdk/src/main/java/io/temporal/internal/nexus/NexusOperationMetadata.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package io.temporal.internal.nexus;
2+
3+
import io.temporal.common.Experimental;
4+
import java.util.Map;
5+
6+
/** Container for in-flight Nexus operation metadata. */
7+
@Experimental
8+
public final class NexusOperationMetadata {
9+
public final String requestId;
10+
public final String callbackUrl;
11+
public final Map<String, String> callbackHeaders;
12+
13+
public String operationToken;
14+
public boolean operationCompleted;
15+
16+
public NexusOperationMetadata(
17+
String requestId, String callbackUrl, Map<String, String> callbackHeaders) {
18+
this.requestId = requestId;
19+
this.callbackUrl = callbackUrl;
20+
this.callbackHeaders = callbackHeaders;
21+
}
22+
}

‎temporal-sdk/src/main/java/io/temporal/nexus/TemporalNexusClientImpl.java‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,9 @@
1616
import io.temporal.internal.client.ActivityClientInternal;
1717
import io.temporal.internal.client.NexusStartActivityResponse;
1818
import io.temporal.internal.client.NexusStartWorkflowResponse;
19+
import io.temporal.internal.nexus.CurrentNexusOperationContext;
20+
import io.temporal.internal.nexus.InternalNexusOperationContext;
21+
import io.temporal.internal.nexus.NexusOperationMetadata;
1922
import io.temporal.internal.nexus.NexusStartActivityHelper;
2023
import io.temporal.internal.nexus.NexusStartWorkflowHelper;
2124
import io.temporal.internal.nexus.OperationTokenUtil;
@@ -473,7 +476,14 @@ public <R> TemporalOperationResult<R> startActivity(
473476
private <R> TemporalOperationResult<R> startActivityImpl(
474477
String activityType, List<Object> args, StartActivityOptions options) {
475478
markAsyncOperationStarted();
479+
InternalNexusOperationContext nexusContext = CurrentNexusOperationContext.get();
476480
try {
481+
NexusOperationMetadata nexusOperationMetadata =
482+
new NexusOperationMetadata(
483+
operationStartDetails.getRequestId(),
484+
operationStartDetails.getCallbackUrl(),
485+
operationStartDetails.getCallbackHeaders());
486+
nexusContext.setNexusOperationMetadata(nexusOperationMetadata);
477487
NexusStartActivityResponse response =
478488
NexusStartActivityHelper.startActivityAndAttachLinks(
479489
operationContext,
@@ -483,19 +493,12 @@ private <R> TemporalOperationResult<R> startActivityImpl(
483493
options,
484494
Header.empty(),
485495
request -> {
486-
ActivityClientCallsInterceptor.CompletionCallback cb =
487-
(request.getCallbackUrl() == null || request.getCallbackUrl().isEmpty())
488-
? null
489-
: new ActivityClientCallsInterceptor.CompletionCallback(
490-
request.getCallbackUrl(), request.getCallbackHeaders());
491496
ActivityClientCallsInterceptor.StartActivityInput input =
492497
new ActivityClientCallsInterceptor.StartActivityInput(
493498
request.getActivityType(),
494499
request.getArgs(),
495500
request.getOptions(),
496-
request.getHeader(),
497-
cb,
498-
request.getLinks());
501+
request.getHeader());
499502
// Build an ActivityClient that mirrors the surrounding WorkflowClient's options
500503
// so that the metrics-tagged scope built by the impl is preserved. setIdentity is
501504
// load-bearing because the cancel RPC reads it from clientOptions.getIdentity().
@@ -514,13 +517,13 @@ private <R> TemporalOperationResult<R> startActivityImpl(
514517
// because the run ID isn't known until after the start RPC returns. The operation
515518
// token returned to the Nexus caller can — and should — include it, so it's
516519
// regenerated here from the same activity ID + the run ID the start RPC produced.
517-
String headerToken = out.getNexusOperationToken();
520+
String headerToken = nexusOperationMetadata.operationToken;
518521
if (headerToken == null) {
519522
throw new HandlerException(
520523
HandlerException.ErrorType.INTERNAL,
521524
"invoker did not return a Nexus operation token for activity start with callback",
522525
new IllegalStateException(
523-
"nexusOperationToken is null on StartActivityOutput when CompletionCallback was set"));
526+
"operationToken is null on NexusOperationMetadata after activity start"));
524527
}
525528
String returnToken;
526529
try {
@@ -544,6 +547,8 @@ private <R> TemporalOperationResult<R> startActivityImpl(
544547
// being blocked by the guard.
545548
asyncOperationStarted.set(false);
546549
throw t;
550+
} finally {
551+
nexusContext.setNexusOperationMetadata(null);
547552
}
548553
}
549554

0 commit comments

Comments
 (0)