Skip to content

Commit 5187ef6

Browse files
c-warrenc-warren
authored
feat: complete ActiveClusterSelectionPolicy wiring for start and signal-with-start (#1089)
* fix: map ActiveClusterSelectionPolicy and CronOverlapPolicy in HistoryMapper started-event attributes
1 parent 7359dff commit 5187ef6

10 files changed

Lines changed: 364 additions & 8 deletions

File tree

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

Lines changed: 37 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import com.cronutils.model.definition.CronDefinitionBuilder;
2626
import com.cronutils.parser.CronParser;
2727
import com.google.common.base.Strings;
28+
import com.uber.cadence.ActiveClusterSelectionPolicy;
2829
import com.uber.cadence.WorkflowIdReusePolicy;
2930
import com.uber.cadence.common.CronSchedule;
3031
import com.uber.cadence.common.MethodRetry;
@@ -67,6 +68,7 @@ public static WorkflowOptions merge(
6768
.setSearchAttributes(o.getSearchAttributes())
6869
.setContextPropagators(o.getContextPropagators())
6970
.setDelayStart(o.delayStart)
71+
.setActiveClusterSelectionPolicy(o.activeClusterSelectionPolicy)
7072
.validateBuildWithDefaults();
7173
}
7274

@@ -94,6 +96,8 @@ public static final class Builder {
9496

9597
private Duration delayStart;
9698

99+
private ActiveClusterSelectionPolicy activeClusterSelectionPolicy;
100+
97101
public Builder() {}
98102

99103
public Builder(WorkflowOptions o) {
@@ -111,6 +115,7 @@ public Builder(WorkflowOptions o) {
111115
this.searchAttributes = o.searchAttributes;
112116
this.contextPropagators = o.contextPropagators;
113117
this.delayStart = o.delayStart;
118+
this.activeClusterSelectionPolicy = o.activeClusterSelectionPolicy;
114119
}
115120

116121
/**
@@ -223,6 +228,18 @@ public Builder setDelayStart(Duration delayStart) {
223228
return this;
224229
}
225230

231+
/**
232+
* Sets the active cluster selection policy for an active-active domain. The cluster attribute
233+
* is a scope/name pair, for example scope {@code "location"} and name {@code "lisbon"}. The
234+
* workflow follows that attribute's failover behavior as configured on the domain. This option
235+
* is only meaningful for active-active domains.
236+
*/
237+
public Builder setActiveClusterSelectionPolicy(
238+
ActiveClusterSelectionPolicy activeClusterSelectionPolicy) {
239+
this.activeClusterSelectionPolicy = activeClusterSelectionPolicy;
240+
return this;
241+
}
242+
226243
public WorkflowOptions build() {
227244
return new WorkflowOptions(
228245
workflowId,
@@ -235,7 +252,8 @@ public WorkflowOptions build() {
235252
memo,
236253
searchAttributes,
237254
contextPropagators,
238-
delayStart);
255+
delayStart,
256+
activeClusterSelectionPolicy);
239257
}
240258

241259
/**
@@ -290,7 +308,8 @@ public WorkflowOptions validateBuildWithDefaults() {
290308
memo,
291309
searchAttributes,
292310
contextPropagators,
293-
delayStart);
311+
delayStart,
312+
activeClusterSelectionPolicy);
294313
}
295314
}
296315

@@ -316,6 +335,8 @@ public WorkflowOptions validateBuildWithDefaults() {
316335

317336
private Duration delayStart;
318337

338+
private ActiveClusterSelectionPolicy activeClusterSelectionPolicy;
339+
319340
private WorkflowOptions(
320341
String workflowId,
321342
WorkflowIdReusePolicy workflowIdReusePolicy,
@@ -327,7 +348,8 @@ private WorkflowOptions(
327348
Map<String, Object> memo,
328349
Map<String, Object> searchAttributes,
329350
List<ContextPropagator> contextPropagators,
330-
Duration delayStart) {
351+
Duration delayStart,
352+
ActiveClusterSelectionPolicy activeClusterSelectionPolicy) {
331353
this.workflowId = workflowId;
332354
this.workflowIdReusePolicy = workflowIdReusePolicy;
333355
this.executionStartToCloseTimeout = executionStartToCloseTimeout;
@@ -339,6 +361,7 @@ private WorkflowOptions(
339361
this.searchAttributes = searchAttributes;
340362
this.contextPropagators = contextPropagators;
341363
this.delayStart = delayStart;
364+
this.activeClusterSelectionPolicy = activeClusterSelectionPolicy;
342365
}
343366

344367
public String getWorkflowId() {
@@ -385,6 +408,10 @@ public Duration getDelayStart() {
385408
return delayStart;
386409
}
387410

411+
public ActiveClusterSelectionPolicy getActiveClusterSelectionPolicy() {
412+
return activeClusterSelectionPolicy;
413+
}
414+
388415
@Override
389416
public boolean equals(Object o) {
390417
if (this == o) return true;
@@ -400,7 +427,8 @@ public boolean equals(Object o) {
400427
&& Objects.equals(memo, that.memo)
401428
&& Objects.equals(searchAttributes, that.searchAttributes)
402429
&& Objects.equals(contextPropagators, that.contextPropagators)
403-
&& Objects.equals(delayStart, that.delayStart);
430+
&& Objects.equals(delayStart, that.delayStart)
431+
&& Objects.equals(activeClusterSelectionPolicy, that.activeClusterSelectionPolicy);
404432
}
405433

406434
@Override
@@ -416,7 +444,8 @@ public int hashCode() {
416444
memo,
417445
searchAttributes,
418446
contextPropagators,
419-
delayStart);
447+
delayStart,
448+
activeClusterSelectionPolicy);
420449
}
421450

422451
@Override
@@ -450,6 +479,9 @@ public String toString() {
450479
+ ", delayStart='"
451480
+ delayStart
452481
+ '\''
482+
+ ", activeClusterSelectionPolicy='"
483+
+ activeClusterSelectionPolicy
484+
+ '\''
453485
+ '}';
454486
}
455487
}

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

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
package com.uber.cadence.internal.common;
1919

20+
import com.uber.cadence.ActiveClusterSelectionPolicy;
2021
import com.uber.cadence.WorkflowIdReusePolicy;
2122
import com.uber.cadence.WorkflowType;
2223
import com.uber.cadence.client.WorkflowOptions;
@@ -56,6 +57,8 @@ public final class StartWorkflowExecutionParameters {
5657

5758
private Duration delayStart;
5859

60+
private ActiveClusterSelectionPolicy activeClusterSelectionPolicy;
61+
5962
/**
6063
* Returns the value of the WorkflowId property for this object.
6164
*
@@ -317,6 +320,15 @@ public Duration getDelayStart() {
317320
return delayStart;
318321
}
319322

323+
public ActiveClusterSelectionPolicy getActiveClusterSelectionPolicy() {
324+
return activeClusterSelectionPolicy;
325+
}
326+
327+
public void setActiveClusterSelectionPolicy(
328+
ActiveClusterSelectionPolicy activeClusterSelectionPolicy) {
329+
this.activeClusterSelectionPolicy = activeClusterSelectionPolicy;
330+
}
331+
320332
public StartWorkflowExecutionParameters withRetryParameters(RetryParameters retryParameters) {
321333
this.retryParameters = retryParameters;
322334
return this;
@@ -352,6 +364,7 @@ public static StartWorkflowExecutionParameters fromWorkflowOptions(WorkflowOptio
352364
if (options.getCronSchedule() != null) {
353365
parameters.setCronSchedule(options.getCronSchedule());
354366
}
367+
parameters.setActiveClusterSelectionPolicy(options.getActiveClusterSelectionPolicy());
355368
return parameters;
356369
}
357370

@@ -396,6 +409,9 @@ public String toString() {
396409
+ ", delayStart='"
397410
+ delayStart
398411
+ '\''
412+
+ ", activeClusterSelectionPolicy='"
413+
+ activeClusterSelectionPolicy
414+
+ '\''
399415
+ '}';
400416
}
401417

@@ -416,7 +432,8 @@ public boolean equals(Object o) {
416432
&& Objects.equals(memo, that.memo)
417433
&& Objects.equals(searchAttributes, that.searchAttributes)
418434
&& Objects.equals(context, that.context)
419-
&& Objects.equals(delayStart, that.delayStart);
435+
&& Objects.equals(delayStart, that.delayStart)
436+
&& Objects.equals(activeClusterSelectionPolicy, that.activeClusterSelectionPolicy);
420437
}
421438

422439
@Override
@@ -434,7 +451,8 @@ public int hashCode() {
434451
memo,
435452
searchAttributes,
436453
context,
437-
delayStart);
454+
delayStart,
455+
activeClusterSelectionPolicy);
438456
result = 31 * result + Arrays.hashCode(input);
439457
return result;
440458
}
@@ -456,6 +474,7 @@ public StartWorkflowExecutionParameters copy() {
456474
result.setSearchAttributes(searchAttributes);
457475
result.setContext(context);
458476
result.setDelayStart(delayStart);
477+
result.setActiveClusterSelectionPolicy(activeClusterSelectionPolicy);
459478
return result;
460479
}
461480
}

‎src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/HistoryMapper.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.cancelExternalWorkflowExecutionFailedCause;
2020
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.childWorkflowExecutionFailedCause;
2121
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.continueAsNewInitiator;
22+
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.cronOverlapPolicy;
2223
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.decisionTaskFailedCause;
2324
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.decisionTaskTimedOutCause;
2425
import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.parentClosePolicy;
@@ -28,6 +29,7 @@
2829
import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.byteStringToArray;
2930
import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.durationToSeconds;
3031
import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.timeToUnixNano;
32+
import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.activeClusterSelectionPolicy;
3133
import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.activityType;
3234
import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.externalInitiatedId;
3335
import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.externalWorkflowExecution;
@@ -1090,6 +1092,9 @@ static com.uber.cadence.TimerStartedEventAttributes timerStartedEventAttributes(
10901092
res.setSearchAttributes(searchAttributes(t.getSearchAttributes()));
10911093
res.setPrevAutoResetPoints(resetPoints(t.getPrevAutoResetPoints()));
10921094
res.setHeader(header(t.getHeader()));
1095+
res.setActiveClusterSelectionPolicy(
1096+
activeClusterSelectionPolicy(t.getActiveClusterSelectionPolicy()));
1097+
res.setCronOverlapPolicy(cronOverlapPolicy(t.getCronOverlapPolicy()));
10931098
return res;
10941099
}
10951100

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,9 @@ private StartWorkflowExecutionRequest getStartRequest(
261261
if (startParameters.getDelayStart() != null) {
262262
request.setDelayStartSeconds((int) startParameters.getDelayStart().getSeconds());
263263
}
264+
if (startParameters.getActiveClusterSelectionPolicy() != null) {
265+
request.setActiveClusterSelectionPolicy(startParameters.getActiveClusterSelectionPolicy());
266+
}
264267

265268
return request;
266269
}
@@ -487,6 +490,9 @@ private SignalWithStartWorkflowExecutionRequest createSignalWithStartRequest(
487490
if (startParameters.getDelayStart() != null) {
488491
request.setDelayStartSeconds((int) startParameters.getDelayStart().getSeconds());
489492
}
493+
if (startParameters.getActiveClusterSelectionPolicy() != null) {
494+
request.setActiveClusterSelectionPolicy(startParameters.getActiveClusterSelectionPolicy());
495+
}
490496
return request;
491497
}
492498

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -554,6 +554,7 @@ private static void startWorkflow(
554554
a.setMemo(request.getMemo());
555555
a.setSearchAttributes((request.getSearchAttributes()));
556556
a.setHeader(request.getHeader());
557+
a.setActiveClusterSelectionPolicy(request.getActiveClusterSelectionPolicy());
557558
Optional<TestWorkflowMutableState> parent = ctx.getWorkflowMutableState().getParent();
558559
if (parent.isPresent()) {
559560
ExecutionId parentExecutionId = parent.get().getExecutionId();

‎src/test/java/com/uber/cadence/client/WorkflowOptionsTest.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
package com.uber.cadence.client;
1919

20+
import com.uber.cadence.ActiveClusterSelectionPolicy;
21+
import com.uber.cadence.ClusterAttribute;
2022
import com.uber.cadence.WorkflowIdReusePolicy;
2123
import com.uber.cadence.common.CronSchedule;
2224
import com.uber.cadence.common.MethodRetry;
@@ -192,6 +194,73 @@ public void testInvalidCronScheduleAnnotation() throws NoSuchMethodException {
192194
Assert.fail("invalid cron schedule not caught");
193195
}
194196

197+
private static ActiveClusterSelectionPolicy testPolicy() {
198+
return new ActiveClusterSelectionPolicy()
199+
.setClusterAttribute(new ClusterAttribute().setScope("location").setName("lisbon"));
200+
}
201+
202+
private static WorkflowOptions.Builder optionsWithPolicy(ActiveClusterSelectionPolicy policy) {
203+
return new WorkflowOptions.Builder()
204+
.setTaskList("foo")
205+
.setExecutionStartToCloseTimeout(Duration.ofSeconds(321))
206+
.setActiveClusterSelectionPolicy(policy);
207+
}
208+
209+
@Test
210+
public void testActiveClusterSelectionPolicySetOnBuilder() {
211+
ActiveClusterSelectionPolicy policy = testPolicy();
212+
Assert.assertEquals(
213+
policy, optionsWithPolicy(policy).build().getActiveClusterSelectionPolicy());
214+
}
215+
216+
@Test
217+
public void testActiveClusterSelectionPolicyDefaultsToNull() {
218+
Assert.assertNull(new WorkflowOptions.Builder().build().getActiveClusterSelectionPolicy());
219+
Assert.assertNull(
220+
optionsWithPolicy(null).validateBuildWithDefaults().getActiveClusterSelectionPolicy());
221+
}
222+
223+
@Test
224+
public void testActiveClusterSelectionPolicyKeptByCopyConstructor() {
225+
ActiveClusterSelectionPolicy policy = testPolicy();
226+
WorkflowOptions o = optionsWithPolicy(policy).build();
227+
Assert.assertEquals(
228+
policy, new WorkflowOptions.Builder(o).build().getActiveClusterSelectionPolicy());
229+
}
230+
231+
@Test
232+
public void testActiveClusterSelectionPolicyKeptByMergeWithAnnotation()
233+
throws NoSuchMethodException {
234+
ActiveClusterSelectionPolicy policy = testPolicy();
235+
WorkflowMethod a =
236+
WorkflowOptionsTest.class
237+
.getMethod("defaultWorkflowOptions")
238+
.getAnnotation(WorkflowMethod.class);
239+
Assert.assertEquals(
240+
policy,
241+
WorkflowOptions.merge(a, null, null, optionsWithPolicy(policy).build())
242+
.getActiveClusterSelectionPolicy());
243+
}
244+
245+
@Test
246+
public void testActiveClusterSelectionPolicyKeptByMergeWithoutAnnotation() {
247+
ActiveClusterSelectionPolicy policy = testPolicy();
248+
Assert.assertEquals(
249+
policy,
250+
WorkflowOptions.merge(null, null, null, optionsWithPolicy(policy).build())
251+
.getActiveClusterSelectionPolicy());
252+
}
253+
254+
@Test
255+
public void testActiveClusterSelectionPolicyConsideredByEqualsAndHashCode() {
256+
WorkflowOptions withPolicy = optionsWithPolicy(testPolicy()).build();
257+
WorkflowOptions samePolicy = optionsWithPolicy(testPolicy()).build();
258+
WorkflowOptions withoutPolicy = optionsWithPolicy(null).build();
259+
Assert.assertEquals(withPolicy, samePolicy);
260+
Assert.assertEquals(withPolicy.hashCode(), samePolicy.hashCode());
261+
Assert.assertNotEquals(withPolicy, withoutPolicy);
262+
}
263+
195264
private Map<String, Object> getTestMemo() {
196265
Map<String, Object> memo = new HashMap<>();
197266
memo.put("testKey", "testObject");

0 commit comments

Comments
 (0)