From f56504f4cb554a50ee15f4246af2c44d0d7c9a7f Mon Sep 17 00:00:00 2001 From: Kowser Date: Sat, 18 Jul 2026 01:21:41 -0700 Subject: [PATCH] refactor: Schedules delegates all scheduler HTTP to SchedulerClient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Schedules no longer builds raw /scheduler/* requests itself — it holds a SchedulerClient (OrkesSchedulerClient) and translates its typed responses to/from Schedule/ScheduleInfo, keeping only agent-scoped naming (prefix/unprefix), declarative reconcile, and runNow sugar (still via WorkflowClient, untouched). Exception translation (404 -> NotFound, 400+"cron" -> InvalidCron, else AgentAPIException) stays in the facade; SchedulerClient itself keeps throwing plain ConductorClientException so standard-client users see no new exception types. toSaveRequest/fromWorkflowSchedule now build/read the typed SaveScheduleRequest/WorkflowSchedule DTOs instead of raw Maps. Public API is unchanged (Example99ScheduledAgent compiles with zero changes). Test fallout is purely mechanical: the four helper-level tests that asserted on raw Map shapes now assert the same values through typed getters. Every behavioural test (prefix/unprefix, validation, runNow*, reconcile) is unmodified and still passes. --- CHANGELOG.md | 4 + .../conductor/ai/schedule/Schedules.java | 200 ++++------ .../conductor/ai/schedule/ScheduleTest.java | 81 ++-- .../conductor/client/SchedulerClient.java | 15 + .../client/http/OrkesSchedulerClient.java | 5 + .../client/http/SchedulerResource.java | 70 +++- .../conductor/client/model/CronSchedule.java | 26 ++ .../client/model/SaveScheduleRequest.java | 13 + .../client/model/WorkflowSchedule.java | 15 + .../client/http/SchedulerResourceTest.java | 348 ++++++++++++++++++ .../test/resources/ser_deser_json_string.json | 24 +- 11 files changed, 619 insertions(+), 182 deletions(-) create mode 100644 conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java create mode 100644 conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index b4752eb97..aab47160e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,8 @@ All notable changes to this project will be documented in this file. - Liveness for stateful runs: a `SCHEDULED` task with zero polls beyond `livenessStallSeconds` surfaces as `WorkerStallError` from the handle's wait instead of burning the full timeout - Swarm hand-offs: transfer tools echo the hand-off `message`; `check_transfer` is first-wins with `transfer_message` and surfaces non-winning transfers in `dropped_transfers` (with a warning) instead of silently discarding them - `agent-e2e` GitHub workflow runs the e2e suites as a two-server matrix: the Conductor OSS server boot JAR `3.32.0-rc.8` (Maven Central) and the released `agentspan-server-0.4.4.jar` +- `SchedulerClient.pauseSchedule(name, reason)` — an optional pause reason, persisted as `pausedReason` +- `WorkflowSchedule`/`SaveScheduleRequest` gain `nextRunTime` (read) and `cronSchedules` (a new `CronSchedule` list — multi-cron schedules) — additive, null-omitting serialization keeps existing save bodies unchanged when unused; `cronSchedules` is sent as-is and silently ignored on servers whose create DTO doesn't expose it ### Removed @@ -26,6 +28,8 @@ All notable changes to this project will be documented in this file. ### Changed - `useEnvVariables(true)` (and the `new ApiClient()` no-arg constructor) no longer throws when `CONDUCTOR_SERVER_URL` is unset — it falls back to `AGENTSPAN_SERVER_URL`, then `http://localhost:8080/api`; the resolved URL is normalized to end in `/api`, and credentials gain the `AGENTSPAN_AUTH_KEY`/`AGENTSPAN_AUTH_SECRET` fallback +- `SchedulerClient.pauseSchedule`/`resumeSchedule` now send `PUT` first (OSS Conductor's verb) and fall back to `GET` — caching that verdict for the client's lifetime — only on a `405`; any other error status propagates immediately. This makes the same client portable across OSS Conductor, Orkes Conductor (`GET`-only today), and agentspan-embedded servers with no configuration +- `SchedulerClient.getSchedule` now throws consistently on a missing schedule ## [5.1.0] diff --git a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java index ea1100a1d..3526ca643 100644 --- a/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java +++ b/conductor-client-ai/src/main/java/org/conductoross/conductor/ai/schedule/Schedules.java @@ -12,37 +12,30 @@ */ package org.conductoross.conductor.ai.schedule; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; +import java.util.function.Supplier; import org.conductoross.conductor.ai.model.AgentHandle; import org.conductoross.conductor.ai.model.AgentResult; import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.client.http.ConductorClient; -import com.netflix.conductor.client.http.ConductorClientRequest; -import com.netflix.conductor.client.http.ConductorClientRequest.Method; import com.netflix.conductor.client.http.WorkflowClient; import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest; import com.netflix.conductor.common.run.Workflow; +import io.orkes.conductor.client.SchedulerClient; import io.orkes.conductor.client.exceptions.AgentAPIException; - -import com.fasterxml.jackson.core.type.TypeReference; +import io.orkes.conductor.client.http.OrkesSchedulerClient; +import io.orkes.conductor.client.model.SaveScheduleRequest; +import io.orkes.conductor.client.model.WorkflowSchedule; /** * Lifecycle API for cron-based agent schedules. Obtained via {@code runtime.schedules()}. * - *

All requests ride the shared native Conductor {@link ConductorClient}/ApiClient - * (same HTTP + token-auth backend as every other client) — the scheduler CRUD via - * {@link ConductorClientRequest}/{@link ConductorClient#execute} ({@code /api/scheduler/*}, - * for which the Conductor client ships no typed {@code SchedulerClient}), and - * {@code runNow} via the typed {@link WorkflowClient}. + *

This class contributes only agent-scoped naming (prefix/unprefix), + * {@link Schedule}/{@link ScheduleInfo} DTO mapping, declarative + * {@link #reconcile}, and {@code runNow} sugar (via the typed {@link WorkflowClient}). * *

Operations are keyed by the wire name (prefixed with * {@code agent-}) returned by {@link #list(String)}. Use {@link Schedule} to @@ -50,65 +43,43 @@ */ public class Schedules { - private static final TypeReference> MAP_TYPE = new TypeReference>() {}; - private static final TypeReference>> LIST_MAP_TYPE = - new TypeReference>>() {}; - private static final TypeReference> LIST_LONG_TYPE = new TypeReference>() {}; - - private final ConductorClient client; + private final SchedulerClient schedulerClient; /** Shared native Conductor client for starting workflows (runNow). */ private final WorkflowClient workflowClient; public Schedules(ConductorClient conductorClient) { - this.client = conductorClient; + this.schedulerClient = new OrkesSchedulerClient(conductorClient); this.workflowClient = new WorkflowClient(conductorClient); } /** Test seam: inject a {@link WorkflowClient} so {@code runNow}/{@code runNowAndWait} can be unit-tested. */ Schedules(ConductorClient conductorClient, WorkflowClient workflowClient) { - this.client = conductorClient; + this.schedulerClient = new OrkesSchedulerClient(conductorClient); this.workflowClient = workflowClient; } // ── CRUD ──────────────────────────────────────────────────────────── public void save(Schedule schedule, String agentName) { - Map body = toSaveRequest(schedule, agentName); - execVoid(ConductorClientRequest.builder() - .method(Method.POST) - .path("/scheduler/schedules") - .body(body) - .build()); + SaveScheduleRequest request = toSaveRequest(schedule, agentName); + translate(() -> { + schedulerClient.saveSchedule(request); + return null; + }); } public ScheduleInfo get(String wireName) { - Map resp = exec( - ConductorClientRequest.builder() - .method(Method.GET) - .path("/scheduler/schedules/{name}") - .addPathParam("name", wireName) - .build(), - MAP_TYPE); - if (resp == null || resp.isEmpty() || resp.get("name") == null) { - throw new ScheduleException.NotFound("Schedule '" + wireName + "' not found"); - } - return fromWorkflowSchedule(resp, null); + WorkflowSchedule ws = translate(() -> schedulerClient.getSchedule(wireName)); + return fromWorkflowSchedule(ws, null); } public List list(String agentName) { - List> resp = exec( - ConductorClientRequest.builder() - .method(Method.GET) - .path("/scheduler/schedules") - .addQueryParam("workflowName", agentName) - .build(), - LIST_MAP_TYPE); + List resp = translate(() -> schedulerClient.getAllSchedules(agentName)); if (resp == null) return new ArrayList<>(); - List out = new ArrayList<>(); - for (Map item : resp) { - if (item != null) out.add(fromWorkflowSchedule(item, agentName)); - } - return out; + return resp.stream() + .filter(Objects::nonNull) + .map(item -> fromWorkflowSchedule(item, agentName)) + .toList(); } public void pause(String wireName) { @@ -116,28 +87,24 @@ public void pause(String wireName) { } public void pause(String wireName, String reason) { - ConductorClientRequest.Builder b = ConductorClientRequest.builder() - .method(Method.PUT) - .path("/scheduler/schedules/{name}/pause") - .addPathParam("name", wireName); - if (reason != null) b.addQueryParam("reason", reason); - execVoid(b.build()); + translate(() -> { + schedulerClient.pauseSchedule(wireName, reason); + return null; + }); } public void resume(String wireName) { - execVoid(ConductorClientRequest.builder() - .method(Method.PUT) - .path("/scheduler/schedules/{name}/resume") - .addPathParam("name", wireName) - .build()); + translate(() -> { + schedulerClient.resumeSchedule(wireName); + return null; + }); } public void delete(String wireName) { - execVoid(ConductorClientRequest.builder() - .method(Method.DELETE) - .path("/scheduler/schedules/{name}") - .addPathParam("name", wireName) - .build()); + translate(() -> { + schedulerClient.deleteSchedule(wireName); + return null; + }); } /** @@ -241,14 +208,7 @@ static boolean isTerminal(Workflow wf) { } public List previewNext(String cron, int n) { - List resp = exec( - ConductorClientRequest.builder() - .method(Method.GET) - .path("/scheduler/nextFewSchedules") - .addQueryParam("cronExpression", cron) - .addQueryParam("limit", Integer.valueOf(n)) - .build(), - LIST_LONG_TYPE); + List resp = translate(() -> schedulerClient.getNextFewSchedules(cron, null, null, n)); return resp != null ? resp : new ArrayList<>(); } @@ -304,68 +264,52 @@ static void checkUniqueNames(List schedules) { } } - static Map toSaveRequest(Schedule s, String agentName) { - Map swr = new LinkedHashMap<>(); - swr.put("name", agentName); - swr.put("input", s.getInput() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(s.getInput())); - - Map req = new LinkedHashMap<>(); - req.put("name", prefix(agentName, s.getName())); - req.put("cronExpression", s.getCron()); - req.put("zoneId", s.getTimezone()); - req.put("runCatchupScheduleInstances", s.isCatchup()); - req.put("paused", s.isPaused()); - if (s.getStartAt() != null) req.put("scheduleStartTime", s.getStartAt()); - if (s.getEndAt() != null) req.put("scheduleEndTime", s.getEndAt()); - if (s.getDescription() != null) req.put("description", s.getDescription()); - req.put("startWorkflowRequest", swr); - return req; + static SaveScheduleRequest toSaveRequest(Schedule s, String agentName) { + StartWorkflowRequest startWorkflowRequest = new StartWorkflowRequest() + .withName(agentName) + .withInput(s.getInput() == null ? Map.of() : new LinkedHashMap<>(s.getInput())); + return new SaveScheduleRequest() + .name(prefix(agentName, s.getName())) + .cronExpression(s.getCron()) + .zoneId(s.getTimezone()) + .runCatchupScheduleInstances(s.isCatchup()) + .paused(s.isPaused()) + .scheduleStartTime(s.getStartAt()) + .scheduleEndTime(s.getEndAt()) + .startWorkflowRequest(startWorkflowRequest) + .description(s.getDescription()); } - @SuppressWarnings("unchecked") - static ScheduleInfo fromWorkflowSchedule(Map ws, String agentHint) { - Map swr = (Map) ws.getOrDefault("startWorkflowRequest", new HashMap<>()); - String wireName = (String) ws.getOrDefault("name", ""); - String swrName = (String) swr.getOrDefault("name", ""); + static ScheduleInfo fromWorkflowSchedule(WorkflowSchedule ws, String agentHint) { + StartWorkflowRequest swr = ws.getStartWorkflowRequest(); + String wireName = ws.getName() != null ? ws.getName() : ""; + String swrName = swr != null && swr.getName() != null ? swr.getName() : ""; String agent = agentHint != null ? agentHint : (swrName.isEmpty() ? "" : swrName); return new ScheduleInfo( wireName, unprefix(agent, wireName), swrName, - (String) ws.getOrDefault("cronExpression", ""), - (String) ws.getOrDefault("zoneId", "UTC"), - (Map) swr.getOrDefault("input", new HashMap<>()), - Boolean.TRUE.equals(ws.get("paused")), - (String) ws.get("pausedReason"), - Boolean.TRUE.equals(ws.get("runCatchupScheduleInstances")), - longOrNull(ws.get("scheduleStartTime")), - longOrNull(ws.get("scheduleEndTime")), - (String) ws.get("description"), - longOrNull(ws.get("nextRunTime")), - longOrNull(ws.get("createTime")), - longOrNull(ws.get("updatedTime")), - (String) ws.get("createdBy"), - (String) ws.get("updatedBy")); - } - - private static Long longOrNull(Object o) { - return o instanceof Number ? ((Number) o).longValue() : null; - } - - /** Execute a scheduler request returning a typed body via the native Conductor client. */ - private T exec(ConductorClientRequest req, TypeReference type) { - try { - return client.execute(req, type).getData(); - } catch (ConductorClientException e) { - throw mapException(e); - } + ws.getCronExpression() != null ? ws.getCronExpression() : "", + ws.getZoneId() != null ? ws.getZoneId() : "UTC", + swr != null && swr.getInput() != null ? swr.getInput() : new HashMap<>(), + Boolean.TRUE.equals(ws.isPaused()), + ws.getPausedReason(), + Boolean.TRUE.equals(ws.isRunCatchupScheduleInstances()), + ws.getScheduleStartTime(), + ws.getScheduleEndTime(), + ws.getDescription(), + ws.getNextRunTime(), + ws.getCreateTime(), + ws.getUpdatedTime(), + ws.getCreatedBy(), + ws.getUpdatedBy()); } - /** Execute a scheduler request that returns no body. */ - private void execVoid(ConductorClientRequest req) { + /** Run a {@link SchedulerClient} call, translating transport errors to the scheduler's typed exceptions. */ + private static T translate(Supplier call) { try { - client.execute(req); + return call.get(); } catch (ConductorClientException e) { throw mapException(e); } diff --git a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java index eb03e4bcd..905b2b83f 100644 --- a/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java +++ b/conductor-client-ai/src/test/java/org/conductoross/conductor/ai/schedule/ScheduleTest.java @@ -19,6 +19,11 @@ import org.junit.jupiter.api.Test; +import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest; + +import io.orkes.conductor.client.model.SaveScheduleRequest; +import io.orkes.conductor.client.model.WorkflowSchedule; + import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; @@ -121,16 +126,15 @@ void agentNameWithHyphen() { @Test void toSaveRequestMinimal() { Schedule s = Schedule.builder().name("daily").cron("0 0 9 * * ?").build(); - Map req = Schedules.toSaveRequest(s, "digest"); - assertEquals("digest-daily", req.get("name")); - assertEquals("0 0 9 * * ?", req.get("cronExpression")); - assertEquals("UTC", req.get("zoneId")); - assertEquals(false, req.get("paused")); - assertEquals(false, req.get("runCatchupScheduleInstances")); - @SuppressWarnings("unchecked") - Map swr = (Map) req.get("startWorkflowRequest"); - assertEquals("digest", swr.get("name")); - assertTrue(((Map) swr.get("input")).isEmpty()); + SaveScheduleRequest req = Schedules.toSaveRequest(s, "digest"); + assertEquals("digest-daily", req.getName()); + assertEquals("0 0 9 * * ?", req.getCronExpression()); + assertEquals("UTC", req.getZoneId()); + assertEquals(false, req.isPaused()); + assertEquals(false, req.isRunCatchupScheduleInstances()); + StartWorkflowRequest swr = req.getStartWorkflowRequest(); + assertEquals("digest", swr.getName()); + assertTrue(swr.getInput().isEmpty()); } @Test @@ -149,13 +153,13 @@ void toSaveRequestFull() { .endAt(2000L) .description("desc") .build(); - Map req = Schedules.toSaveRequest(s, "digest"); - assertEquals("America/Los_Angeles", req.get("zoneId")); - assertEquals(true, req.get("paused")); - assertEquals(true, req.get("runCatchupScheduleInstances")); - assertEquals(1000L, req.get("scheduleStartTime")); - assertEquals(2000L, req.get("scheduleEndTime")); - assertEquals("desc", req.get("description")); + SaveScheduleRequest req = Schedules.toSaveRequest(s, "digest"); + assertEquals("America/Los_Angeles", req.getZoneId()); + assertEquals(true, req.isPaused()); + assertEquals(true, req.isRunCatchupScheduleInstances()); + assertEquals(1000L, req.getScheduleStartTime()); + assertEquals(2000L, req.getScheduleEndTime()); + assertEquals("desc", req.getDescription()); } @Test @@ -164,29 +168,29 @@ void inputCopiedNotShared() { original.put("a", 1); Schedule s = Schedule.builder().name("x").cron("* * * * * ?").input(original).build(); - Map req = Schedules.toSaveRequest(s, "agent"); - @SuppressWarnings("unchecked") - Map swrInput = - (Map) ((Map) req.get("startWorkflowRequest")).get("input"); + SaveScheduleRequest req = Schedules.toSaveRequest(s, "agent"); + Map swrInput = req.getStartWorkflowRequest().getInput(); swrInput.put("mutated", true); assertNull(original.get("mutated")); } @Test void fromWorkflowScheduleBasic() { - Map ws = new LinkedHashMap<>(); - ws.put("name", "digest-daily"); - ws.put("cronExpression", "0 0 9 * * ?"); - ws.put("zoneId", "UTC"); - ws.put("paused", false); - Map swr = new LinkedHashMap<>(); - swr.put("name", "digest"); Map input = new LinkedHashMap<>(); input.put("c", "#eng"); - swr.put("input", input); - ws.put("startWorkflowRequest", swr); - ws.put("createTime", 111L); - ws.put("createdBy", "alice"); + StartWorkflowRequest swr = new StartWorkflowRequest(); + swr.setName("digest"); + swr.setInput(input); + + WorkflowSchedule ws = WorkflowSchedule.builder() + .name("digest-daily") + .cronExpression("0 0 9 * * ?") + .zoneId("UTC") + .paused(false) + .startWorkflowRequest(swr) + .createTime(111L) + .createdBy("alice") + .build(); ScheduleInfo info = Schedules.fromWorkflowSchedule(ws, "digest"); assertEquals("digest-daily", info.getName()); @@ -201,11 +205,14 @@ void fromWorkflowScheduleBasic() { @Test void fromWorkflowScheduleDerivesAgentWhenOmitted() { - Map ws = new LinkedHashMap<>(); - ws.put("name", "digest-daily"); - Map swr = new LinkedHashMap<>(); - swr.put("name", "digest"); - ws.put("startWorkflowRequest", swr); + StartWorkflowRequest swr = new StartWorkflowRequest(); + swr.setName("digest"); + + WorkflowSchedule ws = WorkflowSchedule.builder() + .name("digest-daily") + .startWorkflowRequest(swr) + .build(); + ScheduleInfo info = Schedules.fromWorkflowSchedule(ws, null); assertEquals("digest", info.getAgent()); assertEquals("daily", info.getShortName()); diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java b/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java index 878176fd6..94f38d91d 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/SchedulerClient.java @@ -92,6 +92,21 @@ public interface SchedulerClient { */ void pauseSchedule(String name); + /** + * Pauses a specific schedule, recording an optional reason. + *

+ * The reason is persisted as {@code pausedReason} where the server supports it (e.g. OSS + * Conductor); it is silently ignored otherwise (e.g. Orkes Conductor today). The default + * implementation ignores the reason and delegates to {@link #pauseSchedule(String)} so + * existing {@link SchedulerClient} implementations remain source- and binary-compatible. + * + * @param name the name of the schedule to pause + * @param reason optional reason for the pause, or {@code null} + */ + default void pauseSchedule(String name, String reason) { + pauseSchedule(name); + } + /** * Resumes a paused schedule, allowing new workflow executions. * diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java index bacbbdaaf..dfc7de1dd 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/OrkesSchedulerClient.java @@ -67,6 +67,11 @@ public void pauseSchedule(String name) { schedulerResource.pauseSchedule(name); } + @Override + public void pauseSchedule(String name, String reason) { + schedulerResource.pauseSchedule(name, reason); + } + @Override public void requeueAllExecutionRecords() { schedulerResource.requeueAllExecutionRecords(); diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java index 2506c0df1..ca595cca8 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/http/SchedulerResource.java @@ -15,6 +15,7 @@ import java.util.List; import java.util.Map; +import com.netflix.conductor.client.exception.ConductorClientException; import com.netflix.conductor.client.http.ConductorClient; import com.netflix.conductor.client.http.ConductorClientRequest; import com.netflix.conductor.client.http.ConductorClientRequest.Method; @@ -32,6 +33,13 @@ public class SchedulerResource { private final ConductorClient client; + /** + * Once a server answers {@code 405} to a PUT pause/resume request (OSS Conductor's original + * verb; Orkes Conductor historically accepted only {@code GET}), fall back to {@code GET} for + * the remaining lifetime of this instance rather than probing on every call. + */ + private volatile boolean useGetForScheduleStateChange = false; + public SchedulerResource(ConductorClient client) { this.client = client; } @@ -88,7 +96,14 @@ public WorkflowSchedule getSchedule(String name) { ConductorClientResponse resp = client.execute(request, new TypeReference<>() { }); - return resp.getData(); + WorkflowSchedule schedule = resp.getData(); + if (schedule == null || schedule.getName() == null) { + // OSS Conductor returns 200 with an empty/null body on a miss (Orkes returns 404, + // which already surfaces as a ConductorClientException below); normalize both server + // families to the same not-found contract. + throw new ConductorClientException(404, "Schedule '" + name + "' not found"); + } + return schedule; } public Map pauseAllSchedules() { @@ -104,13 +119,15 @@ public Map pauseAllSchedules() { } public void pauseSchedule(String name) { - ConductorClientRequest request = ConductorClientRequest.builder() - .method(Method.GET) - .path("/scheduler/schedules/{name}/pause") - .addPathParam("name", name) - .build(); + pauseSchedule(name, null); + } - client.execute(request); + /** + * Pauses a schedule, recording an optional reason (persisted as {@code pausedReason} where + * the server supports it; silently ignored otherwise — e.g. Orkes Conductor today). + */ + public void pauseSchedule(String name, String reason) { + executeStateChange("/scheduler/schedules/{name}/pause", name, reason); } public Map requeueAllExecutionRecords() { @@ -138,13 +155,40 @@ public Map resumeAllSchedules() { } public void resumeSchedule(String name) { - ConductorClientRequest request = ConductorClientRequest.builder() - .method(Method.GET) - .path("/scheduler/schedules/{name}/resume") - .addPathParam("name", name) - .build(); + executeStateChange("/scheduler/schedules/{name}/resume", name, null); + } - client.execute(request); + /** + * Sends a schedule state-change (pause/resume) as {@code PUT} first; if the server answers + * {@code 405 Method Not Allowed} (Orkes Conductor, historically), falls back to {@code GET} + * and remembers that verdict for the rest of this instance's lifetime. Any other error + * status (403, 404, 5xx, ...) propagates immediately — only a routing mismatch (405) + * triggers the fallback. + */ + private void executeStateChange(String path, String name, String reason) { + if (!useGetForScheduleStateChange) { + try { + client.execute(stateChangeRequest(Method.PUT, path, name, reason)); + return; + } catch (ConductorClientException e) { + if (e.getStatus() != 405) { + throw e; + } + useGetForScheduleStateChange = true; + } + } + client.execute(stateChangeRequest(Method.GET, path, name, reason)); + } + + private ConductorClientRequest stateChangeRequest(Method method, String path, String name, String reason) { + ConductorClientRequest.Builder builder = ConductorClientRequest.builder() + .method(method) + .path(path) + .addPathParam("name", name); + if (reason != null) { + builder.addQueryParam("reason", reason); + } + return builder.build(); } public void saveSchedule(SaveScheduleRequest saveScheduleRequest) { diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java new file mode 100644 index 000000000..8dbc63689 --- /dev/null +++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/CronSchedule.java @@ -0,0 +1,26 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License 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 io.orkes.conductor.client.model; + +import lombok.*; + +/** One entry of a multi-cron schedule: its own cron expression and timezone. */ +@Data +@NoArgsConstructor +@Builder +@AllArgsConstructor(access = AccessLevel.PRIVATE) +public class CronSchedule { + + private String cronExpression; + private String zoneId; +} diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java index c1190ce99..f8c602435 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/SaveScheduleRequest.java @@ -12,6 +12,8 @@ */ package io.orkes.conductor.client.model; +import java.util.List; + import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest; import lombok.*; @@ -46,6 +48,8 @@ public class SaveScheduleRequest { private String description; + private List cronSchedules; + public SaveScheduleRequest createdBy(String createdBy) { this.createdBy = createdBy; return this; @@ -104,4 +108,13 @@ public SaveScheduleRequest zoneId(String zoneId) { return this; } + public SaveScheduleRequest cronSchedules(List cronSchedules) { + this.cronSchedules = cronSchedules; + return this; + } + + public SaveScheduleRequest description(String description) { + this.description = description; + return this; + } } \ No newline at end of file diff --git a/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java b/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java index f29e1c958..dc36e13fa 100644 --- a/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java +++ b/conductor-client/src/main/java/io/orkes/conductor/client/model/WorkflowSchedule.java @@ -56,6 +56,11 @@ public class WorkflowSchedule { private String description; + /** Server-owned next-scheduled execution time, in epoch millis. Can be null if unavailable. **/ + private Long nextRunTime; + + private List cronSchedules; + public WorkflowSchedule createTime(Long createTime) { this.createTime = createTime; return this; @@ -123,4 +128,14 @@ public WorkflowSchedule zoneId(String zoneId) { this.zoneId = zoneId; return this; } + + public WorkflowSchedule nextRunTime(Long nextRunTime) { + this.nextRunTime = nextRunTime; + return this; + } + + public WorkflowSchedule cronSchedules(List cronSchedules) { + this.cronSchedules = cronSchedules; + return this; + } } \ No newline at end of file diff --git a/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java b/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java new file mode 100644 index 000000000..bf42e4e23 --- /dev/null +++ b/conductor-client/src/test/java/io/orkes/conductor/client/http/SchedulerResourceTest.java @@ -0,0 +1,348 @@ +/* + * Copyright 2026 Conductor Authors. + *

+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License 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 io.orkes.conductor.client.http; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import com.netflix.conductor.client.exception.ConductorClientException; +import com.netflix.conductor.client.http.ConductorClient; + +import io.orkes.conductor.client.SchedulerClient; +import io.orkes.conductor.client.model.CronSchedule; +import io.orkes.conductor.client.model.SaveScheduleRequest; +import io.orkes.conductor.client.model.SearchResultWorkflowScheduleExecution; +import io.orkes.conductor.client.model.TagObject; +import io.orkes.conductor.client.model.WorkflowSchedule; + +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Covers the two behaviour changes to {@link SchedulerResource} that make it portable across + * OSS Conductor (PUT pause/resume, 200-empty get-miss) and Orkes Conductor (GET pause/resume, + * 404 get-miss): the cached PUT-then-405-fallback state machine, and the normalized get-miss + * contract. + */ +class SchedulerResourceTest { + + private MockWebServer server; + private SchedulerResource resource; + + @BeforeEach + void setUp() throws IOException { + server = new MockWebServer(); + server.start(); + resource = new SchedulerResource(new ConductorClient(server.url("/api").toString())); + } + + @AfterEach + void tearDown() throws IOException { + server.shutdown(); + } + + private static MockResponse json(String body) { + return new MockResponse().setHeader("Content-Type", "application/json").setBody(body); + } + + private RecordedRequest takeRequest() throws InterruptedException { + RecordedRequest request = server.takeRequest(5, TimeUnit.SECONDS); + assertNotNull(request, "expected a request to reach the stub server"); + return request; + } + + // ── pause/resume verb fallback ────────────────────────────────────── + + @Test + void pauseSendsPutFirst() throws InterruptedException { + server.enqueue(json("{}")); + + resource.pauseSchedule("s1"); + + RecordedRequest request = takeRequest(); + assertEquals("PUT", request.getMethod()); + assertTrue(request.getPath().endsWith("/scheduler/schedules/s1/pause")); + } + + @Test + void pauseFallsBackToGetOn405AndCachesVerdictAcrossPauseAndResume() throws InterruptedException { + server.enqueue(new MockResponse().setResponseCode(405)); + server.enqueue(json("{}")); + + resource.pauseSchedule("s1"); + + RecordedRequest first = takeRequest(); + assertEquals("PUT", first.getMethod()); + RecordedRequest second = takeRequest(); + assertEquals("GET", second.getMethod()); + assertTrue(second.getPath().endsWith("/scheduler/schedules/s1/pause")); + + // Verdict is cached: resume (a different method) goes straight to GET, no PUT probe. + server.enqueue(json("{}")); + resource.resumeSchedule("s1"); + + RecordedRequest third = takeRequest(); + assertEquals("GET", third.getMethod()); + assertTrue(third.getPath().endsWith("/scheduler/schedules/s1/resume")); + } + + @Test + void pauseRethrowsOn403WithoutFallingBackToGet() throws InterruptedException { + server.enqueue(new MockResponse().setResponseCode(403)); + + ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.pauseSchedule("s1")); + assertEquals(403, ex.getStatus()); + + assertEquals(1, server.getRequestCount()); + } + + @Test + void pauseRethrowsOn404WithoutFallingBackToGet() throws InterruptedException { + server.enqueue(new MockResponse().setResponseCode(404)); + + ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.pauseSchedule("s1")); + assertEquals(404, ex.getStatus()); + + assertEquals(1, server.getRequestCount()); + } + + @Test + void resumeSharesTheSameCachedFallbackVerdictAsPause() throws InterruptedException { + server.enqueue(new MockResponse().setResponseCode(405)); + server.enqueue(json("{}")); + + resource.resumeSchedule("s1"); + takeRequest(); // PUT (405) + RecordedRequest fallback = takeRequest(); + assertEquals("GET", fallback.getMethod()); + + server.enqueue(json("{}")); + resource.pauseSchedule("s2"); + + RecordedRequest thirdCall = takeRequest(); + assertEquals("GET", thirdCall.getMethod()); + } + + // ── reason query param ─────────────────────────────────────────────── + + @Test + void pauseSendsReasonQueryParamWhenProvided() throws InterruptedException { + server.enqueue(json("{}")); + + resource.pauseSchedule("s1", "maintenance window"); + + RecordedRequest request = takeRequest(); + assertTrue(request.getPath().contains("reason=maintenance")); + } + + @Test + void pauseOmitsReasonQueryParamWhenNull() throws InterruptedException { + server.enqueue(json("{}")); + + resource.pauseSchedule("s1"); + + RecordedRequest request = takeRequest(); + assertFalse(request.getPath().contains("reason")); + } + + @Test + void reasonSurvivesTheGetFallback() throws InterruptedException { + server.enqueue(new MockResponse().setResponseCode(405)); + server.enqueue(json("{}")); + + resource.pauseSchedule("s1", "maintenance window"); + + takeRequest(); // PUT (405) + RecordedRequest fallback = takeRequest(); + assertEquals("GET", fallback.getMethod()); + assertTrue(fallback.getPath().contains("reason=maintenance")); + } + + // ── get-miss contract ──────────────────────────────────────────────── + + @Test + void getScheduleThrowsOn404() { + server.enqueue(new MockResponse().setResponseCode(404)); + + ConductorClientException ex = assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing")); + assertEquals(404, ex.getStatus()); + } + + @Test + void getScheduleThrowsOnOssStyleEmptyBody() { + server.enqueue(json("")); + + assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing")); + } + + @Test + void getScheduleThrowsOnEmptyJsonObject() { + server.enqueue(json("{}")); + + assertThrows(ConductorClientException.class, () -> resource.getSchedule("missing")); + } + + @Test + void getScheduleReturnsScheduleOnHitIncludingNextRunTimeAndCronSchedules() { + server.enqueue(json("{" + + "\"name\":\"s1\"," + + "\"cronExpression\":\"0 */5 * * * *\"," + + "\"nextRunTime\":1700000000000," + + "\"cronSchedules\":[{\"cronExpression\":\"0 0 * * * *\",\"zoneId\":\"UTC\"}]" + + "}")); + + WorkflowSchedule schedule = resource.getSchedule("s1"); + + assertEquals("s1", schedule.getName()); + assertEquals(1700000000000L, schedule.getNextRunTime()); + assertEquals(1, schedule.getCronSchedules().size()); + assertEquals("0 0 * * * *", schedule.getCronSchedules().get(0).getCronExpression()); + } + + @Test + void getScheduleToleratesMissingNextRunTime() { + server.enqueue(json("{\"name\":\"s1\",\"cronExpression\":\"0 */5 * * * *\"}")); + + WorkflowSchedule schedule = resource.getSchedule("s1"); + + assertEquals("s1", schedule.getName()); + assertEquals(null, schedule.getNextRunTime()); + } + + // ── save body wire shape ───────────────────────────────────────────── + + @Test + void saveScheduleOmitsCronSchedulesWhenNotSet() throws InterruptedException { + server.enqueue(json("{}")); + + SaveScheduleRequest request = SaveScheduleRequest.builder() + .name("s1") + .cronExpression("0 */5 * * * *") + .build(); + resource.saveSchedule(request); + + RecordedRequest recorded = takeRequest(); + assertFalse(recorded.getBody().readUtf8().contains("cronSchedules")); + } + + @Test + void saveScheduleSendsCronSchedulesWhenSet() throws InterruptedException { + server.enqueue(json("{}")); + + SaveScheduleRequest request = SaveScheduleRequest.builder() + .name("s1") + .cronSchedules(List.of(CronSchedule.builder().cronExpression("0 0 * * * *").zoneId("UTC").build())) + .build(); + resource.saveSchedule(request); + + RecordedRequest recorded = takeRequest(); + String body = recorded.getBody().readUtf8(); + assertTrue(body.contains("\"cronSchedules\"")); + assertTrue(body.contains("0 0 * * * *")); + } + + // ── default-method backward compatibility ─────────────────────────── + + /** A minimal {@link SchedulerClient} implementation predating the reason overload — proves + * the interface addition doesn't break existing implementers. */ + private static class LegacySchedulerClient implements SchedulerClient { + String lastPausedName; + + @Override + public void pauseSchedule(String name) { + lastPausedName = name; + } + + @Override + public void saveSchedule(SaveScheduleRequest saveScheduleRequest) {} + + @Override + public WorkflowSchedule getSchedule(String name) { + return null; + } + + @Override + public List getAllSchedules(String workflowName) { + return null; + } + + @Override + public void deleteSchedule(String name) {} + + @Override + public void resumeSchedule(String name) {} + + @Override + public void pauseAllSchedules() {} + + @Override + public void resumeAllSchedules() {} + + @Override + public com.netflix.conductor.common.model.BulkResponse pauseSchedulers(List schedulerIds) { + return null; + } + + @Override + public com.netflix.conductor.common.model.BulkResponse resumeSchedulers(List schedulerIds) { + return null; + } + + @Override + public List getNextFewSchedules(String cronExpression, Long scheduleStartTime, Long scheduleEndTime, + Integer limit) { + return null; + } + + @Override + public void requeueAllExecutionRecords() {} + + @Override + public SearchResultWorkflowScheduleExecution search(Integer start, Integer size, String sort, + String freeText, String query) { + return null; + } + + @Override + public void setSchedulerTags(List body, String name) {} + + @Override + public void deleteSchedulerTags(List body, String name) {} + + @Override + public List getSchedulerTags(String name) { + return null; + } + } + + @Test + void defaultReasonOverloadDelegatesToOneArgOverloadWhenNotImplemented() { + LegacySchedulerClient client = new LegacySchedulerClient(); + + client.pauseSchedule("s1", "some reason"); + + assertEquals("s1", client.lastPausedName); + } +} diff --git a/conductor-client/src/test/resources/ser_deser_json_string.json b/conductor-client/src/test/resources/ser_deser_json_string.json index fe566189f..c7aa59e6f 100644 --- a/conductor-client/src/test/resources/ser_deser_json_string.json +++ b/conductor-client/src/test/resources/ser_deser_json_string.json @@ -222,12 +222,23 @@ "zoneId": "sample_zoneId", "runCatchupScheduleInstances": true, "scheduleStartTime": 123, - "scheduleEndTime": 123 + "scheduleEndTime": 123, + "cronSchedules": [ + "${CronSchedule}" + ] }, "dependencies": [ - "StartWorkflowRequest" + "StartWorkflowRequest", + "CronSchedule" ] }, + "CronSchedule": { + "content": { + "cronExpression": "sample_cronExpression", + "zoneId": "sample_zoneId" + }, + "dependencies": [] + }, "GrantedAccess": { "content": { "access": [ @@ -1993,11 +2004,16 @@ "createdBy": "sample_createdBy", "name": "sample_name", "zoneId": "sample_zoneId", - "scheduleEndTime": 123 + "scheduleEndTime": 123, + "nextRunTime": 123, + "cronSchedules": [ + "${CronSchedule}" + ] }, "dependencies": [ "Tag", - "StartWorkflowRequest" + "StartWorkflowRequest", + "CronSchedule" ] }, "T": {