Skip to content

Commit 2ac714a

Browse files
vancexuvancexu
committed
Support Search Attributes on Start workflow
1 parent 1b2a2ce commit 2ac714a

File tree

8 files changed

+147
-22
lines changed

8 files changed

+147
-22
lines changed

‎src/main/java/com/uber/cadence/client/WorkflowOptions.java

Lines changed: 30 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ public static WorkflowOptions merge(
6464
.setRetryOptions(RetryOptions.merge(methodRetry, o.getRetryOptions()))
6565
.setCronSchedule(OptionsUtils.merge(cronAnnotation, o.getCronSchedule(), String.class))
6666
.setMemo(o.getMemo())
67+
.setSearchAttributes(o.getSearchAttributes())
6768
.validateBuildWithDefaults();
6869
}
6970

@@ -87,6 +88,8 @@ public static final class Builder {
8788

8889
private Map<String, Object> memo;
8990

91+
private Map<String, Object> searchAttributes;
92+
9093
public Builder() {}
9194

9295
public Builder(WorkflowOptions o) {
@@ -102,6 +105,7 @@ public Builder(WorkflowOptions o) {
102105
this.retryOptions = o.retryOptions;
103106
this.cronSchedule = o.cronSchedule;
104107
this.memo = o.memo;
108+
this.searchAttributes = o.searchAttributes;
105109
}
106110

107111
/**
@@ -192,6 +196,12 @@ public Builder setMemo(Map<String, Object> memo) {
192196
return this;
193197
}
194198

199+
/** Specifies additional indexed information in result of list workflow. */
200+
public Builder setSearchAttributes(Map<String, Object> searchAttributes) {
201+
this.searchAttributes = searchAttributes;
202+
return this;
203+
}
204+
195205
public WorkflowOptions build() {
196206
return new WorkflowOptions(
197207
workflowId,
@@ -202,7 +212,8 @@ public WorkflowOptions build() {
202212
childPolicy,
203213
retryOptions,
204214
cronSchedule,
205-
memo);
215+
memo,
216+
searchAttributes);
206217
}
207218

208219
/**
@@ -248,7 +259,8 @@ public WorkflowOptions validateBuildWithDefaults() {
248259
childPolicy,
249260
retryOptions,
250261
cronSchedule,
251-
memo);
262+
memo,
263+
searchAttributes);
252264
}
253265
}
254266

@@ -270,6 +282,8 @@ public WorkflowOptions validateBuildWithDefaults() {
270282

271283
private Map<String, Object> memo;
272284

285+
private Map<String, Object> searchAttributes;
286+
273287
private WorkflowOptions(
274288
String workflowId,
275289
WorkflowIdReusePolicy workflowIdReusePolicy,
@@ -279,7 +293,8 @@ private WorkflowOptions(
279293
ChildPolicy childPolicy,
280294
RetryOptions retryOptions,
281295
String cronSchedule,
282-
Map<String, Object> memo) {
296+
Map<String, Object> memo,
297+
Map<String, Object> searchAttributes) {
283298
this.workflowId = workflowId;
284299
this.workflowIdReusePolicy = workflowIdReusePolicy;
285300
this.executionStartToCloseTimeout = executionStartToCloseTimeout;
@@ -289,6 +304,7 @@ private WorkflowOptions(
289304
this.retryOptions = retryOptions;
290305
this.cronSchedule = cronSchedule;
291306
this.memo = memo;
307+
this.searchAttributes = searchAttributes;
292308
}
293309

294310
public String getWorkflowId() {
@@ -327,6 +343,10 @@ public Map<String, Object> getMemo() {
327343
return memo;
328344
}
329345

346+
public Map<String, Object> getSearchAttributes() {
347+
return searchAttributes;
348+
}
349+
330350
@Override
331351
public boolean equals(Object o) {
332352
if (this == o) return true;
@@ -340,7 +360,8 @@ public boolean equals(Object o) {
340360
&& childPolicy == that.childPolicy
341361
&& Objects.equals(retryOptions, that.retryOptions)
342362
&& Objects.equals(cronSchedule, that.cronSchedule)
343-
&& Objects.equals(memo, that.memo);
363+
&& Objects.equals(memo, that.memo)
364+
&& Objects.equals(searchAttributes, that.searchAttributes);
344365
}
345366

346367
@Override
@@ -354,7 +375,8 @@ public int hashCode() {
354375
childPolicy,
355376
retryOptions,
356377
cronSchedule,
357-
memo);
378+
memo,
379+
searchAttributes);
358380
}
359381

360382
@Override
@@ -382,6 +404,9 @@ public String toString() {
382404
+ ", memo='"
383405
+ memo
384406
+ '\''
407+
+ ", searchAttributes='"
408+
+ searchAttributes
409+
+ '\''
385410
+ '}';
386411
}
387412
}

‎src/main/java/com/uber/cadence/internal/common/StartWorkflowExecutionParameters.java

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,8 @@ public final class StartWorkflowExecutionParameters {
5353

5454
private Map<String, byte[]> memo;
5555

56+
private Map<String, byte[]> searchAttributes;
57+
5658
/**
5759
* Returns the value of the WorkflowId property for this object.
5860
*
@@ -303,6 +305,14 @@ public void setMemo(Map<String, byte[]> memo) {
303305
this.memo = memo;
304306
}
305307

308+
public Map<String, byte[]> getSearchAttributes() {
309+
return searchAttributes;
310+
}
311+
312+
public void setSearchAttributes(Map<String, byte[]> searchAttributes) {
313+
this.searchAttributes = searchAttributes;
314+
}
315+
306316
public StartWorkflowExecutionParameters withRetryParameters(RetryParameters retryParameters) {
307317
this.retryParameters = retryParameters;
308318
return this;
@@ -378,6 +388,9 @@ public String toString() {
378388
+ ", memo='"
379389
+ memo
380390
+ '\''
391+
+ ", searchAttributes='"
392+
+ searchAttributes
393+
+ '\''
381394
+ '}';
382395
}
383396

@@ -396,7 +409,8 @@ public boolean equals(Object o) {
396409
&& workflowIdReusePolicy == that.workflowIdReusePolicy
397410
&& Objects.equals(retryParameters, that.retryParameters)
398411
&& Objects.equals(cronSchedule, that.cronSchedule)
399-
&& Objects.equals(memo, that.memo);
412+
&& Objects.equals(memo, that.memo)
413+
&& Objects.equals(searchAttributes, that.searchAttributes);
400414
}
401415

402416
@Override
@@ -412,7 +426,8 @@ public int hashCode() {
412426
workflowIdReusePolicy,
413427
retryParameters,
414428
cronSchedule,
415-
memo);
429+
memo,
430+
searchAttributes);
416431
result = 31 * result + Arrays.hashCode(input);
417432
return result;
418433
}

‎src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import com.uber.cadence.QueryWorkflowResponse;
2424
import com.uber.cadence.RequestCancelWorkflowExecutionRequest;
2525
import com.uber.cadence.RetryPolicy;
26+
import com.uber.cadence.SearchAttributes;
2627
import com.uber.cadence.SignalWithStartWorkflowExecutionRequest;
2728
import com.uber.cadence.SignalWorkflowExecutionRequest;
2829
import com.uber.cadence.StartWorkflowExecutionRequest;
@@ -117,6 +118,7 @@ private WorkflowExecution startWorkflowInternal(StartWorkflowExecutionParameters
117118
request.setCronSchedule(startParameters.getCronSchedule());
118119
}
119120
request.setMemo(toMemoThrift(startParameters.getMemo()));
121+
request.setSearchAttributes(toSearchAttributesThrift(startParameters.getSearchAttributes()));
120122

121123
// if(startParameters.getChildPolicy() != null) {
122124
// request.setChildPolicy(startParameters.getChildPolicy());
@@ -154,6 +156,20 @@ private Memo toMemoThrift(Map<String, byte[]> memo) {
154156
return memoThrift;
155157
}
156158

159+
private SearchAttributes toSearchAttributesThrift(Map<String, byte[]> searchAttributes) {
160+
if (searchAttributes == null || searchAttributes.isEmpty()) {
161+
return null;
162+
}
163+
164+
Map<String, ByteBuffer> fields = new HashMap<>();
165+
for (Map.Entry<String, byte[]> item : searchAttributes.entrySet()) {
166+
fields.put(item.getKey(), ByteBuffer.wrap(item.getValue()));
167+
}
168+
SearchAttributes searchAttrThrift = new SearchAttributes();
169+
searchAttrThrift.setIndexedFields(fields);
170+
return searchAttrThrift;
171+
}
172+
157173
private RetryPolicy toRetryPolicy(RetryParameters retryParameters) {
158174
return new RetryPolicy()
159175
.setBackoffCoefficient(retryParameters.getBackoffCoefficient())

‎src/main/java/com/uber/cadence/internal/sync/WorkflowStubImpl.java

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -139,24 +139,24 @@ private StartWorkflowExecutionParameters getStartWorkflowExecutionParameters(
139139
}
140140
p.setInput(dataConverter.toData(args));
141141
p.setWorkflowType(new WorkflowType().setName(workflowType.get()));
142-
p.setMemo(convertMemoFromObjectToBytes(o.getMemo()));
142+
p.setMemo(convertMapFromObjectToBytes(o.getMemo()));
143+
p.setSearchAttributes(convertMapFromObjectToBytes(o.getSearchAttributes()));
143144
return p;
144145
}
145146

146-
private Map<String, byte[]> convertMemoFromObjectToBytes(Map<String, Object> memoFromOption) {
147-
if (memoFromOption == null) {
147+
private Map<String, byte[]> convertMapFromObjectToBytes(Map<String, Object> map) {
148+
if (map == null) {
148149
return null;
149150
}
150-
Map<String, byte[]> memo = new HashMap<>();
151-
for (Map.Entry<String, Object> item : memoFromOption.entrySet()) {
151+
Map<String, byte[]> result = new HashMap<>();
152+
for (Map.Entry<String, Object> item : map.entrySet()) {
152153
try {
153-
memo.put(item.getKey(), dataConverter.toData(item.getValue()));
154+
result.put(item.getKey(), dataConverter.toData(item.getValue()));
154155
} catch (DataConverterException e) {
155-
throw new DataConverterException(
156-
"Cannot serialize memo for key " + item.getKey(), e.getCause());
156+
throw new DataConverterException("Cannot serialize key " + item.getKey(), e.getCause());
157157
}
158158
}
159-
return memo;
159+
return result;
160160
}
161161

162162
@Override

‎src/main/java/com/uber/cadence/internal/testservice/StateMachines.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -539,6 +539,7 @@ private static void startWorkflow(
539539
}
540540
a.setLastCompletionResult(data.lastCompletionResult);
541541
a.setMemo(request.getMemo());
542+
a.setSearchAttributes((request.getSearchAttributes()));
542543
HistoryEvent event =
543544
new HistoryEvent()
544545
.setEventType(EventType.WorkflowExecutionStarted)

‎src/main/java/com/uber/cadence/workflow/ChildWorkflowOptions.java

Lines changed: 30 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ public static ChildWorkflowOptions merge(
5555
.setRetryOptions(RetryOptions.merge(r, o.getRetryOptions()))
5656
.setCronSchedule(OptionsUtils.merge(cronAnnotation, o.getCronSchedule(), String.class))
5757
.setMemo(o.getMemo())
58+
.setSearchAttributes(o.getSearchAttributes())
5859
.validateAndBuildWithDefaults();
5960
}
6061

@@ -80,6 +81,8 @@ public static final class Builder {
8081

8182
private Map<String, Object> memo;
8283

84+
private Map<String, Object> searchAttributes;
85+
8386
public Builder() {}
8487

8588
public Builder(ChildWorkflowOptions source) {
@@ -96,6 +99,7 @@ public Builder(ChildWorkflowOptions source) {
9699
this.retryOptions = source.getRetryOptions();
97100
this.cronSchedule = source.getCronSchedule();
98101
this.memo = source.getMemo();
102+
this.searchAttributes = source.getSearchAttributes();
99103
}
100104

101105
/**
@@ -200,6 +204,12 @@ public Builder setMemo(Map<String, Object> memo) {
200204
return this;
201205
}
202206

207+
/** Specifies additional indexed information in result of list workflow. */
208+
public Builder setSearchAttributes(Map<String, Object> searchAttributes) {
209+
this.searchAttributes = searchAttributes;
210+
return this;
211+
}
212+
203213
public ChildWorkflowOptions build() {
204214
return new ChildWorkflowOptions(
205215
domain,
@@ -211,7 +221,8 @@ public ChildWorkflowOptions build() {
211221
retryOptions,
212222
childPolicy,
213223
cronSchedule,
214-
memo);
224+
memo,
225+
searchAttributes);
215226
}
216227

217228
public ChildWorkflowOptions validateAndBuildWithDefaults() {
@@ -225,7 +236,8 @@ public ChildWorkflowOptions validateAndBuildWithDefaults() {
225236
retryOptions,
226237
childPolicy,
227238
cronSchedule,
228-
memo);
239+
memo,
240+
searchAttributes);
229241
}
230242
}
231243

@@ -249,6 +261,8 @@ public ChildWorkflowOptions validateAndBuildWithDefaults() {
249261

250262
private final Map<String, Object> memo;
251263

264+
private final Map<String, Object> searchAttributes;
265+
252266
private ChildWorkflowOptions(
253267
String domain,
254268
String workflowId,
@@ -259,7 +273,8 @@ private ChildWorkflowOptions(
259273
RetryOptions retryOptions,
260274
ChildPolicy childPolicy,
261275
String cronSchedule,
262-
Map<String, Object> memo) {
276+
Map<String, Object> memo,
277+
Map<String, Object> searchAttributes) {
263278
this.domain = domain;
264279
this.workflowId = workflowId;
265280
this.workflowIdReusePolicy = workflowIdReusePolicy;
@@ -270,6 +285,7 @@ private ChildWorkflowOptions(
270285
this.childPolicy = childPolicy;
271286
this.cronSchedule = cronSchedule;
272287
this.memo = memo;
288+
this.searchAttributes = searchAttributes;
273289
}
274290

275291
public String getDomain() {
@@ -312,6 +328,10 @@ public Map<String, Object> getMemo() {
312328
return memo;
313329
}
314330

331+
public Map<String, Object> getSearchAttributes() {
332+
return searchAttributes;
333+
}
334+
315335
@Override
316336
public boolean equals(Object o) {
317337
if (this == o) return true;
@@ -327,7 +347,8 @@ public boolean equals(Object o) {
327347
&& Objects.equals(retryOptions, that.retryOptions)
328348
&& childPolicy == that.childPolicy
329349
&& Objects.equals(cronSchedule, that.cronSchedule)
330-
&& Objects.equals(memo, that.memo);
350+
&& Objects.equals(memo, that.memo)
351+
&& Objects.equals(searchAttributes, that.searchAttributes);
331352
}
332353

333354
@Override
@@ -342,7 +363,8 @@ public int hashCode() {
342363
retryOptions,
343364
childPolicy,
344365
cronSchedule,
345-
memo);
366+
memo,
367+
searchAttributes);
346368
}
347369

348370
@Override
@@ -372,6 +394,9 @@ public String toString() {
372394
+ ", memo='"
373395
+ memo
374396
+ '\''
397+
+ ", searchAttributes='"
398+
+ searchAttributes
399+
+ '\''
375400
+ '}';
376401
}
377402
}

0 commit comments

Comments
 (0)