From 0d686c7e8e00440ffb5aaa139e41c2727a1bfd6b Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Fri, 26 Jun 2026 13:36:24 -0700 Subject: [PATCH 01/17] feat: add client-facing schedule model types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds com.uber.cadence.client.schedule — public Java API types for the Schedules feature using Instant/Duration instead of raw nanoseconds. Types added: - ScheduleSpec — cron expression, start/end time, jitter - ScheduleAction — wraps StartWorkflowAction for triggered workflows - SchedulePolicies — overlap, catch-up, and pause-on-failure behavior - ScheduleState — pause state (paused flag, reason, time, actor) - ScheduleInfo — runtime stats (runs, backfills, missed/skipped) - ScheduleDescription — full describe response; mutable for update calls - ScheduleListEntry — lightweight list entry from visibility store - ListSchedulesResult — paginated list response with next-page token - ScheduleOverlapPolicy — SKIP_NEW, BUFFER, CONCURRENT, CANCEL_PREVIOUS, TERMINATE_PREVIOUS - ScheduleCatchUpPolicy — SKIP, ONE, ALL --- .../client/schedule/ListSchedulesResult.java | 88 ++++++++++++++++++ .../client/schedule/ScheduleListEntry.java | 89 +++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java create mode 100644 src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java diff --git a/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java b/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java new file mode 100644 index 000000000..53a9151a5 --- /dev/null +++ b/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java @@ -0,0 +1,88 @@ +/** + * Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + *

Modifications copyright (C) 2017 Uber Technologies, Inc. + * + *

Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file + * except in compliance with the License. A copy of the License is located at + * + *

http://aws.amazon.com/apache2.0 + * + *

or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package com.uber.cadence.client.schedule; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * Result of a {@link com.uber.cadence.client.WorkflowClient#listSchedules} call. + * + *

To page through all schedules: + * + * 

{@code
+ * byte[] token = null;
+ * do {
+ *     ListSchedulesResult result = client.listSchedules(100, token);
+ *     for (ScheduleListEntry entry : result.getSchedules()) {
+ *         System.out.println(entry.getScheduleId());
+ *     }
+ *     token = result.getNextPageToken();
+ * } while (token != null && token.length > 0);
+ * }
+ */ +public final class ListSchedulesResult { + + private final List schedules; + private final byte[] nextPageToken; + + public ListSchedulesResult(List schedules, byte[] nextPageToken) { + this.schedules = + schedules == null + ? Collections.emptyList() + : Collections.unmodifiableList(new ArrayList<>(schedules)); + this.nextPageToken = nextPageToken == null ? null : nextPageToken.clone(); + } + + /** Schedules in this page. May be empty on the last page. */ + public List getSchedules() { + return schedules; + } + + /** + * Opaque token to pass as {@code nextPageToken} on the next {@code listSchedules} call. {@code + * null} or empty byte array means no more pages. + */ + public byte[] getNextPageToken() { + return nextPageToken == null ? null : nextPageToken.clone(); + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof ListSchedulesResult)) return false; + ListSchedulesResult that = (ListSchedulesResult) o; + return Objects.equals(schedules, that.schedules) + && Arrays.equals(nextPageToken, that.nextPageToken); + } + + @Override + public int hashCode() { + return Objects.hash(schedules, Arrays.hashCode(nextPageToken)); + } + + @Override + public String toString() { + return "ListSchedulesResult{" + + "schedules=" + + schedules + + ", nextPageToken.length=" + + (nextPageToken == null ? "null" : nextPageToken.length) + + '}'; + } +} diff --git a/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java b/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java new file mode 100644 index 000000000..94ef7e254 --- /dev/null +++ b/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java @@ -0,0 +1,89 @@ +/** + * Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + *

Modifications copyright (C) 2017 Uber Technologies, Inc. + * + *

Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file + * except in compliance with the License. A copy of the License is located at + * + *

http://aws.amazon.com/apache2.0 + * + *

or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ +package com.uber.cadence.client.schedule; + +import java.util.Objects; + +/** + * A single entry returned by {@link com.uber.cadence.client.WorkflowClient#listSchedules}. + * + *

Contains only the data available from the visibility store. For full detail including policies + * and runtime info, call {@link com.uber.cadence.client.WorkflowClient#describeSchedule}. + */ +public final class ScheduleListEntry { + + private final String scheduleId; + private final String workflowType; + private final ScheduleState state; + private final String cronExpression; + + public ScheduleListEntry( + String scheduleId, String workflowType, ScheduleState state, String cronExpression) { + this.scheduleId = scheduleId; + this.workflowType = workflowType; + this.state = state; + this.cronExpression = cronExpression; + } + + /** The unique schedule identifier within the domain. */ + public String getScheduleId() { + return scheduleId; + } + + /** Workflow type configured in the schedule action. */ + public String getWorkflowType() { + return workflowType; + } + + /** Current pause state of the schedule. */ + public ScheduleState getState() { + return state; + } + + /** Cron expression configured in the spec. */ + public String getCronExpression() { + return cronExpression; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (!(o instanceof ScheduleListEntry)) return false; + ScheduleListEntry that = (ScheduleListEntry) o; + return Objects.equals(scheduleId, that.scheduleId) + && Objects.equals(workflowType, that.workflowType) + && Objects.equals(state, that.state) + && Objects.equals(cronExpression, that.cronExpression); + } + + @Override + public int hashCode() { + return Objects.hash(scheduleId, workflowType, state, cronExpression); + } + + @Override + public String toString() { + return "ScheduleListEntry{" + + "scheduleId='" + + scheduleId + + "', workflowType='" + + workflowType + + "', cronExpression='" + + cronExpression + + "', paused=" + + (state != null && state.isPaused()) + + '}'; + } +} From eb0a1354a6131afb5a3cfa83c32c8a6aa85be876 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Wed, 1 Jul 2026 15:10:28 -0700 Subject: [PATCH 02/17] fix(schedule): correct type and design issues in schedule service contract types - Change ScheduleBackfill start/end time from ZonedDateTime to Instant; the Thrift wire format stores epoch nanoseconds (timezone-unaware), so Instant is the correct representation and is consistent with ScheduleSpec - Make ScheduleDescription fully immutable: remove all setters (setSpec, setAction, setPolicies, setMemo, setSearchAttributes), make fields final, and update class Javadoc to document the correct read-modify-write pattern via UpdateScheduleRequest - Fix stale WorkflowClient Javadoc @link references across all client/schedule types (ScheduleDescription, ScheduleInfo, ScheduleListEntry, ScheduleState, ScheduleOverlapPolicy, SchedulePolicies, ListSchedulesResult); all schedule methods live on ScheduleClient, not WorkflowClient - Soften getPauseReason() Javadoc: "Non-null only when isPaused() is true" changed to "null when not paused" to avoid implied contract beyond what the field stores - Remove ScheduleTypesTest tests that exercised removed setters (scheduleDescription_setters, setSpec_rejectsNull, setAction_rejectsNull, setPolicies_rejectsNull); rewrite notEqualOnDifferentSpec and toString tests to construct ScheduleDescription directly without setters --- .../com/uber/cadence/client/schedule/ListSchedulesResult.java | 2 +- .../com/uber/cadence/client/schedule/ScheduleListEntry.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java b/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java index 53a9151a5..b48c407b3 100644 --- a/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java +++ b/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java @@ -21,7 +21,7 @@ import java.util.Objects; /** - * Result of a {@link com.uber.cadence.client.WorkflowClient#listSchedules} call. + * Result of a {@link com.uber.cadence.client.ScheduleClient#listSchedules} call. * *

To page through all schedules: * diff --git a/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java b/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java index 94ef7e254..7281cda6c 100644 --- a/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java +++ b/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java @@ -17,10 +17,10 @@ import java.util.Objects; /** - * A single entry returned by {@link com.uber.cadence.client.WorkflowClient#listSchedules}. + * A single entry returned by {@link com.uber.cadence.client.ScheduleClient#listSchedules}. * *

Contains only the data available from the visibility store. For full detail including policies - * and runtime info, call {@link com.uber.cadence.client.WorkflowClient#describeSchedule}. + * and runtime info, call {@link com.uber.cadence.client.ScheduleClient#describeSchedule}. */ public final class ScheduleListEntry { From 618ad15e20d25ee9c6ec25aa562df1ca56a801cb Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Wed, 1 Jul 2026 15:17:55 -0700 Subject: [PATCH 03/17] fix(schedule): add equals/hashCode to ScheduleBackfill and harden ScheduleDescription immutability ScheduleBackfill is an immutable value type but was missing equals() and hashCode(), unlike every other schedule model type. Add value-based implementations over startTime, endTime, and overlapPolicy. ScheduleDescription.getMemo() and getSearchAttributes() were returning the caller-supplied map by reference, allowing post-construction mutation and defeating the immutability established by making the fields final. Defensive-copy and null-normalize both maps in the constructor (null -> emptyMap, non-null -> unmodifiableMap of a new HashMap), matching the pattern used by ScheduleInfo and ListSchedulesResult. Getters are now guaranteed non-null. Add tests covering the new equals/hashCode contract for ScheduleBackfill and the null-normalization, unmodifiability, and non-aliasing guarantees for ScheduleDescription maps. --- .../client/schedule/ScheduleTypesTest.java | 99 +++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java b/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java index a57726a05..b6ed42c08 100644 --- a/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java +++ b/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java @@ -529,6 +529,33 @@ public void scheduleDescription_nullMemoNormalizesToEmpty() { assertTrue(minimalDescription().getSearchAttributes().isEmpty()); } + @Test + public void listSchedulesResult_tokenDefensiveCopy() { + byte[] token = {7, 8, 9}; + ListSchedulesResult result = new ListSchedulesResult(Collections.emptyList(), token); + token[0] = 99; + assertEquals(7, result.getNextPageToken()[0]); + } + + @Test + public void listSchedulesResult_getTokenDoesNotExposeInternal() { + ListSchedulesResult result = + new ListSchedulesResult(Collections.emptyList(), new byte[] {1, 2, 3}); + result.getNextPageToken()[0] = 99; + assertEquals(1, result.getNextPageToken()[0]); + } + + @Test + public void listSchedulesResult_nullTokenRoundTrips() { + assertNull(new ListSchedulesResult(Collections.emptyList(), null).getNextPageToken()); + } + + @Test + public void listSchedulesResult_nullSchedulesNormalizesToEmpty() { + assertNotNull(new ListSchedulesResult(null, null).getSchedules()); + assertTrue(new ListSchedulesResult(null, null).getSchedules().isEmpty()); + } + @Test(expected = UnsupportedOperationException.class) public void scheduleDescription_memoIsUnmodifiable() { Map memo = new HashMap<>(); @@ -640,6 +667,78 @@ public void scheduleBackfill_toString() { assertTrue(s.contains("BUFFER")); } + @Test + public void scheduleBackfill_twoArgGetters() { + Instant start = Instant.ofEpochSecond(100); + Instant end = Instant.ofEpochSecond(200); + ScheduleBackfill bf = new ScheduleBackfill(start, end); + + assertEquals(start, bf.getStartTime()); + assertEquals(end, bf.getEndTime()); + assertNull(bf.getOverlapPolicy()); + } + + @Test + public void scheduleBackfill_threeArgGetters() { + Instant start = Instant.ofEpochSecond(100); + Instant end = Instant.ofEpochSecond(200); + ScheduleBackfill bf = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.BUFFER); + + assertEquals(start, bf.getStartTime()); + assertEquals(end, bf.getEndTime()); + assertEquals(ScheduleOverlapPolicy.BUFFER, bf.getOverlapPolicy()); + } + + @Test + public void scheduleBackfill_equals() { + Instant start = Instant.ofEpochSecond(100); + Instant end = Instant.ofEpochSecond(200); + ScheduleBackfill a = new ScheduleBackfill(start, end); + ScheduleBackfill b = new ScheduleBackfill(start, end); + assertEquals(a, b); + assertEquals(a.hashCode(), b.hashCode()); + } + + @Test + public void scheduleBackfill_equalsWithPolicy() { + Instant start = Instant.ofEpochSecond(100); + Instant end = Instant.ofEpochSecond(200); + ScheduleBackfill a = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW); + ScheduleBackfill b = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW); + assertEquals(a, b); + assertEquals(a.hashCode(), b.hashCode()); + } + + @Test + public void scheduleBackfill_notEqualOnDifferentPolicy() { + Instant start = Instant.ofEpochSecond(100); + Instant end = Instant.ofEpochSecond(200); + assertNotEquals( + new ScheduleBackfill(start, end), + new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW)); + } + + @Test + public void scheduleBackfill_notEqualOnDifferentStart() { + Instant end = Instant.ofEpochSecond(200); + assertNotEquals( + new ScheduleBackfill(Instant.ofEpochSecond(100), end), + new ScheduleBackfill(Instant.ofEpochSecond(101), end)); + } + + @Test + public void scheduleBackfill_toString() { + String s = + new ScheduleBackfill( + Instant.ofEpochSecond(100), + Instant.ofEpochSecond(200), + ScheduleOverlapPolicy.BUFFER) + .toString(); + assertTrue(s.contains("startTime=")); + assertTrue(s.contains("endTime=")); + assertTrue(s.contains("BUFFER")); + } + private static ScheduleAction.StartWorkflowAction.Builder minimalStartWorkflowAction() { return ScheduleAction.StartWorkflowAction.newBuilder() .setWorkflowType("MyWorkflow") From 859417f2a0c1ecbdcea25645c4a39103436aec45 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Wed, 1 Jul 2026 15:55:26 -0700 Subject: [PATCH 04/17] fix(schedule): return clean model types from ScheduleClient describeSchedule now returns ScheduleDescription instead of DescribeScheduleResponse, and listSchedules now returns ListSchedulesResult instead of ListSchedulesResponse. The client/schedule/ model layer was already documented and built to be the public API surface; returning Thrift types was inconsistent with that design. The implementation layer (PR4) will perform the Thrift-to-model mapping. --- src/main/java/com/uber/cadence/client/ScheduleClient.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/uber/cadence/client/ScheduleClient.java b/src/main/java/com/uber/cadence/client/ScheduleClient.java index 41aca3a5f..9fe43dbf0 100644 --- a/src/main/java/com/uber/cadence/client/ScheduleClient.java +++ b/src/main/java/com/uber/cadence/client/ScheduleClient.java @@ -21,11 +21,11 @@ import com.uber.cadence.CreateScheduleRequest; import com.uber.cadence.CreateScheduleResponse; import com.uber.cadence.DeleteScheduleResponse; -import com.uber.cadence.ListSchedulesResponse; import com.uber.cadence.PauseScheduleResponse; import com.uber.cadence.UnpauseScheduleResponse; import com.uber.cadence.UpdateScheduleRequest; import com.uber.cadence.UpdateScheduleResponse; +import com.uber.cadence.client.schedule.ListSchedulesResult; import com.uber.cadence.client.schedule.ScheduleDescription; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -114,5 +114,5 @@ CompletableFuture> backfillSchedule( * @param nextPageToken continuation token from a previous response, or {@code null} for the first * page */ - CompletableFuture listSchedules(int pageSize, byte[] nextPageToken); + CompletableFuture listSchedules(int pageSize, byte[] nextPageToken); } From 1cc0620a767aa9574799a0b77284552ef7f60c5e Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 10:54:51 -0700 Subject: [PATCH 05/17] refactor(schedule): replace ListSchedulesResult wrapper with ListSchedulesResponse Remove the custom ListSchedulesResult and ScheduleListEntry types; listSchedules now returns the generated ListSchedulesResponse directly. --- .../uber/cadence/client/ScheduleClient.java | 4 +- .../client/schedule/ListSchedulesResult.java | 88 ------------------ .../client/schedule/ScheduleListEntry.java | 89 ------------------- 3 files changed, 2 insertions(+), 179 deletions(-) delete mode 100644 src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java delete mode 100644 src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java diff --git a/src/main/java/com/uber/cadence/client/ScheduleClient.java b/src/main/java/com/uber/cadence/client/ScheduleClient.java index 9fe43dbf0..41aca3a5f 100644 --- a/src/main/java/com/uber/cadence/client/ScheduleClient.java +++ b/src/main/java/com/uber/cadence/client/ScheduleClient.java @@ -21,11 +21,11 @@ import com.uber.cadence.CreateScheduleRequest; import com.uber.cadence.CreateScheduleResponse; import com.uber.cadence.DeleteScheduleResponse; +import com.uber.cadence.ListSchedulesResponse; import com.uber.cadence.PauseScheduleResponse; import com.uber.cadence.UnpauseScheduleResponse; import com.uber.cadence.UpdateScheduleRequest; import com.uber.cadence.UpdateScheduleResponse; -import com.uber.cadence.client.schedule.ListSchedulesResult; import com.uber.cadence.client.schedule.ScheduleDescription; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -114,5 +114,5 @@ CompletableFuture> backfillSchedule( * @param nextPageToken continuation token from a previous response, or {@code null} for the first * page */ - CompletableFuture listSchedules(int pageSize, byte[] nextPageToken); + CompletableFuture listSchedules(int pageSize, byte[] nextPageToken); } diff --git a/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java b/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java deleted file mode 100644 index b48c407b3..000000000 --- a/src/main/java/com/uber/cadence/client/schedule/ListSchedulesResult.java +++ /dev/null @@ -1,88 +0,0 @@ -/** - * Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. - * - *

Modifications copyright (C) 2017 Uber Technologies, Inc. - * - *

Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file - * except in compliance with the License. A copy of the License is located at - * - *

http://aws.amazon.com/apache2.0 - * - *

or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ -package com.uber.cadence.client.schedule; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; -import java.util.Objects; - -/** - * Result of a {@link com.uber.cadence.client.ScheduleClient#listSchedules} call. - * - *

To page through all schedules: - * - * 

{@code
- * byte[] token = null;
- * do {
- *     ListSchedulesResult result = client.listSchedules(100, token);
- *     for (ScheduleListEntry entry : result.getSchedules()) {
- *         System.out.println(entry.getScheduleId());
- *     }
- *     token = result.getNextPageToken();
- * } while (token != null && token.length > 0);
- * }
- */ -public final class ListSchedulesResult { - - private final List schedules; - private final byte[] nextPageToken; - - public ListSchedulesResult(List schedules, byte[] nextPageToken) { - this.schedules = - schedules == null - ? Collections.emptyList() - : Collections.unmodifiableList(new ArrayList<>(schedules)); - this.nextPageToken = nextPageToken == null ? null : nextPageToken.clone(); - } - - /** Schedules in this page. May be empty on the last page. */ - public List getSchedules() { - return schedules; - } - - /** - * Opaque token to pass as {@code nextPageToken} on the next {@code listSchedules} call. {@code - * null} or empty byte array means no more pages. - */ - public byte[] getNextPageToken() { - return nextPageToken == null ? null : nextPageToken.clone(); - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (!(o instanceof ListSchedulesResult)) return false; - ListSchedulesResult that = (ListSchedulesResult) o; - return Objects.equals(schedules, that.schedules) - && Arrays.equals(nextPageToken, that.nextPageToken); - } - - @Override - public int hashCode() { - return Objects.hash(schedules, Arrays.hashCode(nextPageToken)); - } - - @Override - public String toString() { - return "ListSchedulesResult{" - + "schedules=" - + schedules - + ", nextPageToken.length=" - + (nextPageToken == null ? "null" : nextPageToken.length) - + '}'; - } -} diff --git a/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java b/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java deleted file mode 100644 index 7281cda6c..000000000 --- a/src/main/java/com/uber/cadence/client/schedule/ScheduleListEntry.java +++ /dev/null @@ -1,89 +0,0 @@ -/** - * Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. - * - *

Modifications copyright (C) 2017 Uber Technologies, Inc. - * - *

Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file - * except in compliance with the License. A copy of the License is located at - * - *

http://aws.amazon.com/apache2.0 - * - *

or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the - * specific language governing permissions and limitations under the License. - */ -package com.uber.cadence.client.schedule; - -import java.util.Objects; - -/** - * A single entry returned by {@link com.uber.cadence.client.ScheduleClient#listSchedules}. - * - *

Contains only the data available from the visibility store. For full detail including policies - * and runtime info, call {@link com.uber.cadence.client.ScheduleClient#describeSchedule}. - */ -public final class ScheduleListEntry { - - private final String scheduleId; - private final String workflowType; - private final ScheduleState state; - private final String cronExpression; - - public ScheduleListEntry( - String scheduleId, String workflowType, ScheduleState state, String cronExpression) { - this.scheduleId = scheduleId; - this.workflowType = workflowType; - this.state = state; - this.cronExpression = cronExpression; - } - - /** The unique schedule identifier within the domain. */ - public String getScheduleId() { - return scheduleId; - } - - /** Workflow type configured in the schedule action. */ - public String getWorkflowType() { - return workflowType; - } - - /** Current pause state of the schedule. */ - public ScheduleState getState() { - return state; - } - - /** Cron expression configured in the spec. */ - public String getCronExpression() { - return cronExpression; - } - - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (!(o instanceof ScheduleListEntry)) return false; - ScheduleListEntry that = (ScheduleListEntry) o; - return Objects.equals(scheduleId, that.scheduleId) - && Objects.equals(workflowType, that.workflowType) - && Objects.equals(state, that.state) - && Objects.equals(cronExpression, that.cronExpression); - } - - @Override - public int hashCode() { - return Objects.hash(scheduleId, workflowType, state, cronExpression); - } - - @Override - public String toString() { - return "ScheduleListEntry{" - + "scheduleId='" - + scheduleId - + "', workflowType='" - + workflowType - + "', cronExpression='" - + cronExpression - + "', paused=" - + (state != null && state.isPaused()) - + '}'; - } -} From 0f4d49abdf5bbb8eb6e506237866d5a00d41decb Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Wed, 1 Jul 2026 23:32:08 -0700 Subject: [PATCH 06/17] feat: implement schedule gRPC wiring and ScheduleClientImpl - Add ScheduleAPIGrpc blocking/future stubs to IGrpcServiceStubs and GrpcServiceStubs - Add schedule enum mappers (OverlapPolicy, CatchUpPolicy) to EnumMapper - Add schedule type mappers (ScheduleSpec, ScheduleAction, SchedulePolicies, ScheduleState, ScheduleInfo, ScheduleListEntry) to TypeMapper - Add schedule request mappers (create/describe/update/delete/pause/unpause/ backfill/list) to RequestMapper - Add schedule response mappers (create/describe/list) to ResponseMapper - Implement all 8 schedule methods in WorkflowServiceGrpc using ScheduleAPIGrpc - Add ScheduleClientImpl implementing ScheduleClient, mapping Thrift responses to client model types (Instant, Duration, enums) - Wire WorkflowClientInternal.scheduleClient() to return ScheduleClientImpl --- .../proto/mappers/EnumMapper.java | 76 ++++ .../proto/mappers/RequestMapper.java | 112 +++++ .../proto/mappers/ResponseMapper.java | 93 +++++ .../proto/mappers/TypeMapper.java | 165 ++++++++ .../proto/serviceclient/GrpcServiceStubs.java | 15 + .../serviceclient/IGrpcServiceStubs.java | 7 + .../internal/sync/ScheduleClientImpl.java | 390 ++++++++++++++++++ .../internal/sync/WorkflowClientInternal.java | 2 +- .../serviceclient/WorkflowServiceGrpc.java | 72 +++- 9 files changed, 923 insertions(+), 9 deletions(-) create mode 100644 src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/EnumMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/EnumMapper.java index c2179b445..e917becc8 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/EnumMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/EnumMapper.java @@ -613,4 +613,80 @@ public static com.uber.cadence.CronOverlapPolicy cronOverlapPolicy(CronOverlapPo } throw new IllegalArgumentException("unexpected enum value"); } + + public static ScheduleOverlapPolicy scheduleOverlapPolicy( + com.uber.cadence.ScheduleOverlapPolicy t) { + if (t == null) { + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_INVALID; + } + switch (t) { + case INVALID: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_INVALID; + case SKIP_NEW: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_SKIP_NEW; + case BUFFER: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_BUFFER; + case CONCURRENT: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_CONCURRENT; + case CANCEL_PREVIOUS: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_CANCEL_PREVIOUS; + case TERMINATE_PREVIOUS: + return ScheduleOverlapPolicy.SCHEDULE_OVERLAP_POLICY_TERMINATE_PREVIOUS; + } + throw new IllegalArgumentException("unexpected enum value"); + } + + public static com.uber.cadence.ScheduleOverlapPolicy scheduleOverlapPolicy( + ScheduleOverlapPolicy t) { + switch (t) { + case SCHEDULE_OVERLAP_POLICY_INVALID: + case UNRECOGNIZED: + return com.uber.cadence.ScheduleOverlapPolicy.INVALID; + case SCHEDULE_OVERLAP_POLICY_SKIP_NEW: + return com.uber.cadence.ScheduleOverlapPolicy.SKIP_NEW; + case SCHEDULE_OVERLAP_POLICY_BUFFER: + return com.uber.cadence.ScheduleOverlapPolicy.BUFFER; + case SCHEDULE_OVERLAP_POLICY_CONCURRENT: + return com.uber.cadence.ScheduleOverlapPolicy.CONCURRENT; + case SCHEDULE_OVERLAP_POLICY_CANCEL_PREVIOUS: + return com.uber.cadence.ScheduleOverlapPolicy.CANCEL_PREVIOUS; + case SCHEDULE_OVERLAP_POLICY_TERMINATE_PREVIOUS: + return com.uber.cadence.ScheduleOverlapPolicy.TERMINATE_PREVIOUS; + } + throw new IllegalArgumentException("unexpected enum value"); + } + + public static ScheduleCatchUpPolicy scheduleCatchUpPolicy( + com.uber.cadence.ScheduleCatchUpPolicy t) { + if (t == null) { + return ScheduleCatchUpPolicy.SCHEDULE_CATCH_UP_POLICY_INVALID; + } + switch (t) { + case INVALID: + return ScheduleCatchUpPolicy.SCHEDULE_CATCH_UP_POLICY_INVALID; + case SKIP: + return ScheduleCatchUpPolicy.SCHEDULE_CATCH_UP_POLICY_SKIP; + case ONE: + return ScheduleCatchUpPolicy.SCHEDULE_CATCH_UP_POLICY_ONE; + case ALL: + return ScheduleCatchUpPolicy.SCHEDULE_CATCH_UP_POLICY_ALL; + } + throw new IllegalArgumentException("unexpected enum value"); + } + + public static com.uber.cadence.ScheduleCatchUpPolicy scheduleCatchUpPolicy( + ScheduleCatchUpPolicy t) { + switch (t) { + case SCHEDULE_CATCH_UP_POLICY_INVALID: + case UNRECOGNIZED: + return com.uber.cadence.ScheduleCatchUpPolicy.INVALID; + case SCHEDULE_CATCH_UP_POLICY_SKIP: + return com.uber.cadence.ScheduleCatchUpPolicy.SKIP; + case SCHEDULE_CATCH_UP_POLICY_ONE: + return com.uber.cadence.ScheduleCatchUpPolicy.ONE; + case SCHEDULE_CATCH_UP_POLICY_ALL: + return com.uber.cadence.ScheduleCatchUpPolicy.ALL; + } + throw new IllegalArgumentException("unexpected enum value"); + } } diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java index 987d20653..d5e35615b 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java @@ -30,6 +30,7 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.newFieldMask; import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.nullToEmpty; import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.secondsToDuration; +import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.unixNanoToTime; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.activeClusterSelectionPolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.activeClusters; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.badBinaries; @@ -39,6 +40,9 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.memo; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.payload; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.retryPolicy; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleAction; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.schedulePolicies; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleSpec; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.searchAttributes; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.startTimeFilter; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.statusFilter; @@ -1011,4 +1015,112 @@ public static RefreshWorkflowTasksRequest refreshWorkflowTasksRequest( .setDomain(request.getDomain() != null ? request.getDomain() : "") .build(); } + + public static com.uber.cadence.api.v1.CreateScheduleRequest createScheduleRequest( + com.uber.cadence.CreateScheduleRequest t) { + if (t == null) { + return null; + } + com.uber.cadence.api.v1.CreateScheduleRequest.Builder b = + com.uber.cadence.api.v1.CreateScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())); + if (t.getSpec() != null) b.setSpec(scheduleSpec(t.getSpec())); + if (t.getAction() != null) b.setAction(scheduleAction(t.getAction())); + if (t.getPolicies() != null) b.setPolicies(schedulePolicies(t.getPolicies())); + if (t.getMemo() != null) b.setMemo(memo(t.getMemo())); + if (t.getSearchAttributes() != null) + b.setSearchAttributes(searchAttributes(t.getSearchAttributes())); + return b.build(); + } + + public static com.uber.cadence.api.v1.DescribeScheduleRequest describeScheduleRequest( + com.uber.cadence.DescribeScheduleRequest t) { + if (t == null) { + return null; + } + return com.uber.cadence.api.v1.DescribeScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .build(); + } + + public static com.uber.cadence.api.v1.UpdateScheduleRequest updateScheduleRequest( + com.uber.cadence.UpdateScheduleRequest t) { + if (t == null) { + return null; + } + com.uber.cadence.api.v1.UpdateScheduleRequest.Builder b = + com.uber.cadence.api.v1.UpdateScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())); + if (t.getSpec() != null) b.setSpec(scheduleSpec(t.getSpec())); + if (t.getAction() != null) b.setAction(scheduleAction(t.getAction())); + if (t.getPolicies() != null) b.setPolicies(schedulePolicies(t.getPolicies())); + if (t.getSearchAttributes() != null) + b.setSearchAttributes(searchAttributes(t.getSearchAttributes())); + return b.build(); + } + + public static com.uber.cadence.api.v1.DeleteScheduleRequest deleteScheduleRequest( + com.uber.cadence.DeleteScheduleRequest t) { + if (t == null) { + return null; + } + return com.uber.cadence.api.v1.DeleteScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .build(); + } + + public static com.uber.cadence.api.v1.PauseScheduleRequest pauseScheduleRequest( + com.uber.cadence.PauseScheduleRequest t) { + if (t == null) { + return null; + } + return com.uber.cadence.api.v1.PauseScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .setReason(nullToEmpty(t.getReason())) + .setIdentity(nullToEmpty(t.getIdentity())) + .build(); + } + + public static com.uber.cadence.api.v1.UnpauseScheduleRequest unpauseScheduleRequest( + com.uber.cadence.UnpauseScheduleRequest t) { + if (t == null) { + return null; + } + return com.uber.cadence.api.v1.UnpauseScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .setReason(nullToEmpty(t.getReason())) + .build(); + } + + public static com.uber.cadence.api.v1.BackfillScheduleRequest backfillScheduleRequest( + com.uber.cadence.BackfillScheduleRequest t) { + if (t == null) { + return null; + } + return com.uber.cadence.api.v1.BackfillScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .setStartTime(unixNanoToTime(t.getStartTimeNano())) + .setEndTime(unixNanoToTime(t.getEndTimeNano())) + .build(); + } + + public static com.uber.cadence.api.v1.ListSchedulesRequest listSchedulesRequest( + com.uber.cadence.ListSchedulesRequest t) { + if (t == null) { + return null; + } + com.uber.cadence.api.v1.ListSchedulesRequest.Builder b = + com.uber.cadence.api.v1.ListSchedulesRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setPageSize(t.getPageSize()); + if (t.getNextPageToken() != null) b.setNextPageToken(arrayToByteString(t.getNextPageToken())); + return b.build(); + } } diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/ResponseMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/ResponseMapper.java index 71fbf67cd..fbbf257ee 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/ResponseMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/ResponseMapper.java @@ -30,12 +30,20 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.describeDomainResponseArray; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.header; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.indexedValueTypeMap; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.memo; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.payload; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.pendingActivityInfoArray; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.pendingChildExecutionInfoArray; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.pendingDecisionInfo; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.pollerInfoArray; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.queryRejected; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleAction; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleInfo; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleListEntry; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.schedulePolicies; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleSpec; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.scheduleState; +import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.searchAttributes; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.supportedClientVersions; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.taskList; import static com.uber.cadence.internal.compatibility.proto.mappers.TypeMapper.taskListPartitionMetadataArray; @@ -518,4 +526,89 @@ public static com.uber.cadence.GetTaskListsByDomainResponse getTaskListsByDomain Collectors.toMap(Map.Entry::getKey, e -> describeTaskListResponse(e.getValue())))); return res; } + + public static com.uber.cadence.CreateScheduleResponse createScheduleResponse( + com.uber.cadence.api.v1.CreateScheduleResponse t) { + if (t == null) { + return null; + } + com.uber.cadence.CreateScheduleResponse res = new com.uber.cadence.CreateScheduleResponse(); + res.setScheduleId(t.getScheduleId()); + return res; + } + + public static com.uber.cadence.DescribeScheduleResponse describeScheduleResponse( + com.uber.cadence.api.v1.DescribeScheduleResponse t) { + if (t == null) { + return null; + } + com.uber.cadence.DescribeScheduleResponse res = new com.uber.cadence.DescribeScheduleResponse(); + res.setSpec(scheduleSpec(t.getSpec())); + res.setAction(scheduleAction(t.getAction())); + res.setPolicies(schedulePolicies(t.getPolicies())); + res.setState(scheduleState(t.getState())); + res.setInfo(scheduleInfo(t.getInfo())); + res.setMemo(memo(t.getMemo())); + res.setSearchAttributes(searchAttributes(t.getSearchAttributes())); + return res; + } + + public static com.uber.cadence.ListSchedulesResponse listSchedulesResponse( + com.uber.cadence.api.v1.ListSchedulesResponse t) { + if (t == null) { + return null; + } + com.uber.cadence.ListSchedulesResponse res = new com.uber.cadence.ListSchedulesResponse(); + if (t.getSchedulesCount() > 0) { + java.util.List entries = new java.util.ArrayList<>(); + for (com.uber.cadence.api.v1.ScheduleListEntry e : t.getSchedulesList()) { + entries.add(scheduleListEntry(e)); + } + res.setSchedules(entries); + } + if (t.getNextPageToken() != null && !t.getNextPageToken().isEmpty()) { + res.setNextPageToken(byteStringToArray(t.getNextPageToken())); + } + return res; + } + + public static com.uber.cadence.UpdateScheduleResponse updateScheduleResponse( + com.uber.cadence.api.v1.UpdateScheduleResponse t) { + if (t == null) { + return null; + } + return new com.uber.cadence.UpdateScheduleResponse(); + } + + public static com.uber.cadence.DeleteScheduleResponse deleteScheduleResponse( + com.uber.cadence.api.v1.DeleteScheduleResponse t) { + if (t == null) { + return null; + } + return new com.uber.cadence.DeleteScheduleResponse(); + } + + public static com.uber.cadence.PauseScheduleResponse pauseScheduleResponse( + com.uber.cadence.api.v1.PauseScheduleResponse t) { + if (t == null) { + return null; + } + return new com.uber.cadence.PauseScheduleResponse(); + } + + public static com.uber.cadence.UnpauseScheduleResponse unpauseScheduleResponse( + com.uber.cadence.api.v1.UnpauseScheduleResponse t) { + if (t == null) { + return null; + } + return new com.uber.cadence.UnpauseScheduleResponse(); + } + + public static com.uber.cadence.BackfillScheduleResponse backfillScheduleResponse( + com.uber.cadence.api.v1.BackfillScheduleResponse t) { + if (t == null) { + return null; + } + return new com.uber.cadence.BackfillScheduleResponse(); + } } diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/TypeMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/TypeMapper.java index d993f61e6..d2c4a7f76 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/TypeMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/TypeMapper.java @@ -23,6 +23,8 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.pendingActivityState; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.pendingDecisionState; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryResultType; +import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.scheduleCatchUpPolicy; +import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.scheduleOverlapPolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.taskListKind; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.workflowExecutionCloseStatus; import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.arrayToByteString; @@ -1088,4 +1090,167 @@ static com.uber.cadence.ClusterAttribute clusterAttribute(ClusterAttribute t) { attr.setName(t.getName()); return attr; } + + static ScheduleSpec scheduleSpec(com.uber.cadence.ScheduleSpec t) { + if (t == null) { + return ScheduleSpec.getDefaultInstance(); + } + return ScheduleSpec.newBuilder() + .setCronExpression(Helpers.nullToEmpty(t.getCronExpression())) + .setStartTime(unixNanoToTime(t.getStartTimeNano())) + .setEndTime(unixNanoToTime(t.getEndTimeNano())) + .setJitter(secondsToDuration(t.getJitterInSeconds())) + .build(); + } + + static com.uber.cadence.ScheduleSpec scheduleSpec(ScheduleSpec t) { + if (t == null || t == ScheduleSpec.getDefaultInstance()) { + return null; + } + com.uber.cadence.ScheduleSpec res = new com.uber.cadence.ScheduleSpec(); + res.setCronExpression(t.getCronExpression()); + res.setStartTimeNano(timeToUnixNano(t.getStartTime())); + res.setEndTimeNano(timeToUnixNano(t.getEndTime())); + res.setJitterInSeconds(durationToSeconds(t.getJitter())); + return res; + } + + static ScheduleAction scheduleAction(com.uber.cadence.ScheduleAction t) { + if (t == null || t.getStartWorkflow() == null) { + return ScheduleAction.getDefaultInstance(); + } + com.uber.cadence.ScheduleStartWorkflowAction sw = t.getStartWorkflow(); + ScheduleAction.StartWorkflowAction.Builder swb = + ScheduleAction.StartWorkflowAction.newBuilder(); + if (sw.getWorkflowType() != null) swb.setWorkflowType(workflowType(sw.getWorkflowType())); + if (sw.getTaskList() != null) swb.setTaskList(taskList(sw.getTaskList())); + if (sw.getInput() != null) swb.setInput(payload(sw.getInput())); + swb.setWorkflowIdPrefix(Helpers.nullToEmpty(sw.getWorkflowIdPrefix())); + swb.setExecutionStartToCloseTimeout( + secondsToDuration(sw.getExecutionStartToCloseTimeoutSeconds())); + swb.setTaskStartToCloseTimeout(secondsToDuration(sw.getTaskStartToCloseTimeoutSeconds())); + if (sw.getRetryPolicy() != null) swb.setRetryPolicy(retryPolicy(sw.getRetryPolicy())); + if (sw.getMemo() != null) swb.setMemo(memo(sw.getMemo())); + if (sw.getSearchAttributes() != null) + swb.setSearchAttributes(searchAttributes(sw.getSearchAttributes())); + return ScheduleAction.newBuilder().setStartWorkflow(swb.build()).build(); + } + + static com.uber.cadence.ScheduleAction scheduleAction(ScheduleAction t) { + if (t == null || !t.hasStartWorkflow()) { + return null; + } + ScheduleAction.StartWorkflowAction proto = t.getStartWorkflow(); + com.uber.cadence.ScheduleStartWorkflowAction sw = + new com.uber.cadence.ScheduleStartWorkflowAction(); + sw.setWorkflowType(workflowType(proto.getWorkflowType())); + sw.setTaskList(taskList(proto.getTaskList())); + sw.setInput(payload(proto.getInput())); + sw.setWorkflowIdPrefix(proto.getWorkflowIdPrefix()); + sw.setExecutionStartToCloseTimeoutSeconds( + durationToSeconds(proto.getExecutionStartToCloseTimeout())); + sw.setTaskStartToCloseTimeoutSeconds(durationToSeconds(proto.getTaskStartToCloseTimeout())); + sw.setRetryPolicy(retryPolicy(proto.getRetryPolicy())); + sw.setMemo(memo(proto.getMemo())); + sw.setSearchAttributes(searchAttributes(proto.getSearchAttributes())); + com.uber.cadence.ScheduleAction action = new com.uber.cadence.ScheduleAction(); + action.setStartWorkflow(sw); + return action; + } + + static SchedulePolicies schedulePolicies(com.uber.cadence.SchedulePolicies t) { + if (t == null) { + return SchedulePolicies.getDefaultInstance(); + } + return SchedulePolicies.newBuilder() + .setOverlapPolicy(scheduleOverlapPolicy(t.getOverlapPolicy())) + .setCatchUpPolicy(scheduleCatchUpPolicy(t.getCatchUpPolicy())) + .setCatchUpWindow(secondsToDuration(t.getCatchUpWindowInSeconds())) + .setPauseOnFailure(t.isPauseOnFailure()) + .setBufferLimit(t.getBufferLimit()) + .setConcurrencyLimit(t.getConcurrencyLimit()) + .build(); + } + + static com.uber.cadence.SchedulePolicies schedulePolicies(SchedulePolicies t) { + if (t == null || t == SchedulePolicies.getDefaultInstance()) { + return null; + } + com.uber.cadence.SchedulePolicies res = new com.uber.cadence.SchedulePolicies(); + res.setOverlapPolicy(scheduleOverlapPolicy(t.getOverlapPolicy())); + res.setCatchUpPolicy(scheduleCatchUpPolicy(t.getCatchUpPolicy())); + res.setCatchUpWindowInSeconds(durationToSeconds(t.getCatchUpWindow())); + res.setPauseOnFailure(t.getPauseOnFailure()); + res.setBufferLimit(t.getBufferLimit()); + res.setConcurrencyLimit(t.getConcurrencyLimit()); + return res; + } + + static com.uber.cadence.SchedulePauseInfo schedulePauseInfo(SchedulePauseInfo t) { + if (t == null || t == SchedulePauseInfo.getDefaultInstance()) { + return null; + } + com.uber.cadence.SchedulePauseInfo res = new com.uber.cadence.SchedulePauseInfo(); + res.setReason(t.getReason()); + res.setPausedTimeNano(timeToUnixNano(t.getPausedAt())); + res.setPausedBy(t.getPausedBy()); + return res; + } + + static com.uber.cadence.ScheduleState scheduleState(ScheduleState t) { + if (t == null || t == ScheduleState.getDefaultInstance()) { + return null; + } + com.uber.cadence.ScheduleState res = new com.uber.cadence.ScheduleState(); + res.setPaused(t.getPaused()); + res.setPauseInfo(schedulePauseInfo(t.getPauseInfo())); + return res; + } + + static com.uber.cadence.BackfillInfo backfillInfo(BackfillInfo t) { + if (t == null || t == BackfillInfo.getDefaultInstance()) { + return null; + } + com.uber.cadence.BackfillInfo res = new com.uber.cadence.BackfillInfo(); + res.setBackfillId(t.getBackfillId()); + res.setStartTimeNano(timeToUnixNano(t.getStartTime())); + res.setEndTimeNano(timeToUnixNano(t.getEndTime())); + res.setRunsCompleted(t.getRunsCompleted()); + res.setRunsTotal(t.getRunsTotal()); + return res; + } + + static com.uber.cadence.ScheduleInfo scheduleInfo(ScheduleInfo t) { + if (t == null || t == ScheduleInfo.getDefaultInstance()) { + return null; + } + com.uber.cadence.ScheduleInfo res = new com.uber.cadence.ScheduleInfo(); + res.setLastRunTimeNano(timeToUnixNano(t.getLastRunTime())); + res.setNextRunTimeNano(timeToUnixNano(t.getNextRunTime())); + res.setTotalRuns(t.getTotalRuns()); + res.setCreateTimeNano(timeToUnixNano(t.getCreateTime())); + res.setLastUpdateTimeNano(timeToUnixNano(t.getLastUpdateTime())); + if (t.getOngoingBackfillsCount() > 0) { + List backfills = new ArrayList<>(); + for (BackfillInfo b : t.getOngoingBackfillsList()) { + backfills.add(backfillInfo(b)); + } + res.setOngoingBackfills(backfills); + } + res.setMissedRuns(t.getMissedRuns()); + res.setSkippedRuns(t.getSkippedRuns()); + return res; + } + + static com.uber.cadence.ScheduleListEntry scheduleListEntry(ScheduleListEntry t) { + if (t == null || t == ScheduleListEntry.getDefaultInstance()) { + return null; + } + com.uber.cadence.ScheduleListEntry res = new com.uber.cadence.ScheduleListEntry(); + res.setScheduleId(t.getScheduleId()); + res.setWorkflowType(workflowType(t.getWorkflowType())); + res.setState(scheduleState(t.getState())); + res.setCronExpression(t.getCronExpression()); + return res; + } } diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java index 6e76e5abd..f0290559e 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/GrpcServiceStubs.java @@ -22,6 +22,7 @@ import com.uber.cadence.api.v1.MetaAPIGrpc; import com.uber.cadence.api.v1.MetaAPIGrpc.MetaAPIBlockingStub; import com.uber.cadence.api.v1.MetaAPIGrpc.MetaAPIFutureStub; +import com.uber.cadence.api.v1.ScheduleAPIGrpc; import com.uber.cadence.api.v1.VisibilityAPIGrpc; import com.uber.cadence.api.v1.VisibilityAPIGrpc.VisibilityAPIBlockingStub; import com.uber.cadence.api.v1.VisibilityAPIGrpc.VisibilityAPIFutureStub; @@ -93,6 +94,8 @@ final class GrpcServiceStubs implements IGrpcServiceStubs { private final WorkflowAPIGrpc.WorkflowAPIFutureStub workflowFutureStub; private final MetaAPIGrpc.MetaAPIBlockingStub metaBlockingStub; private final MetaAPIGrpc.MetaAPIFutureStub metaFutureStub; + private final ScheduleAPIGrpc.ScheduleAPIBlockingStub scheduleBlockingStub; + private final ScheduleAPIGrpc.ScheduleAPIFutureStub scheduleFutureStub; GrpcServiceStubs(ClientOptions options) { this.options = options; @@ -146,6 +149,8 @@ final class GrpcServiceStubs implements IGrpcServiceStubs { this.workflowFutureStub = WorkflowAPIGrpc.newFutureStub(interceptedChannel); this.metaBlockingStub = MetaAPIGrpc.newBlockingStub(interceptedChannel); this.metaFutureStub = MetaAPIGrpc.newFutureStub(interceptedChannel); + this.scheduleBlockingStub = ScheduleAPIGrpc.newBlockingStub(interceptedChannel); + this.scheduleFutureStub = ScheduleAPIGrpc.newFutureStub(interceptedChannel); } private ClientInterceptor newAuthorizationInterceptor(IAuthorizationProvider provider) { @@ -412,6 +417,16 @@ public WorkflowAPIFutureStub workflowFutureStub() { return workflowFutureStub; } + @Override + public ScheduleAPIGrpc.ScheduleAPIBlockingStub scheduleBlockingStub() { + return scheduleBlockingStub; + } + + @Override + public ScheduleAPIGrpc.ScheduleAPIFutureStub scheduleFutureStub() { + return scheduleFutureStub; + } + @Override public void shutdown() { shutdownRequested.set(true); diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/IGrpcServiceStubs.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/IGrpcServiceStubs.java index b0b0da2ec..4e4656d97 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/IGrpcServiceStubs.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/serviceclient/IGrpcServiceStubs.java @@ -18,6 +18,7 @@ import com.uber.cadence.api.v1.DomainAPIGrpc; import com.uber.cadence.api.v1.MetaAPIGrpc.MetaAPIBlockingStub; import com.uber.cadence.api.v1.MetaAPIGrpc.MetaAPIFutureStub; +import com.uber.cadence.api.v1.ScheduleAPIGrpc; import com.uber.cadence.api.v1.VisibilityAPIGrpc; import com.uber.cadence.api.v1.WorkerAPIGrpc; import com.uber.cadence.api.v1.WorkflowAPIGrpc; @@ -65,6 +66,12 @@ static IGrpcServiceStubs newInstance(ClientOptions options) { /** @return Future (asynchronous) stub to workflow service. */ WorkflowAPIGrpc.WorkflowAPIFutureStub workflowFutureStub(); + /** @return Blocking (synchronous) stub to schedule service. */ + ScheduleAPIGrpc.ScheduleAPIBlockingStub scheduleBlockingStub(); + + /** @return Future (asynchronous) stub to schedule service. */ + ScheduleAPIGrpc.ScheduleAPIFutureStub scheduleFutureStub(); + /** @return Blocking (synchronous) stub to meta service. */ MetaAPIFutureStub metaFutureStub(); diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java new file mode 100644 index 000000000..437de7980 --- /dev/null +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -0,0 +1,390 @@ +/* + * Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Modifications copyright (C) 2017 Uber Technologies, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"). You may not + * use this file except in compliance with the License. A copy of the License is + * located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package com.uber.cadence.internal.sync; + +import com.uber.cadence.BackfillScheduleRequest; +import com.uber.cadence.BackfillScheduleResponse; +import com.uber.cadence.CreateScheduleRequest; +import com.uber.cadence.CreateScheduleResponse; +import com.uber.cadence.DeleteScheduleRequest; +import com.uber.cadence.DeleteScheduleResponse; +import com.uber.cadence.DescribeScheduleRequest; +import com.uber.cadence.DescribeScheduleResponse; +import com.uber.cadence.ListSchedulesRequest; +import com.uber.cadence.ListSchedulesResponse; +import com.uber.cadence.PauseScheduleRequest; +import com.uber.cadence.PauseScheduleResponse; +import com.uber.cadence.UnpauseScheduleRequest; +import com.uber.cadence.UnpauseScheduleResponse; +import com.uber.cadence.UpdateScheduleRequest; +import com.uber.cadence.UpdateScheduleResponse; +import com.uber.cadence.client.ScheduleBackfill; +import com.uber.cadence.client.ScheduleClient; +import com.uber.cadence.client.schedule.ListSchedulesResult; +import com.uber.cadence.client.schedule.ScheduleAction; +import com.uber.cadence.client.schedule.ScheduleCatchUpPolicy; +import com.uber.cadence.client.schedule.ScheduleDescription; +import com.uber.cadence.client.schedule.ScheduleInfo; +import com.uber.cadence.client.schedule.ScheduleListEntry; +import com.uber.cadence.client.schedule.ScheduleOverlapPolicy; +import com.uber.cadence.client.schedule.SchedulePolicies; +import com.uber.cadence.client.schedule.ScheduleSpec; +import com.uber.cadence.client.schedule.ScheduleState; +import com.uber.cadence.common.RetryOptions; +import com.uber.cadence.serviceclient.IWorkflowService; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +final class ScheduleClientImpl implements ScheduleClient { + + private final IWorkflowService service; + private final String domain; + + ScheduleClientImpl(IWorkflowService service, String domain) { + this.service = service; + this.domain = domain; + } + + @Override + public CompletableFuture createSchedule( + String scheduleId, CreateScheduleRequest request) { + request.setDomain(domain); + request.setScheduleId(scheduleId); + return CompletableFuture.supplyAsync( + () -> { + try { + return service.CreateSchedule(request); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture describeSchedule(String scheduleId) { + DescribeScheduleRequest request = + new DescribeScheduleRequest().setDomain(domain).setScheduleId(scheduleId); + return CompletableFuture.supplyAsync( + () -> { + try { + DescribeScheduleResponse resp = service.DescribeSchedule(request); + return toScheduleDescription(resp); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture updateSchedule( + String scheduleId, UpdateScheduleRequest request) { + request.setDomain(domain); + request.setScheduleId(scheduleId); + return CompletableFuture.supplyAsync( + () -> { + try { + return service.UpdateSchedule(request); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture deleteSchedule(String scheduleId) { + DeleteScheduleRequest request = + new DeleteScheduleRequest().setDomain(domain).setScheduleId(scheduleId); + return CompletableFuture.supplyAsync( + () -> { + try { + return service.DeleteSchedule(request); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture pauseSchedule(String scheduleId, String reason) { + PauseScheduleRequest request = + new PauseScheduleRequest().setDomain(domain).setScheduleId(scheduleId).setReason(reason); + return CompletableFuture.supplyAsync( + () -> { + try { + return service.PauseSchedule(request); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture unpauseSchedule( + String scheduleId, String reason) { + UnpauseScheduleRequest request = + new UnpauseScheduleRequest().setDomain(domain).setScheduleId(scheduleId).setReason(reason); + return CompletableFuture.supplyAsync( + () -> { + try { + return service.UnpauseSchedule(request); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + @Override + public CompletableFuture> backfillSchedule( + String scheduleId, List backfills) { + return CompletableFuture.supplyAsync( + () -> { + List results = new ArrayList<>(); + for (ScheduleBackfill bf : backfills) { + BackfillScheduleRequest request = + new BackfillScheduleRequest() + .setDomain(domain) + .setScheduleId(scheduleId) + .setStartTimeNano(bf.getStartTime().toEpochMilli() * 1_000_000L) + .setEndTimeNano(bf.getEndTime().toEpochMilli() * 1_000_000L); + if (bf.getOverlapPolicy() != null) { + request.setOverlapPolicy(toThriftOverlapPolicy(bf.getOverlapPolicy())); + } + try { + results.add(service.BackfillSchedule(request)); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + return results; + }); + } + + @Override + public CompletableFuture listSchedules(int pageSize, byte[] nextPageToken) { + ListSchedulesRequest request = + new ListSchedulesRequest() + .setDomain(domain) + .setPageSize(pageSize) + .setNextPageToken(nextPageToken); + return CompletableFuture.supplyAsync( + () -> { + try { + ListSchedulesResponse resp = service.ListSchedules(request); + return toListSchedulesResult(resp); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + private static ScheduleDescription toScheduleDescription(DescribeScheduleResponse r) { + if (r == null) { + return null; + } + return new ScheduleDescription( + toScheduleSpec(r.getSpec()), + toScheduleAction(r.getAction()), + toSchedulePolicies(r.getPolicies()), + toScheduleState(r.getState()), + toScheduleInfo(r.getInfo()), + toObjectMap(r.getMemo() != null ? r.getMemo().getFields() : null), + toObjectMap( + r.getSearchAttributes() != null ? r.getSearchAttributes().getIndexedFields() : null)); + } + + private static ScheduleSpec toScheduleSpec(com.uber.cadence.ScheduleSpec t) { + if (t == null) return null; + return ScheduleSpec.newBuilder() + .setCronExpression(t.getCronExpression()) + .setStartTime(nanosToInstant(t.getStartTimeNano())) + .setEndTime(nanosToInstant(t.getEndTimeNano())) + .setJitter(Duration.ofSeconds(t.getJitterInSeconds())) + .build(); + } + + private static ScheduleAction toScheduleAction(com.uber.cadence.ScheduleAction t) { + if (t == null || t.getStartWorkflow() == null) return null; + com.uber.cadence.ScheduleStartWorkflowAction sw = t.getStartWorkflow(); + ScheduleAction.StartWorkflowAction.Builder swb = + ScheduleAction.StartWorkflowAction.newBuilder() + .setWorkflowType(sw.getWorkflowType() != null ? sw.getWorkflowType().getName() : null) + .setTaskList(sw.getTaskList() != null ? sw.getTaskList().getName() : null) + .setInput(sw.getInput()) + .setWorkflowIdPrefix(sw.getWorkflowIdPrefix()) + .setExecutionStartToCloseTimeout( + Duration.ofSeconds(sw.getExecutionStartToCloseTimeoutSeconds())) + .setTaskStartToCloseTimeout(Duration.ofSeconds(sw.getTaskStartToCloseTimeoutSeconds())); + if (sw.getRetryPolicy() != null) { + swb.setRetryOptions(toRetryOptions(sw.getRetryPolicy())); + } + if (sw.getMemo() != null) { + swb.setMemo(toObjectMap(sw.getMemo().getFields())); + } + if (sw.getSearchAttributes() != null) { + swb.setSearchAttributes(toObjectMap(sw.getSearchAttributes().getIndexedFields())); + } + return ScheduleAction.newBuilder().setStartWorkflow(swb.build()).build(); + } + + private static SchedulePolicies toSchedulePolicies(com.uber.cadence.SchedulePolicies t) { + if (t == null) return null; + return SchedulePolicies.newBuilder() + .setOverlapPolicy(toClientOverlapPolicy(t.getOverlapPolicy())) + .setCatchUpPolicy(toClientCatchUpPolicy(t.getCatchUpPolicy())) + .setCatchUpWindow(Duration.ofSeconds(t.getCatchUpWindowInSeconds())) + .setPauseOnFailure(t.isPauseOnFailure()) + .setBufferLimit(t.getBufferLimit()) + .setConcurrencyLimit(t.getConcurrencyLimit()) + .build(); + } + + private static ScheduleState toScheduleState(com.uber.cadence.ScheduleState t) { + if (t == null) return null; + String pauseReason = null; + Instant pausedAt = null; + String pausedBy = null; + if (t.getPauseInfo() != null) { + com.uber.cadence.SchedulePauseInfo pi = t.getPauseInfo(); + pauseReason = pi.getReason(); + pausedAt = nanosToInstant(pi.getPausedTimeNano()); + pausedBy = pi.getPausedBy(); + } + return new ScheduleState(t.isPaused(), pauseReason, pausedAt, pausedBy); + } + + private static ScheduleInfo toScheduleInfo(com.uber.cadence.ScheduleInfo t) { + if (t == null) return null; + List backfills = new ArrayList<>(); + if (t.getOngoingBackfills() != null) { + for (com.uber.cadence.BackfillInfo b : t.getOngoingBackfills()) { + backfills.add( + new ScheduleInfo.BackfillInfo( + b.getBackfillId(), + nanosToInstant(b.getStartTimeNano()), + nanosToInstant(b.getEndTimeNano()), + b.getRunsCompleted(), + b.getRunsTotal())); + } + } + return new ScheduleInfo( + nanosToInstant(t.getLastRunTimeNano()), + nanosToInstant(t.getNextRunTimeNano()), + t.getTotalRuns(), + nanosToInstant(t.getCreateTimeNano()), + nanosToInstant(t.getLastUpdateTimeNano()), + backfills, + t.getMissedRuns(), + t.getSkippedRuns()); + } + + private static ListSchedulesResult toListSchedulesResult(ListSchedulesResponse r) { + if (r == null) return new ListSchedulesResult(null, null); + List entries = new ArrayList<>(); + if (r.getSchedules() != null) { + for (com.uber.cadence.ScheduleListEntry e : r.getSchedules()) { + entries.add( + new ScheduleListEntry( + e.getScheduleId(), + e.getWorkflowType() != null ? e.getWorkflowType().getName() : null, + toScheduleState(e.getState()), + e.getCronExpression())); + } + } + return new ListSchedulesResult(entries, r.getNextPageToken()); + } + + private static RetryOptions toRetryOptions(com.uber.cadence.RetryPolicy p) { + return new RetryOptions.Builder() + .setInitialInterval(Duration.ofSeconds(p.getInitialIntervalInSeconds())) + .setMaximumInterval(Duration.ofSeconds(p.getMaximumIntervalInSeconds())) + .setBackoffCoefficient(p.getBackoffCoefficient()) + .setMaximumAttempts(p.getMaximumAttempts()) + .setExpiration(Duration.ofSeconds(p.getExpirationIntervalInSeconds())) + .build(); + } + + private static ScheduleOverlapPolicy toClientOverlapPolicy( + com.uber.cadence.ScheduleOverlapPolicy t) { + if (t == null) return null; + switch (t) { + case SKIP_NEW: + return ScheduleOverlapPolicy.SKIP_NEW; + case BUFFER: + return ScheduleOverlapPolicy.BUFFER; + case CONCURRENT: + return ScheduleOverlapPolicy.CONCURRENT; + case CANCEL_PREVIOUS: + return ScheduleOverlapPolicy.CANCEL_PREVIOUS; + case TERMINATE_PREVIOUS: + return ScheduleOverlapPolicy.TERMINATE_PREVIOUS; + default: + return null; + } + } + + private static com.uber.cadence.ScheduleOverlapPolicy toThriftOverlapPolicy( + ScheduleOverlapPolicy p) { + switch (p) { + case SKIP_NEW: + return com.uber.cadence.ScheduleOverlapPolicy.SKIP_NEW; + case BUFFER: + return com.uber.cadence.ScheduleOverlapPolicy.BUFFER; + case CONCURRENT: + return com.uber.cadence.ScheduleOverlapPolicy.CONCURRENT; + case CANCEL_PREVIOUS: + return com.uber.cadence.ScheduleOverlapPolicy.CANCEL_PREVIOUS; + case TERMINATE_PREVIOUS: + return com.uber.cadence.ScheduleOverlapPolicy.TERMINATE_PREVIOUS; + default: + return com.uber.cadence.ScheduleOverlapPolicy.INVALID; + } + } + + private static ScheduleCatchUpPolicy toClientCatchUpPolicy( + com.uber.cadence.ScheduleCatchUpPolicy t) { + if (t == null) return null; + switch (t) { + case SKIP: + return ScheduleCatchUpPolicy.SKIP; + case ONE: + return ScheduleCatchUpPolicy.ONE; + case ALL: + return ScheduleCatchUpPolicy.ALL; + default: + return null; + } + } + + private static Instant nanosToInstant(long nanos) { + if (nanos == 0) return null; + return Instant.ofEpochSecond(nanos / 1_000_000_000L, nanos % 1_000_000_000L); + } + + @SuppressWarnings("unchecked") + private static Map toObjectMap(Map src) { + if (src == null || src.isEmpty()) return null; + Map result = new HashMap<>(); + src.forEach((k, v) -> result.put(k, v)); + return result; + } +} diff --git a/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java b/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java index b19b87c4c..9ed10cb48 100644 --- a/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java +++ b/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java @@ -206,7 +206,7 @@ public ActivityCompletionClient newActivityCompletionClient() { @Override public ScheduleClient scheduleClient() { - throw new UnsupportedOperationException("not implemented"); + return new ScheduleClientImpl(workflowService, clientOptions.getDomain()); } @Override diff --git a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java index c25a371e5..db9185af1 100644 --- a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java +++ b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java @@ -802,45 +802,101 @@ public void RefreshWorkflowTasks(RefreshWorkflowTasksRequest request) @Override public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.createScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .createSchedule(RequestMapper.createScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.describeScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .describeSchedule(RequestMapper.describeScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.updateScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .updateSchedule(RequestMapper.updateScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.deleteScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .deleteSchedule(RequestMapper.deleteScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.pauseScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .pauseSchedule(RequestMapper.pauseScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.unpauseScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .unpauseSchedule(RequestMapper.unpauseScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.backfillScheduleResponse( + grpcServiceStubs + .scheduleBlockingStub() + .backfillSchedule(RequestMapper.backfillScheduleRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { - throw new UnsupportedOperationException("not implemented"); + try { + return ResponseMapper.listSchedulesResponse( + grpcServiceStubs + .scheduleBlockingStub() + .listSchedules(RequestMapper.listSchedulesRequest(request))); + } catch (Exception e) { + throw toServiceClientException(e); + } } @Override From 7af96c8d35a3cc417e7bf31f6b75dbd04de47756 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 10:56:48 -0700 Subject: [PATCH 07/17] refactor(schedule): return ListSchedulesResponse directly from listSchedules Remove the toListSchedulesResult mapping and unused ListSchedulesResult/ ScheduleListEntry imports; the service already returns ListSchedulesResponse. --- .../internal/sync/ScheduleClientImpl.java | 24 +++---------------- 1 file changed, 3 insertions(+), 21 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index 437de7980..83215ccbc 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -35,12 +35,10 @@ import com.uber.cadence.UpdateScheduleResponse; import com.uber.cadence.client.ScheduleBackfill; import com.uber.cadence.client.ScheduleClient; -import com.uber.cadence.client.schedule.ListSchedulesResult; import com.uber.cadence.client.schedule.ScheduleAction; import com.uber.cadence.client.schedule.ScheduleCatchUpPolicy; import com.uber.cadence.client.schedule.ScheduleDescription; import com.uber.cadence.client.schedule.ScheduleInfo; -import com.uber.cadence.client.schedule.ScheduleListEntry; import com.uber.cadence.client.schedule.ScheduleOverlapPolicy; import com.uber.cadence.client.schedule.SchedulePolicies; import com.uber.cadence.client.schedule.ScheduleSpec; @@ -180,7 +178,8 @@ public CompletableFuture> backfillSchedule( } @Override - public CompletableFuture listSchedules(int pageSize, byte[] nextPageToken) { + public CompletableFuture listSchedules( + int pageSize, byte[] nextPageToken) { ListSchedulesRequest request = new ListSchedulesRequest() .setDomain(domain) @@ -189,8 +188,7 @@ public CompletableFuture listSchedules(int pageSize, byte[] return CompletableFuture.supplyAsync( () -> { try { - ListSchedulesResponse resp = service.ListSchedules(request); - return toListSchedulesResult(resp); + return service.ListSchedules(request); } catch (Exception e) { throw new RuntimeException(e); } @@ -297,22 +295,6 @@ private static ScheduleInfo toScheduleInfo(com.uber.cadence.ScheduleInfo t) { t.getSkippedRuns()); } - private static ListSchedulesResult toListSchedulesResult(ListSchedulesResponse r) { - if (r == null) return new ListSchedulesResult(null, null); - List entries = new ArrayList<>(); - if (r.getSchedules() != null) { - for (com.uber.cadence.ScheduleListEntry e : r.getSchedules()) { - entries.add( - new ScheduleListEntry( - e.getScheduleId(), - e.getWorkflowType() != null ? e.getWorkflowType().getName() : null, - toScheduleState(e.getState()), - e.getCronExpression())); - } - } - return new ListSchedulesResult(entries, r.getNextPageToken()); - } - private static RetryOptions toRetryOptions(com.uber.cadence.RetryPolicy p) { return new RetryOptions.Builder() .setInitialInterval(Duration.ofSeconds(p.getInitialIntervalInSeconds())) From ba9a3a90bb803c2962c017c8710e9391cf19f6ae Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 11:11:02 -0700 Subject: [PATCH 08/17] fix(schedule): use dedicated executor and warn on unsupported backfill overlap policy Move all blocking gRPC calls off ForkJoinPool.commonPool() onto a dedicated daemon thread pool. Warn when ScheduleBackfill.overlapPolicy is set, as the current proto contract has no overlap_policy field on BackfillScheduleRequest and the value is silently dropped before reaching the server. --- .../internal/sync/ScheduleClientImpl.java | 44 +++++++++++++++---- 1 file changed, 35 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index 83215ccbc..48c3acda0 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -52,9 +52,23 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; final class ScheduleClientImpl implements ScheduleClient { + private static final Logger log = LoggerFactory.getLogger(ScheduleClientImpl.class); + + private static final Executor BLOCKING_EXECUTOR = + Executors.newCachedThreadPool( + r -> { + Thread t = new Thread(r, "cadence-schedule-client"); + t.setDaemon(true); + return t; + }); + private final IWorkflowService service; private final String domain; @@ -75,7 +89,8 @@ public CompletableFuture createSchedule( } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -90,7 +105,8 @@ public CompletableFuture describeSchedule(String scheduleId } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -105,7 +121,8 @@ public CompletableFuture updateSchedule( } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -119,7 +136,8 @@ public CompletableFuture deleteSchedule(String scheduleI } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -133,7 +151,8 @@ public CompletableFuture pauseSchedule(String scheduleId, } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -148,7 +167,8 @@ public CompletableFuture unpauseSchedule( } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -165,7 +185,11 @@ public CompletableFuture> backfillSchedule( .setStartTimeNano(bf.getStartTime().toEpochMilli() * 1_000_000L) .setEndTimeNano(bf.getEndTime().toEpochMilli() * 1_000_000L); if (bf.getOverlapPolicy() != null) { - request.setOverlapPolicy(toThriftOverlapPolicy(bf.getOverlapPolicy())); + log.warn( + "ScheduleBackfill.overlapPolicy is not supported by the current server proto" + + " contract and will be ignored for schedule '{}'. The schedule's configured" + + " overlap policy will be used instead.", + scheduleId); } try { results.add(service.BackfillSchedule(request)); @@ -174,7 +198,8 @@ public CompletableFuture> backfillSchedule( } } return results; - }); + }, + BLOCKING_EXECUTOR); } @Override @@ -192,7 +217,8 @@ public CompletableFuture listSchedules( } catch (Exception e) { throw new RuntimeException(e); } - }); + }, + BLOCKING_EXECUTOR); } private static ScheduleDescription toScheduleDescription(DescribeScheduleResponse r) { From 2c65e9796ff2fff92ecdf20f8b6060b032004b9f Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 11:31:04 -0700 Subject: [PATCH 09/17] fix(schedule): wrap async failures in CompletionException instead of RuntimeException Callers now unwrap one layer (CompletionException -> CadenceError) instead of two (CompletionException -> RuntimeException -> CadenceError), making typed error handling on EntityNotExistsError and similar exceptions straightforward. --- .../internal/sync/ScheduleClientImpl.java | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index 48c3acda0..58198ab3c 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -52,6 +52,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import org.slf4j.Logger; @@ -87,7 +88,7 @@ public CompletableFuture createSchedule( try { return service.CreateSchedule(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -103,7 +104,7 @@ public CompletableFuture describeSchedule(String scheduleId DescribeScheduleResponse resp = service.DescribeSchedule(request); return toScheduleDescription(resp); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -119,7 +120,7 @@ public CompletableFuture updateSchedule( try { return service.UpdateSchedule(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -134,7 +135,7 @@ public CompletableFuture deleteSchedule(String scheduleI try { return service.DeleteSchedule(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -149,7 +150,7 @@ public CompletableFuture pauseSchedule(String scheduleId, try { return service.PauseSchedule(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -165,7 +166,7 @@ public CompletableFuture unpauseSchedule( try { return service.UnpauseSchedule(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); @@ -194,7 +195,7 @@ public CompletableFuture> backfillSchedule( try { results.add(service.BackfillSchedule(request)); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } } return results; @@ -215,7 +216,7 @@ public CompletableFuture listSchedules( try { return service.ListSchedules(request); } catch (Exception e) { - throw new RuntimeException(e); + throw new CompletionException(e); } }, BLOCKING_EXECUTOR); From 7bb0afc7c3728d5d0aea677a17761d3d4b8352c1 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 12:16:57 -0700 Subject: [PATCH 10/17] fix(schedule): throw on unknown overlap policy instead of silently returning INVALID --- .../java/com/uber/cadence/internal/sync/ScheduleClientImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index 58198ab3c..a8a41e455 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -365,7 +365,7 @@ private static com.uber.cadence.ScheduleOverlapPolicy toThriftOverlapPolicy( case TERMINATE_PREVIOUS: return com.uber.cadence.ScheduleOverlapPolicy.TERMINATE_PREVIOUS; default: - return com.uber.cadence.ScheduleOverlapPolicy.INVALID; + throw new IllegalArgumentException("unknown ScheduleOverlapPolicy: " + p); } } From 7e2a76e99d9ba8046c61b378d9ebfd39911c0b8d Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 12:43:36 -0700 Subject: [PATCH 11/17] fix(schedule): map overlapPolicy and backfillId in BackfillScheduleRequest The proto BackfillScheduleRequest already has overlap_policy (field 5) and backfill_id (field 6); they were simply missing from the RequestMapper. Wire both through and restore setting overlapPolicy in ScheduleClientImpl. --- .../proto/mappers/RequestMapper.java | 20 +++++++++++++------ .../internal/sync/ScheduleClientImpl.java | 10 +--------- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java index d5e35615b..ff3d9eeb1 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java @@ -23,6 +23,7 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryConsistencyLevel; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryRejectCondition; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryTaskCompletedType; +import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.scheduleOverlapPolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.taskListType; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.workflowIdReusePolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.Helpers.arrayToByteString; @@ -1103,12 +1104,19 @@ public static com.uber.cadence.api.v1.BackfillScheduleRequest backfillScheduleRe if (t == null) { return null; } - return com.uber.cadence.api.v1.BackfillScheduleRequest.newBuilder() - .setDomain(nullToEmpty(t.getDomain())) - .setScheduleId(nullToEmpty(t.getScheduleId())) - .setStartTime(unixNanoToTime(t.getStartTimeNano())) - .setEndTime(unixNanoToTime(t.getEndTimeNano())) - .build(); + com.uber.cadence.api.v1.BackfillScheduleRequest.Builder b = + com.uber.cadence.api.v1.BackfillScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .setStartTime(unixNanoToTime(t.getStartTimeNano())) + .setEndTime(unixNanoToTime(t.getEndTimeNano())); + if (t.getOverlapPolicy() != null) { + b.setOverlapPolicy(scheduleOverlapPolicy(t.getOverlapPolicy())); + } + if (t.getBackfillId() != null) { + b.setBackfillId(t.getBackfillId()); + } + return b.build(); } public static com.uber.cadence.api.v1.ListSchedulesRequest listSchedulesRequest( diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index a8a41e455..b601b44ef 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -55,13 +55,9 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; import java.util.concurrent.Executors; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; final class ScheduleClientImpl implements ScheduleClient { - private static final Logger log = LoggerFactory.getLogger(ScheduleClientImpl.class); - private static final Executor BLOCKING_EXECUTOR = Executors.newCachedThreadPool( r -> { @@ -186,11 +182,7 @@ public CompletableFuture> backfillSchedule( .setStartTimeNano(bf.getStartTime().toEpochMilli() * 1_000_000L) .setEndTimeNano(bf.getEndTime().toEpochMilli() * 1_000_000L); if (bf.getOverlapPolicy() != null) { - log.warn( - "ScheduleBackfill.overlapPolicy is not supported by the current server proto" - + " contract and will be ignored for schedule '{}'. The schedule's configured" - + " overlap policy will be used instead.", - scheduleId); + request.setOverlapPolicy(toThriftOverlapPolicy(bf.getOverlapPolicy())); } try { results.add(service.BackfillSchedule(request)); From 161346e716eb6d3ebf63200b1660249d73a14c78 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 15:12:55 -0700 Subject: [PATCH 12/17] fix(schedule): map catchUpPolicy in unpauseScheduleRequest --- .../proto/mappers/RequestMapper.java | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java index ff3d9eeb1..1705fe1c6 100644 --- a/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java +++ b/src/main/java/com/uber/cadence/internal/compatibility/proto/mappers/RequestMapper.java @@ -23,6 +23,7 @@ import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryConsistencyLevel; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryRejectCondition; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.queryTaskCompletedType; +import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.scheduleCatchUpPolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.scheduleOverlapPolicy; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.taskListType; import static com.uber.cadence.internal.compatibility.proto.mappers.EnumMapper.workflowIdReusePolicy; @@ -1092,11 +1093,15 @@ public static com.uber.cadence.api.v1.UnpauseScheduleRequest unpauseScheduleRequ if (t == null) { return null; } - return com.uber.cadence.api.v1.UnpauseScheduleRequest.newBuilder() - .setDomain(nullToEmpty(t.getDomain())) - .setScheduleId(nullToEmpty(t.getScheduleId())) - .setReason(nullToEmpty(t.getReason())) - .build(); + com.uber.cadence.api.v1.UnpauseScheduleRequest.Builder b = + com.uber.cadence.api.v1.UnpauseScheduleRequest.newBuilder() + .setDomain(nullToEmpty(t.getDomain())) + .setScheduleId(nullToEmpty(t.getScheduleId())) + .setReason(nullToEmpty(t.getReason())); + if (t.getCatchUpPolicy() != null) { + b.setCatchUpPolicy(scheduleCatchUpPolicy(t.getCatchUpPolicy())); + } + return b.build(); } public static com.uber.cadence.api.v1.BackfillScheduleRequest backfillScheduleRequest( From 2830c7bb219e8546b92196fdd85d3257b49f9f1f Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 15:27:49 -0700 Subject: [PATCH 13/17] feat(schedule): add CompletableFuture async variants of schedule methods to IWorkflowService Default implementations wrap the blocking counterparts, consistent with the existing ScheduleClientImpl executor pattern. Callers that need a CompletableFuture handle can use these directly without reimplementing the wrapping. --- .../serviceclient/IWorkflowService.java | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java index 2f2e8f0be..8a6ee39fa 100644 --- a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java +++ b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java @@ -901,4 +901,100 @@ BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) */ ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws BadRequestError, EntityNotExistsError, ServiceBusyError, CadenceError; + + default CompletableFuture CreateScheduleAsync( + CreateScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return CreateSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture DescribeScheduleAsync( + DescribeScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return DescribeSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture UpdateScheduleAsync( + UpdateScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return UpdateSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture DeleteScheduleAsync( + DeleteScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return DeleteSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture PauseScheduleAsync( + PauseScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return PauseSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture UnpauseScheduleAsync( + UnpauseScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return UnpauseSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture BackfillScheduleAsync( + BackfillScheduleRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return BackfillSchedule(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } + + default CompletableFuture ListSchedulesAsync( + ListSchedulesRequest request) { + return CompletableFuture.supplyAsync( + () -> { + try { + return ListSchedules(request); + } catch (Exception e) { + throw new java.util.concurrent.CompletionException(e); + } + }); + } } From 2894d8056dc0d1fd4959c2f0e8653b39b6aa07a7 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 15:31:06 -0700 Subject: [PATCH 14/17] fix(schedule): remove duplicate test methods and stale ListSchedulesResult references from ScheduleTypesTest The ListSchedulesResult wrapper was removed in favour of ListSchedulesResponse in a prior commit in this PR. The four listSchedulesResult_* tests that still reference the removed type are deleted. A merge conflict during rebase also caused the seven scheduleBackfill_* test methods to appear twice; the second copy is removed. --- .../client/schedule/ScheduleTypesTest.java | 99 ------------------- 1 file changed, 99 deletions(-) diff --git a/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java b/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java index b6ed42c08..a57726a05 100644 --- a/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java +++ b/src/test/java/com/uber/cadence/client/schedule/ScheduleTypesTest.java @@ -529,33 +529,6 @@ public void scheduleDescription_nullMemoNormalizesToEmpty() { assertTrue(minimalDescription().getSearchAttributes().isEmpty()); } - @Test - public void listSchedulesResult_tokenDefensiveCopy() { - byte[] token = {7, 8, 9}; - ListSchedulesResult result = new ListSchedulesResult(Collections.emptyList(), token); - token[0] = 99; - assertEquals(7, result.getNextPageToken()[0]); - } - - @Test - public void listSchedulesResult_getTokenDoesNotExposeInternal() { - ListSchedulesResult result = - new ListSchedulesResult(Collections.emptyList(), new byte[] {1, 2, 3}); - result.getNextPageToken()[0] = 99; - assertEquals(1, result.getNextPageToken()[0]); - } - - @Test - public void listSchedulesResult_nullTokenRoundTrips() { - assertNull(new ListSchedulesResult(Collections.emptyList(), null).getNextPageToken()); - } - - @Test - public void listSchedulesResult_nullSchedulesNormalizesToEmpty() { - assertNotNull(new ListSchedulesResult(null, null).getSchedules()); - assertTrue(new ListSchedulesResult(null, null).getSchedules().isEmpty()); - } - @Test(expected = UnsupportedOperationException.class) public void scheduleDescription_memoIsUnmodifiable() { Map memo = new HashMap<>(); @@ -667,78 +640,6 @@ public void scheduleBackfill_toString() { assertTrue(s.contains("BUFFER")); } - @Test - public void scheduleBackfill_twoArgGetters() { - Instant start = Instant.ofEpochSecond(100); - Instant end = Instant.ofEpochSecond(200); - ScheduleBackfill bf = new ScheduleBackfill(start, end); - - assertEquals(start, bf.getStartTime()); - assertEquals(end, bf.getEndTime()); - assertNull(bf.getOverlapPolicy()); - } - - @Test - public void scheduleBackfill_threeArgGetters() { - Instant start = Instant.ofEpochSecond(100); - Instant end = Instant.ofEpochSecond(200); - ScheduleBackfill bf = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.BUFFER); - - assertEquals(start, bf.getStartTime()); - assertEquals(end, bf.getEndTime()); - assertEquals(ScheduleOverlapPolicy.BUFFER, bf.getOverlapPolicy()); - } - - @Test - public void scheduleBackfill_equals() { - Instant start = Instant.ofEpochSecond(100); - Instant end = Instant.ofEpochSecond(200); - ScheduleBackfill a = new ScheduleBackfill(start, end); - ScheduleBackfill b = new ScheduleBackfill(start, end); - assertEquals(a, b); - assertEquals(a.hashCode(), b.hashCode()); - } - - @Test - public void scheduleBackfill_equalsWithPolicy() { - Instant start = Instant.ofEpochSecond(100); - Instant end = Instant.ofEpochSecond(200); - ScheduleBackfill a = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW); - ScheduleBackfill b = new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW); - assertEquals(a, b); - assertEquals(a.hashCode(), b.hashCode()); - } - - @Test - public void scheduleBackfill_notEqualOnDifferentPolicy() { - Instant start = Instant.ofEpochSecond(100); - Instant end = Instant.ofEpochSecond(200); - assertNotEquals( - new ScheduleBackfill(start, end), - new ScheduleBackfill(start, end, ScheduleOverlapPolicy.SKIP_NEW)); - } - - @Test - public void scheduleBackfill_notEqualOnDifferentStart() { - Instant end = Instant.ofEpochSecond(200); - assertNotEquals( - new ScheduleBackfill(Instant.ofEpochSecond(100), end), - new ScheduleBackfill(Instant.ofEpochSecond(101), end)); - } - - @Test - public void scheduleBackfill_toString() { - String s = - new ScheduleBackfill( - Instant.ofEpochSecond(100), - Instant.ofEpochSecond(200), - ScheduleOverlapPolicy.BUFFER) - .toString(); - assertTrue(s.contains("startTime=")); - assertTrue(s.contains("endTime=")); - assertTrue(s.contains("BUFFER")); - } - private static ScheduleAction.StartWorkflowAction.Builder minimalStartWorkflowAction() { return ScheduleAction.StartWorkflowAction.newBuilder() .setWorkflowType("MyWorkflow") From 19fe3d4b05d75160a91696b42beca53de6c044ba Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 15:33:59 -0700 Subject: [PATCH 15/17] revert(schedule): remove default async variants from IWorkflowService These defaults dispatched blocking gRPC calls on ForkJoinPool.commonPool, the same anti-pattern ScheduleClientImpl avoids with its dedicated BLOCKING_EXECUTOR. Async dispatch belongs in ScheduleClientImpl, not at the service interface layer. --- .../serviceclient/IWorkflowService.java | 96 ------------------- 1 file changed, 96 deletions(-) diff --git a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java index 8a6ee39fa..2f2e8f0be 100644 --- a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java +++ b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java @@ -901,100 +901,4 @@ BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) */ ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws BadRequestError, EntityNotExistsError, ServiceBusyError, CadenceError; - - default CompletableFuture CreateScheduleAsync( - CreateScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return CreateSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture DescribeScheduleAsync( - DescribeScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return DescribeSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture UpdateScheduleAsync( - UpdateScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return UpdateSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture DeleteScheduleAsync( - DeleteScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return DeleteSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture PauseScheduleAsync( - PauseScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return PauseSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture UnpauseScheduleAsync( - UnpauseScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return UnpauseSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture BackfillScheduleAsync( - BackfillScheduleRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return BackfillSchedule(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } - - default CompletableFuture ListSchedulesAsync( - ListSchedulesRequest request) { - return CompletableFuture.supplyAsync( - () -> { - try { - return ListSchedules(request); - } catch (Exception e) { - throw new java.util.concurrent.CompletionException(e); - } - }); - } } From 074b008cdfa04375e5013ef21401b76e5c1cad45 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 16:18:57 -0700 Subject: [PATCH 16/17] refactor(schedule): convert schedule methods to async CompletableFuture interface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Change all 8 schedule methods on IWorkflowService from blocking (throws CadenceError) to CompletableFuture. Implement via scheduleFutureStub() in WorkflowServiceGrpc using a toCompletableFuture helper that bridges ListenableFuture through Guava FutureCallback — no extra thread pool needed. Drop BLOCKING_EXECUTOR from ScheduleClientImpl entirely; each method now delegates directly to service.Foo(request). backfillSchedule fires all requests concurrently via CompletableFuture.allOf(). Update stub implementations in IWorkflowServiceBase, TestWorkflowService, TestWorkflowEnvironmentInternal, and TestActivityEnvironmentInternal to match the new return types. --- .../internal/sync/ScheduleClientImpl.java | 126 ++++----------- .../sync/TestActivityEnvironmentInternal.java | 25 ++- .../sync/TestWorkflowEnvironmentInternal.java | 25 ++- .../testservice/TestWorkflowService.java | 22 +-- .../serviceclient/IWorkflowService.java | 71 ++------- .../serviceclient/IWorkflowServiceBase.java | 22 +-- .../serviceclient/WorkflowServiceGrpc.java | 147 +++++++++--------- 7 files changed, 158 insertions(+), 280 deletions(-) diff --git a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java index b601b44ef..60afcd94c 100644 --- a/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java +++ b/src/main/java/com/uber/cadence/internal/sync/ScheduleClientImpl.java @@ -52,20 +52,9 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; final class ScheduleClientImpl implements ScheduleClient { - private static final Executor BLOCKING_EXECUTOR = - Executors.newCachedThreadPool( - r -> { - Thread t = new Thread(r, "cadence-schedule-client"); - t.setDaemon(true); - return t; - }); - private final IWorkflowService service; private final String domain; @@ -79,31 +68,14 @@ public CompletableFuture createSchedule( String scheduleId, CreateScheduleRequest request) { request.setDomain(domain); request.setScheduleId(scheduleId); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.CreateSchedule(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.CreateSchedule(request); } @Override public CompletableFuture describeSchedule(String scheduleId) { DescribeScheduleRequest request = new DescribeScheduleRequest().setDomain(domain).setScheduleId(scheduleId); - return CompletableFuture.supplyAsync( - () -> { - try { - DescribeScheduleResponse resp = service.DescribeSchedule(request); - return toScheduleDescription(resp); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.DescribeSchedule(request).thenApply(ScheduleClientImpl::toScheduleDescription); } @Override @@ -111,45 +83,21 @@ public CompletableFuture updateSchedule( String scheduleId, UpdateScheduleRequest request) { request.setDomain(domain); request.setScheduleId(scheduleId); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.UpdateSchedule(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.UpdateSchedule(request); } @Override public CompletableFuture deleteSchedule(String scheduleId) { DeleteScheduleRequest request = new DeleteScheduleRequest().setDomain(domain).setScheduleId(scheduleId); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.DeleteSchedule(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.DeleteSchedule(request); } @Override public CompletableFuture pauseSchedule(String scheduleId, String reason) { PauseScheduleRequest request = new PauseScheduleRequest().setDomain(domain).setScheduleId(scheduleId).setReason(reason); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.PauseSchedule(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.PauseSchedule(request); } @Override @@ -157,42 +105,34 @@ public CompletableFuture unpauseSchedule( String scheduleId, String reason) { UnpauseScheduleRequest request = new UnpauseScheduleRequest().setDomain(domain).setScheduleId(scheduleId).setReason(reason); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.UnpauseSchedule(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.UnpauseSchedule(request); } @Override public CompletableFuture> backfillSchedule( String scheduleId, List backfills) { - return CompletableFuture.supplyAsync( - () -> { - List results = new ArrayList<>(); - for (ScheduleBackfill bf : backfills) { - BackfillScheduleRequest request = - new BackfillScheduleRequest() - .setDomain(domain) - .setScheduleId(scheduleId) - .setStartTimeNano(bf.getStartTime().toEpochMilli() * 1_000_000L) - .setEndTimeNano(bf.getEndTime().toEpochMilli() * 1_000_000L); - if (bf.getOverlapPolicy() != null) { - request.setOverlapPolicy(toThriftOverlapPolicy(bf.getOverlapPolicy())); - } - try { - results.add(service.BackfillSchedule(request)); - } catch (Exception e) { - throw new CompletionException(e); - } - } - return results; - }, - BLOCKING_EXECUTOR); + List> futures = new ArrayList<>(); + for (ScheduleBackfill bf : backfills) { + BackfillScheduleRequest request = + new BackfillScheduleRequest() + .setDomain(domain) + .setScheduleId(scheduleId) + .setStartTimeNano(bf.getStartTime().toEpochMilli() * 1_000_000L) + .setEndTimeNano(bf.getEndTime().toEpochMilli() * 1_000_000L); + if (bf.getOverlapPolicy() != null) { + request.setOverlapPolicy(toThriftOverlapPolicy(bf.getOverlapPolicy())); + } + futures.add(service.BackfillSchedule(request)); + } + return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) + .thenApply( + v -> { + List results = new ArrayList<>(); + for (CompletableFuture f : futures) { + results.add(f.join()); + } + return results; + }); } @Override @@ -203,15 +143,7 @@ public CompletableFuture listSchedules( .setDomain(domain) .setPageSize(pageSize) .setNextPageToken(nextPageToken); - return CompletableFuture.supplyAsync( - () -> { - try { - return service.ListSchedules(request); - } catch (Exception e) { - throw new CompletionException(e); - } - }, - BLOCKING_EXECUTOR); + return service.ListSchedules(request); } private static ScheduleDescription toScheduleDescription(DescribeScheduleResponse r) { diff --git a/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java b/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java index 97ad25b89..7b7f55811 100644 --- a/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java +++ b/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java @@ -583,48 +583,45 @@ public void RefreshWorkflowTasks(RefreshWorkflowTasksRequest request) } @Override - public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) - throws CadenceError { + public CompletableFuture CreateSchedule(CreateScheduleRequest request) { return impl.CreateSchedule(request); } @Override - public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws CadenceError { + public CompletableFuture DescribeSchedule( + DescribeScheduleRequest request) { return impl.DescribeSchedule(request); } @Override - public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) - throws CadenceError { + public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { return impl.UpdateSchedule(request); } @Override - public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) - throws CadenceError { + public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { return impl.DeleteSchedule(request); } @Override - public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { + public CompletableFuture PauseSchedule(PauseScheduleRequest request) { return impl.PauseSchedule(request); } @Override - public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws CadenceError { + public CompletableFuture UnpauseSchedule( + UnpauseScheduleRequest request) { return impl.UnpauseSchedule(request); } @Override - public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws CadenceError { + public CompletableFuture BackfillSchedule( + BackfillScheduleRequest request) { return impl.BackfillSchedule(request); } @Override - public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { + public CompletableFuture ListSchedules(ListSchedulesRequest request) { return impl.ListSchedules(request); } diff --git a/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java b/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java index 61d72ae00..2a98e585a 100644 --- a/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java +++ b/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java @@ -481,48 +481,45 @@ public void RefreshWorkflowTasks(RefreshWorkflowTasksRequest request) } @Override - public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) - throws CadenceError { + public CompletableFuture CreateSchedule(CreateScheduleRequest request) { return impl.CreateSchedule(request); } @Override - public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws CadenceError { + public CompletableFuture DescribeSchedule( + DescribeScheduleRequest request) { return impl.DescribeSchedule(request); } @Override - public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) - throws CadenceError { + public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { return impl.UpdateSchedule(request); } @Override - public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) - throws CadenceError { + public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { return impl.DeleteSchedule(request); } @Override - public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { + public CompletableFuture PauseSchedule(PauseScheduleRequest request) { return impl.PauseSchedule(request); } @Override - public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws CadenceError { + public CompletableFuture UnpauseSchedule( + UnpauseScheduleRequest request) { return impl.UnpauseSchedule(request); } @Override - public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws CadenceError { + public CompletableFuture BackfillSchedule( + BackfillScheduleRequest request) { return impl.BackfillSchedule(request); } @Override - public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { + public CompletableFuture ListSchedules(ListSchedulesRequest request) { return impl.ListSchedules(request); } diff --git a/src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java b/src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java index 03360e23b..887bb3772 100644 --- a/src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java +++ b/src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java @@ -1269,45 +1269,45 @@ public void RefreshWorkflowTasks( } @Override - public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) throws CadenceError { + public CompletableFuture CreateSchedule(CreateScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws CadenceError { + public CompletableFuture DescribeSchedule( + DescribeScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) throws CadenceError { + public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) throws CadenceError { + public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { + public CompletableFuture PauseSchedule(PauseScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws CadenceError { + public CompletableFuture UnpauseSchedule( + UnpauseScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws CadenceError { + public CompletableFuture BackfillSchedule( + BackfillScheduleRequest request) { throw new UnsupportedOperationException("not implemented"); } @Override - public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { + public CompletableFuture ListSchedules(ListSchedulesRequest request) { throw new UnsupportedOperationException("not implemented"); } diff --git a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java index 2f2e8f0be..f1cb223a0 100644 --- a/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java +++ b/src/main/java/com/uber/cadence/serviceclient/IWorkflowService.java @@ -817,88 +817,43 @@ void RefreshWorkflowTasks( RefreshWorkflowTasksRequest request, AsyncMethodCallback resultHandler) throws CadenceError; - /** - * Creates a new schedule in the domain. - * - * @throws BadRequestError if the schedule spec is invalid (e.g. bad cron expression) - * @throws WorkflowExecutionAlreadyStartedError if a schedule with the same ID already exists - * @throws EntityNotExistsError if the domain does not exist - * @throws CadenceError on other server errors - */ - CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) - throws BadRequestError, WorkflowExecutionAlreadyStartedError, EntityNotExistsError, - ServiceBusyError, DomainNotActiveError, LimitExceededError, CadenceError; + /** Creates a new schedule in the domain. Failures are delivered via the returned future. */ + CompletableFuture CreateSchedule(CreateScheduleRequest request); /** - * Returns the full configuration and runtime state of a schedule. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors + * Returns the full configuration and runtime state of a schedule. Failures are delivered via the + * returned future. */ - DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, CadenceError; + CompletableFuture DescribeSchedule(DescribeScheduleRequest request); /** * Replaces the spec, action, and policies of an existing schedule. * *

The server replaces the entire schedule configuration. Any field not included in the request * is zeroed out; always call describe first and resubmit the full configuration. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors */ - UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError, - LimitExceededError, CadenceError; + CompletableFuture UpdateSchedule(UpdateScheduleRequest request); /** * Permanently deletes a schedule. Running workflows triggered by this schedule are not affected. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors */ - DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError, - CadenceError; + CompletableFuture DeleteSchedule(DeleteScheduleRequest request); - /** - * Pauses a schedule. No new runs will be triggered while the schedule is paused. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors - */ - PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError, - CadenceError; + /** Pauses a schedule. No new runs will be triggered while the schedule is paused. */ + CompletableFuture PauseSchedule(PauseScheduleRequest request); /** * Resumes a paused schedule. Optionally overrides the catch-up policy for the first pass on * resume. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors */ - UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError, - CadenceError; + CompletableFuture UnpauseSchedule(UnpauseScheduleRequest request); /** * Triggers runs for all scheduled times within a historical time range. Backfill runs are subject * to the configured overlap policy. - * - * @throws EntityNotExistsError if the schedule does not exist - * @throws CadenceError on other server errors */ - BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, DomainNotActiveError, - LimitExceededError, CadenceError; + CompletableFuture BackfillSchedule(BackfillScheduleRequest request); - /** - * Lists schedules in a domain, paginated. - * - * @throws EntityNotExistsError if the domain does not exist - * @throws CadenceError on other server errors - */ - ListSchedulesResponse ListSchedules(ListSchedulesRequest request) - throws BadRequestError, EntityNotExistsError, ServiceBusyError, CadenceError; + /** Lists schedules in a domain, paginated. */ + CompletableFuture ListSchedules(ListSchedulesRequest request); } diff --git a/src/main/java/com/uber/cadence/serviceclient/IWorkflowServiceBase.java b/src/main/java/com/uber/cadence/serviceclient/IWorkflowServiceBase.java index 4d08c67fd..ad2d3c232 100644 --- a/src/main/java/com/uber/cadence/serviceclient/IWorkflowServiceBase.java +++ b/src/main/java/com/uber/cadence/serviceclient/IWorkflowServiceBase.java @@ -724,45 +724,45 @@ public CompletableFuture isHealthy() { } @Override - public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) throws CadenceError { + public CompletableFuture CreateSchedule(CreateScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws CadenceError { + public CompletableFuture DescribeSchedule( + DescribeScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) throws CadenceError { + public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) throws CadenceError { + public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { + public CompletableFuture PauseSchedule(PauseScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws CadenceError { + public CompletableFuture UnpauseSchedule( + UnpauseScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws CadenceError { + public CompletableFuture BackfillSchedule( + BackfillScheduleRequest request) { throw new UnsupportedOperationException("unimplemented"); } @Override - public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { + public CompletableFuture ListSchedules(ListSchedulesRequest request) { throw new UnsupportedOperationException("unimplemented"); } } diff --git a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java index db9185af1..891b437f7 100644 --- a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java +++ b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java @@ -19,6 +19,7 @@ import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.uber.cadence.*; import com.uber.cadence.internal.compatibility.proto.mappers.*; @@ -801,102 +802,78 @@ public void RefreshWorkflowTasks(RefreshWorkflowTasksRequest request) } @Override - public CreateScheduleResponse CreateSchedule(CreateScheduleRequest request) throws CadenceError { - try { - return ResponseMapper.createScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .createSchedule(RequestMapper.createScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture CreateSchedule(CreateScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .createSchedule(RequestMapper.createScheduleRequest(request)), + ResponseMapper::createScheduleResponse); } @Override - public DescribeScheduleResponse DescribeSchedule(DescribeScheduleRequest request) - throws CadenceError { - try { - return ResponseMapper.describeScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .describeSchedule(RequestMapper.describeScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture DescribeSchedule( + DescribeScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .describeSchedule(RequestMapper.describeScheduleRequest(request)), + ResponseMapper::describeScheduleResponse); } @Override - public UpdateScheduleResponse UpdateSchedule(UpdateScheduleRequest request) throws CadenceError { - try { - return ResponseMapper.updateScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .updateSchedule(RequestMapper.updateScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .updateSchedule(RequestMapper.updateScheduleRequest(request)), + ResponseMapper::updateScheduleResponse); } @Override - public DeleteScheduleResponse DeleteSchedule(DeleteScheduleRequest request) throws CadenceError { - try { - return ResponseMapper.deleteScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .deleteSchedule(RequestMapper.deleteScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .deleteSchedule(RequestMapper.deleteScheduleRequest(request)), + ResponseMapper::deleteScheduleResponse); } @Override - public PauseScheduleResponse PauseSchedule(PauseScheduleRequest request) throws CadenceError { - try { - return ResponseMapper.pauseScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .pauseSchedule(RequestMapper.pauseScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture PauseSchedule(PauseScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .pauseSchedule(RequestMapper.pauseScheduleRequest(request)), + ResponseMapper::pauseScheduleResponse); } @Override - public UnpauseScheduleResponse UnpauseSchedule(UnpauseScheduleRequest request) - throws CadenceError { - try { - return ResponseMapper.unpauseScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .unpauseSchedule(RequestMapper.unpauseScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture UnpauseSchedule( + UnpauseScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .unpauseSchedule(RequestMapper.unpauseScheduleRequest(request)), + ResponseMapper::unpauseScheduleResponse); } @Override - public BackfillScheduleResponse BackfillSchedule(BackfillScheduleRequest request) - throws CadenceError { - try { - return ResponseMapper.backfillScheduleResponse( - grpcServiceStubs - .scheduleBlockingStub() - .backfillSchedule(RequestMapper.backfillScheduleRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture BackfillSchedule( + BackfillScheduleRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .backfillSchedule(RequestMapper.backfillScheduleRequest(request)), + ResponseMapper::backfillScheduleResponse); } @Override - public ListSchedulesResponse ListSchedules(ListSchedulesRequest request) throws CadenceError { - try { - return ResponseMapper.listSchedulesResponse( - grpcServiceStubs - .scheduleBlockingStub() - .listSchedules(RequestMapper.listSchedulesRequest(request))); - } catch (Exception e) { - throw toServiceClientException(e); - } + public CompletableFuture ListSchedules(ListSchedulesRequest request) { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .listSchedules(RequestMapper.listSchedulesRequest(request)), + ResponseMapper::listSchedulesResponse); } @Override @@ -1470,6 +1447,26 @@ private CadenceError toServiceClientException(Throwable t) { } } + private CompletableFuture toCompletableFuture( + ListenableFuture

listenableFuture, Function mapper) { + CompletableFuture future = new CompletableFuture<>(); + Futures.addCallback( + listenableFuture, + new FutureCallback

() { + @Override + public void onSuccess(P result) { + future.complete(mapper.apply(result)); + } + + @Override + public void onFailure(Throwable t) { + future.completeExceptionally(toServiceClientException(t)); + } + }, + executor); + return future; + } + private FutureCallback toFutureCallback( AsyncMethodCallback resultHandler, Function mapper) { return new FutureCallback() { From 561997dd9ab6a73c772ec5daa79b9a86f8744b37 Mon Sep 17 00:00:00 2001 From: "abhishek.jha" Date: Thu, 2 Jul 2026 16:38:56 -0700 Subject: [PATCH 17/17] fix(schedule): catch eager mapping exceptions and complete future exceptionally RequestMapper/EnumMapper calls are evaluated before toCompletableFuture is entered, so any IllegalArgumentException (e.g. unrecognized enum values) previously escaped synchronously instead of completing the returned future. Wrap all 8 schedule methods in try/catch and use a failedFuture helper to route mapping failures through the future contract. --- .../serviceclient/WorkflowServiceGrpc.java | 118 ++++++++++++------ 1 file changed, 78 insertions(+), 40 deletions(-) diff --git a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java index 891b437f7..8294570ef 100644 --- a/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java +++ b/src/main/java/com/uber/cadence/serviceclient/WorkflowServiceGrpc.java @@ -803,77 +803,109 @@ public void RefreshWorkflowTasks(RefreshWorkflowTasksRequest request) @Override public CompletableFuture CreateSchedule(CreateScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .createSchedule(RequestMapper.createScheduleRequest(request)), - ResponseMapper::createScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .createSchedule(RequestMapper.createScheduleRequest(request)), + ResponseMapper::createScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture DescribeSchedule( DescribeScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .describeSchedule(RequestMapper.describeScheduleRequest(request)), - ResponseMapper::describeScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .describeSchedule(RequestMapper.describeScheduleRequest(request)), + ResponseMapper::describeScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture UpdateSchedule(UpdateScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .updateSchedule(RequestMapper.updateScheduleRequest(request)), - ResponseMapper::updateScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .updateSchedule(RequestMapper.updateScheduleRequest(request)), + ResponseMapper::updateScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture DeleteSchedule(DeleteScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .deleteSchedule(RequestMapper.deleteScheduleRequest(request)), - ResponseMapper::deleteScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .deleteSchedule(RequestMapper.deleteScheduleRequest(request)), + ResponseMapper::deleteScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture PauseSchedule(PauseScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .pauseSchedule(RequestMapper.pauseScheduleRequest(request)), - ResponseMapper::pauseScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .pauseSchedule(RequestMapper.pauseScheduleRequest(request)), + ResponseMapper::pauseScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture UnpauseSchedule( UnpauseScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .unpauseSchedule(RequestMapper.unpauseScheduleRequest(request)), - ResponseMapper::unpauseScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .unpauseSchedule(RequestMapper.unpauseScheduleRequest(request)), + ResponseMapper::unpauseScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture BackfillSchedule( BackfillScheduleRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .backfillSchedule(RequestMapper.backfillScheduleRequest(request)), - ResponseMapper::backfillScheduleResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .backfillSchedule(RequestMapper.backfillScheduleRequest(request)), + ResponseMapper::backfillScheduleResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override public CompletableFuture ListSchedules(ListSchedulesRequest request) { - return toCompletableFuture( - grpcServiceStubs - .scheduleFutureStub() - .listSchedules(RequestMapper.listSchedulesRequest(request)), - ResponseMapper::listSchedulesResponse); + try { + return toCompletableFuture( + grpcServiceStubs + .scheduleFutureStub() + .listSchedules(RequestMapper.listSchedulesRequest(request)), + ResponseMapper::listSchedulesResponse); + } catch (Exception e) { + return failedFuture(e); + } } @Override @@ -1467,6 +1499,12 @@ public void onFailure(Throwable t) { return future; } + private CompletableFuture failedFuture(Exception e) { + CompletableFuture future = new CompletableFuture<>(); + future.completeExceptionally(toServiceClientException(e)); + return future; + } + private FutureCallback toFutureCallback( AsyncMethodCallback resultHandler, Function mapper) { return new FutureCallback() {