diff --git a/.gitignore b/.gitignore index 8327150d..d5aee4ac 100644 --- a/.gitignore +++ b/.gitignore @@ -40,6 +40,9 @@ bin/ ### VS Code ### .vscode/ +### Claude Code ### +.claude/ + # OS generated files # .DS_Store ehthumbs.db diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/Action.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/Action.java index 20a16f94..d64a23cf 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/Action.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/Action.java @@ -3,6 +3,7 @@ import com.netflix.maestro.annotations.Nullable; import com.netflix.maestro.flow.Constants; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.models.Task; /** @@ -32,7 +33,9 @@ record FlowDown() implements Action {} Action FLOW_DOWN = new FlowDown(); - record FlowWakeUp(String flowReference, @Nullable String taskRef, int code) implements Action {} + record FlowWakeUp( + String flowReference, @Nullable String taskRef, int code, MessagePayload payload) + implements Action {} // actions for flow actors record FlowStart(boolean resume) implements Action {} @@ -61,7 +64,7 @@ record TaskDown() implements Action {} Action TASK_DOWN = new TaskDown(); - record TaskWakeUp(@Nullable String taskRef, int code) implements Action {} + record TaskWakeUp(@Nullable String taskRef, int code, MessagePayload payload) implements Action {} // actions for task actors record TaskStart(boolean resume) implements Action {} @@ -69,18 +72,18 @@ record TaskStart(boolean resume) implements Action {} Action TASK_START = new TaskStart(false); Action TASK_RESUME = new TaskStart(true); - record TaskActivate(int code) implements Action {} + record TaskActivate(int code, MessagePayload payload) implements Action {} - Action TASK_ACTIVATE = new TaskActivate(Constants.TASK_PING_CODE); + Action TASK_ACTIVATE = new TaskActivate(Constants.TASK_PING_CODE, MessagePayload.DEFAULT); record TaskStop() implements Action {} Action TASK_STOP = new TaskStop(); // used to wakeup actor, which might be directly scheduled by using TASK_PING constant. - record TaskPing(int code) implements Action {} + record TaskPing(int code, MessagePayload payload) implements Action {} - Action TASK_PING = new TaskPing(Constants.TASK_PING_CODE); + Action TASK_PING = new TaskPing(Constants.TASK_PING_CODE, MessagePayload.DEFAULT); record TaskUpdate(Task updatedTask) implements Action {} diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/FlowActor.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/FlowActor.java index 9b304431..39c190b3 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/FlowActor.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/FlowActor.java @@ -7,6 +7,7 @@ import com.netflix.maestro.flow.Constants; import com.netflix.maestro.flow.engine.ExecutionContext; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.models.Task; import com.netflix.maestro.flow.models.TaskDef; import com.netflix.maestro.utils.Checks; @@ -68,7 +69,7 @@ void runForAction(Action action) { case Action.FlowRefresh fr -> refresh(); case Action.FlowTaskRetry t -> retryTask(t.taskRefName()); case Action.TaskUpdate u -> updateFlow(u.updatedTask()); - case Action.TaskWakeUp w -> wakeup(w.taskRef(), w.code()); + case Action.TaskWakeUp w -> wakeup(w.taskRef(), w.code(), w.payload()); case Action.FlowTimeout ft -> timeoutFlow(); case Action.FlowShutdown sd -> startShutdown(Action.TASK_SHUTDOWN); case Action.TaskDown td -> checkShutdown(); @@ -194,7 +195,10 @@ private void updateFlow(Task updatedTask) { schedule(Action.FLOW_REFRESH, delayForNext(refreshInterval)); if (!updatedTask.isActive()) { schedule( - new Action.TaskWakeUp(updatedTask.referenceTaskName(), Constants.TASK_PING_CODE), + new Action.TaskWakeUp( + updatedTask.referenceTaskName(), + Constants.TASK_PING_CODE, + MessagePayload.DEFAULT), delayForNext(updatedTask.getStartDelayInMillis())); } } else { @@ -229,11 +233,11 @@ private void scheduleRetryableTask(Task task) { // This is the best effort. The task actor might not run while flow thinks it's running or the // actor is shutdown. In those cases, missing wakeup will cause the step won't take any action // during retry backoff delay. Callers have to retry for wakeup. - private void wakeup(@Nullable String taskRef, int code) { + private void wakeup(@Nullable String taskRef, int code, MessagePayload payload) { getMetrics() .counter("num_of_wakeup_flows", getClass(), "forall", taskRef == null ? "true" : "false"); if (taskRef == null) { // wakeup all tasks if taskRef is null - wakeupAll(code); + wakeupAll(code, payload); return; } Task snapshot = flow.getRunningTasks().get(taskRef); @@ -244,20 +248,20 @@ private void wakeup(@Nullable String taskRef, int code) { } else if (!snapshot.isActive()) { snapshot.setActive(true); } - if (code == Constants.TASK_PING_CODE) { + if (code == Constants.TASK_PING_CODE && payload == MessagePayload.DEFAULT) { wakeUpChildActor(taskRef, Action.TASK_ACTIVATE); } else { - wakeUpChildActor(taskRef, new Action.TaskActivate(code)); + wakeUpChildActor(taskRef, new Action.TaskActivate(code, payload)); } } // wake up all tasks but do not activate inactive tasks - private void wakeupAll(int code) { + private void wakeupAll(int code, MessagePayload payload) { dequeRetryActions().forEach(this::retryTask); - if (code == Constants.TASK_PING_CODE) { + if (code == Constants.TASK_PING_CODE && payload == MessagePayload.DEFAULT) { wakeUpChildActors(Action.TASK_PING); } else { - wakeUpChildActors(new Action.TaskPing(code)); + wakeUpChildActors(new Action.TaskPing(code, payload)); } } diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/GroupActor.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/GroupActor.java index 02ce4595..cd540fee 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/GroupActor.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/GroupActor.java @@ -62,7 +62,8 @@ void runForAction(Action action) { case Action.GroupStart g -> startGroup(); case Action.FlowLaunch l -> runFlow(l); case Action.FlowWakeUp w -> - wakeUpChildActor(w.flowReference(), new Action.TaskWakeUp(w.taskRef(), w.code())); + wakeUpChildActor( + w.flowReference(), new Action.TaskWakeUp(w.taskRef(), w.code(), w.payload())); case Action.GroupHeartbeat h -> heartbeat(); case Action.GroupShutdown s -> startShutdown(Action.FLOW_SHUTDOWN); case Action.FlowDown d -> checkShutdown(); diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/TaskActor.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/TaskActor.java index 4164f5ca..afa19416 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/TaskActor.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/actor/TaskActor.java @@ -4,6 +4,7 @@ import com.netflix.maestro.flow.Constants; import com.netflix.maestro.flow.engine.ExecutionContext; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.models.Task; import lombok.extern.slf4j.Slf4j; import org.slf4j.Logger; @@ -44,9 +45,9 @@ void runForAction(Action action) { switch (action) { case Action.TaskStart s -> start(s.resume()); case Action.TaskStop ts -> stop(); - case Action.TaskPing p -> execute(p.code()); - case Action.TaskActivate a -> activate(a.code()); - case Action.TaskTimeout t -> execute(Constants.TIMEOUT_TASK_CODE); + case Action.TaskPing p -> execute(p.code(), p.payload()); + case Action.TaskActivate a -> activate(a.code(), a.payload()); + case Action.TaskTimeout t -> execute(Constants.TIMEOUT_TASK_CODE, MessagePayload.DEFAULT); case Action.TaskShutdown d -> shutdown(); default -> throw new MaestroUnprocessableEntityException( @@ -83,18 +84,20 @@ private void start(boolean resume) { task.setStarted(true); } - private void activate(int code) { + private void activate(int code, MessagePayload payload) { if (!task.isActive()) { task.setActive(true); } // cancel any existing scheduled task ping as the activate call does the same action. var dedupAction = - code == Constants.TASK_PING_CODE ? Action.TASK_PING : new Action.TaskPing(code); + code == Constants.TASK_PING_CODE + ? Action.TASK_PING + : new Action.TaskPing(code, MessagePayload.DEFAULT); var future = getScheduledActions().get(dedupAction); if (future != null) { future.cancel(false); } - execute(code); + execute(code, payload); } private void stop() { @@ -114,7 +117,7 @@ private void stop() { * check will discover it as it also checks the parent status. Maestro engine makes sure to flip * the active flag if the task should not execution (e.g. NOT_CREATED case). */ - private void execute(int code) { + private void execute(int code, MessagePayload payload) { if (!task.isStarted()) { LOG.info("Flow task [{}] is not started yet, skip execution", name); return; @@ -122,8 +125,13 @@ private void execute(int code) { boolean changed = false; if (task.isActive()) { // execution only for active tasks task.setCode(code); - changed = getContext().execute(flow, task); - task.setCode(Constants.TASK_PING_CODE); + task.setMessagePayload(payload); + try { + changed = getContext().execute(flow, task); + } finally { + task.setCode(Constants.TASK_PING_CODE); + task.setMessagePayload(MessagePayload.DEFAULT); + } } if (task.getStatus().isTerminal()) { terminateNow(); // if terminal state, then stop diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/engine/FlowExecutor.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/engine/FlowExecutor.java index 0d15f6d8..5c8f65a2 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/engine/FlowExecutor.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/engine/FlowExecutor.java @@ -7,6 +7,7 @@ import com.netflix.maestro.flow.models.Flow; import com.netflix.maestro.flow.models.FlowDef; import com.netflix.maestro.flow.models.FlowGroup; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.utils.ExecutionHelper; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -209,9 +210,29 @@ private Actor getOrCreateNewGroup(long groupId) { */ public boolean wakeUp( long groupId, String flowReference, @Nullable String taskReference, int code) { + return wakeUp(groupId, flowReference, taskReference, code, MessagePayload.DEFAULT); + } + + /** + * Wake up a flow or a task with a payload. + * + * @param groupId group id to group flow instances + * @param flowReference flow reference + * @param taskReference task reference. If it is null, it wakes up all the tasks in the flow. + * @param code notification signaling code, which is passed to the task when it is woken up + * @param payload payload passed to the task when it is woken up + * @return true if the flow or task is woken up successfully, otherwise, false. The caller can + * retry based on the returned result. + */ + public boolean wakeUp( + long groupId, + String flowReference, + @Nullable String taskReference, + int code, + MessagePayload payload) { Actor groupActor = groupActors.get(groupId); if (groupActor != null && groupActor.isRunning()) { - groupActor.post(new Action.FlowWakeUp(flowReference, taskReference, code)); + groupActor.post(new Action.FlowWakeUp(flowReference, taskReference, code, payload)); return true; } return false; diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/models/DefaultMessagePayload.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/DefaultMessagePayload.java new file mode 100644 index 00000000..33f4328d --- /dev/null +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/DefaultMessagePayload.java @@ -0,0 +1,35 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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 com.netflix.maestro.flow.models; + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonPropertyOrder; +import com.fasterxml.jackson.databind.PropertyNamingStrategies; +import com.fasterxml.jackson.databind.annotation.JsonNaming; +import lombok.Data; +import lombok.ToString; + +/** Default empty message payload carrying no data. */ +@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) +@JsonInclude(JsonInclude.Include.NON_NULL) +@JsonPropertyOrder( + value = {"type"}, + alphabetic = true) +@Data +@ToString +public class DefaultMessagePayload implements MessagePayload { + @Override + public Type getType() { + return Type.DEFAULT; + } +} diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/models/MessagePayload.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/MessagePayload.java new file mode 100644 index 00000000..2c78333a --- /dev/null +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/MessagePayload.java @@ -0,0 +1,38 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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 com.netflix.maestro.flow.models; + +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; + +/** Payload carried by a message to a task. */ +@JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + property = "type", + include = JsonTypeInfo.As.EXISTING_PROPERTY, + defaultImpl = DefaultMessagePayload.class) +@JsonSubTypes({@JsonSubTypes.Type(name = "DEFAULT", value = DefaultMessagePayload.class)}) +@SuppressWarnings("PMD.ImplicitFunctionalInterface") +public interface MessagePayload { + /** empty payload used when a message carries no data. */ + MessagePayload DEFAULT = new DefaultMessagePayload(); + + /** Get message payload type info. */ + Type getType(); + + /** supported message payload types. */ + enum Type { + /** default empty payload. */ + DEFAULT + } +} diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/models/Task.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/Task.java index f18921a1..0497f3f8 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/models/Task.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/models/Task.java @@ -71,6 +71,7 @@ public enum Status { private boolean active = true; // flag to indicate if a running task is active or not private boolean started; // flag to indicate if a running task is started or not private int code = Constants.TASK_PING_CODE; // wakeup code for custom signaling + private MessagePayload messagePayload = MessagePayload.DEFAULT; // message payload for signaling private Long startTime; // used to record the execution start time private Long timeoutInMillis; // keep unset timeout value from maestro engine private Long endTime; // used to record the execution end time diff --git a/maestro-flow/src/main/java/com/netflix/maestro/flow/runtime/FlowOperation.java b/maestro-flow/src/main/java/com/netflix/maestro/flow/runtime/FlowOperation.java index 523a5d10..bee27553 100644 --- a/maestro-flow/src/main/java/com/netflix/maestro/flow/runtime/FlowOperation.java +++ b/maestro-flow/src/main/java/com/netflix/maestro/flow/runtime/FlowOperation.java @@ -2,6 +2,7 @@ import com.netflix.maestro.flow.Constants; import com.netflix.maestro.flow.models.FlowDef; +import com.netflix.maestro.flow.models.MessagePayload; import java.util.Map; import java.util.Set; @@ -63,6 +64,20 @@ default boolean wakeUp(long groupId, String flowReference, String taskReference) return wakeUp(groupId, flowReference, taskReference, Constants.TASK_PING_CODE); } + /** + * Wake up a single task in a flow for a group with a payload. + * + * @param groupId group id to group flow instances + * @param flowReference reference is what the caller would like to refer a flow + * @param taskReference task reference + * @param code notification signaling code, which is passed to the task when it is woken up + * @param payload payload passed to the task when it is woken up + * @return true if the task is woken up successfully. Otherwise, false. The caller can retry based + * on the returned result. + */ + boolean wakeUp( + long groupId, String flowReference, String taskReference, int code, MessagePayload payload); + /** * Wake up all the tasks in a list of flows for a group. * diff --git a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/FlowActorTest.java b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/FlowActorTest.java index 9bb86236..cbf99b09 100644 --- a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/FlowActorTest.java +++ b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/FlowActorTest.java @@ -26,6 +26,7 @@ import com.netflix.maestro.AssertHelper; import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.models.Task; import com.netflix.maestro.flow.models.TaskDef; import java.util.List; @@ -269,7 +270,7 @@ public void testTaskUpdateForRunningInactiveTask() { flowActor.runForAction(new Action.TaskUpdate(task2)); assertEquals( - Set.of(Action.FLOW_REFRESH, new Action.TaskWakeUp("task1", 0)), + Set.of(Action.FLOW_REFRESH, new Action.TaskWakeUp("task1", 0, MessagePayload.DEFAULT)), flowActor.getScheduledActions().keySet()); assertEquals(task2, flow.getRunningTasks().get("task1")); } @@ -318,7 +319,7 @@ public void testTaskWakeUpWithRunningTask() { Task task2 = flow.getRunningTasks().get("task1"); task2.setActive(false); - flowActor.runForAction(new Action.TaskWakeUp("task1", 0)); + flowActor.runForAction(new Action.TaskWakeUp("task1", 0, MessagePayload.DEFAULT)); assertTrue(task2.isActive()); verifyActions(flowActor.getChild("task1"), Action.TASK_ACTIVATE); } @@ -330,7 +331,7 @@ public void testTaskWakeUpWithQueuedTask() { @Test public void testTaskWakeUpWithCustomActionCode() { - testTaskWakeUp(123, new Action.TaskActivate(123)); + testTaskWakeUp(123, new Action.TaskActivate(123, MessagePayload.DEFAULT)); } private void testTaskWakeUp(int code, Action expectedAction) { @@ -345,7 +346,7 @@ private void testTaskWakeUp(int code, Action expectedAction) { flowActor.runForAction(Action.FLOW_RESUME); assertFalse(flowActor.containsChild("task1")); - flowActor.runForAction(new Action.TaskWakeUp("task1", code)); + flowActor.runForAction(new Action.TaskWakeUp("task1", code, MessagePayload.DEFAULT)); verify(future, times(1)).cancel(false); verifyActions(flowActor.getChild("task1"), Action.TASK_START, expectedAction); assertTrue(flowActor.containsChild("task1")); @@ -360,7 +361,7 @@ public void testTaskWakeUpForAllFlowTasks() { @Test public void testTaskWakeUpAllWithCustomActionCode() { - testTaskWakeUpForAll(123, new Action.TaskPing(123)); + testTaskWakeUpForAll(123, new Action.TaskPing(123, MessagePayload.DEFAULT)); } public void testTaskWakeUpForAll(int code, Action expectedAction) { @@ -379,7 +380,7 @@ public void testTaskWakeUpForAll(int code, Action expectedAction) { assertFalse(flowActor.containsChild("task1")); assertTrue(flowActor.containsChild("task2")); - flowActor.runForAction(new Action.TaskWakeUp(null, code)); + flowActor.runForAction(new Action.TaskWakeUp(null, code, MessagePayload.DEFAULT)); verify(future, times(1)).cancel(false); verifyActions(flowActor.getChild("task1"), Action.TASK_START, expectedAction); verifyActions(flowActor.getChild("task2"), Action.TASK_RESUME, expectedAction); @@ -436,7 +437,7 @@ public void testUnexpectedAction() { AssertHelper.assertThrows( "should throw for unexpected action", MaestroUnprocessableEntityException.class, - "Unexpected action: [TaskPing[code=0]] for flow ", + "Unexpected action: [TaskPing[code=0, payload=DefaultMessagePayload()]] for flow ", () -> flowActor.runForAction(Action.TASK_PING)); } diff --git a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/GroupActorTest.java b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/GroupActorTest.java index 131033a0..e19ae045 100644 --- a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/GroupActorTest.java +++ b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/GroupActorTest.java @@ -25,6 +25,7 @@ import com.netflix.maestro.AssertHelper; import com.netflix.maestro.exceptions.MaestroUnprocessableEntityException; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import java.util.List; import java.util.Set; import org.junit.Before; @@ -120,7 +121,8 @@ public void testRunForActionFlowLaunchStart() { @Test public void testRunForActionFlowWakeUp() { - groupActor.runForAction(new Action.FlowWakeUp(flow.getReference(), "taskRef", 0)); + groupActor.runForAction( + new Action.FlowWakeUp(flow.getReference(), "taskRef", 0, MessagePayload.DEFAULT)); assertNull(groupActor.getChild(flow.getReference())); groupActor.runForAction(new Action.FlowLaunch(flow, false)); @@ -128,11 +130,13 @@ public void testRunForActionFlowWakeUp() { var child = groupActor.getChild(flow.getReference()); verifyActions(child, Action.FLOW_START); - groupActor.runForAction(new Action.FlowWakeUp(flow.getReference(), "taskRef", 0)); - verifyActions(child, new Action.TaskWakeUp("taskRef", 0)); + groupActor.runForAction( + new Action.FlowWakeUp(flow.getReference(), "taskRef", 0, MessagePayload.DEFAULT)); + verifyActions(child, new Action.TaskWakeUp("taskRef", 0, MessagePayload.DEFAULT)); - groupActor.runForAction(new Action.FlowWakeUp(flow.getReference(), "taskRef", 123)); - verifyActions(child, new Action.TaskWakeUp("taskRef", 123)); + groupActor.runForAction( + new Action.FlowWakeUp(flow.getReference(), "taskRef", 123, MessagePayload.DEFAULT)); + verifyActions(child, new Action.TaskWakeUp("taskRef", 123, MessagePayload.DEFAULT)); } @Test diff --git a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/TaskActorTest.java b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/TaskActorTest.java index 02f27814..12ad5194 100644 --- a/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/TaskActorTest.java +++ b/maestro-flow/src/test/java/com/netflix/maestro/flow/actor/TaskActorTest.java @@ -15,13 +15,16 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.netflix.maestro.flow.models.DefaultMessagePayload; import com.netflix.maestro.flow.models.Flow; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.models.Task; import com.netflix.maestro.flow.models.TaskDef; import java.util.Set; @@ -155,8 +158,10 @@ public void testRunForActionTaskActivate() { @Test public void testRunForActionTaskActivateWithoutCancel() { var mockFuture = Mockito.mock(ScheduledFuture.class); - taskActor.getScheduledActions().put(new Action.TaskPing(123), mockFuture); - verifyExecute(Action.TASK_ACTIVATE, false, new Action.TaskPing(123)); + taskActor + .getScheduledActions() + .put(new Action.TaskPing(123, MessagePayload.DEFAULT), mockFuture); + verifyExecute(Action.TASK_ACTIVATE, false, new Action.TaskPing(123, MessagePayload.DEFAULT)); verify(mockFuture, times(0)).cancel(false); } @@ -179,8 +184,8 @@ private void verifyExecute(Action action, boolean activeFlag, Action extra) { @Test public void testExecuteNotStarted() { taskActor.runForAction(Action.TASK_PING); - taskActor.runForAction(new Action.TaskPing(123)); - taskActor.runForAction(new Action.TaskActivate(123)); + taskActor.runForAction(new Action.TaskPing(123, MessagePayload.DEFAULT)); + taskActor.runForAction(new Action.TaskActivate(123, MessagePayload.DEFAULT)); taskActor.runForAction(Action.TASK_TIMEOUT); // Should not execute if task is not started @@ -199,12 +204,29 @@ public void testExecuteWithCustomCode() { assertEquals(123, taskArg.getCode()); return false; }); - taskActor.runForAction(new Action.TaskActivate(123)); + taskActor.runForAction(new Action.TaskActivate(123, MessagePayload.DEFAULT)); verify(context, times(1)).execute(any(), any()); assertEquals(0, task.getCode()); // reset action code } + @Test + public void testExecuteWithPayload() { + task.setStarted(true); + MessagePayload payload = new DefaultMessagePayload(); + when(context.execute(flow, task)) + .thenAnswer( + invocation -> { + Task taskArg = invocation.getArgument(1); + assertSame(payload, taskArg.getMessagePayload()); + return false; + }); + taskActor.runForAction(new Action.TaskActivate(123, payload)); + + verify(context, times(1)).execute(any(), any()); + assertSame(MessagePayload.DEFAULT, task.getMessagePayload()); // reset payload + } + @Test public void testRunForActionTaskTimeout() { verifyExecute(Action.TASK_TIMEOUT, true, null); diff --git a/maestro-flow/src/test/java/com/netflix/maestro/flow/engine/FlowExecutorTest.java b/maestro-flow/src/test/java/com/netflix/maestro/flow/engine/FlowExecutorTest.java index 00ff499e..93021a5b 100644 --- a/maestro-flow/src/test/java/com/netflix/maestro/flow/engine/FlowExecutorTest.java +++ b/maestro-flow/src/test/java/com/netflix/maestro/flow/engine/FlowExecutorTest.java @@ -27,6 +27,7 @@ import com.netflix.maestro.flow.models.Flow; import com.netflix.maestro.flow.models.FlowDef; import com.netflix.maestro.flow.models.FlowGroup; +import com.netflix.maestro.flow.models.MessagePayload; import java.util.Map; import org.junit.Before; import org.junit.Test; @@ -109,4 +110,16 @@ public void testWakeUp() { assertFalse(executor.wakeUp(2L, "wf-1", "task1", -1)); assertTrue(executor.wakeUp(1L, "wf-2", null, 123)); } + + @Test + public void testWakeUpWithPayload() { + assertFalse(executor.wakeUp(1L, "wf-1", "task1", 0, MessagePayload.DEFAULT)); + when(context.trySaveGroup(1, "test-address")) + .thenReturn(new FlowGroup(1, 1, "test-address", 12345)); + + executor.startFlow(1, "test-id", "wf-1", new FlowDef(), Map.of()); + assertTrue(executor.wakeUp(1L, "wf-1", "task1", 0, MessagePayload.DEFAULT)); + assertFalse(executor.wakeUp(2L, "wf-1", "task1", 0, MessagePayload.DEFAULT)); + assertTrue(executor.wakeUp(1L, "wf-2", "task1", 123, MessagePayload.DEFAULT)); + } } diff --git a/maestro-flow/src/test/java/com/netflix/maestro/flow/models/MessagePayloadTest.java b/maestro-flow/src/test/java/com/netflix/maestro/flow/models/MessagePayloadTest.java new file mode 100644 index 00000000..614114e6 --- /dev/null +++ b/maestro-flow/src/test/java/com/netflix/maestro/flow/models/MessagePayloadTest.java @@ -0,0 +1,37 @@ +/* + * Copyright 2026 Netflix, Inc. + * + * 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 com.netflix.maestro.flow.models; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import com.netflix.maestro.MaestroBaseTest; +import org.junit.Test; + +public class MessagePayloadTest extends MaestroBaseTest { + + @Test + public void testRoundTripSerde() throws Exception { + MessagePayload actual = + MAPPER.readValue(MAPPER.writeValueAsString(MessagePayload.DEFAULT), MessagePayload.class); + assertTrue(actual instanceof DefaultMessagePayload); + assertEquals(MessagePayload.Type.DEFAULT, actual.getType()); + } + + @Test + public void testDeserializeEmptyBodyToDefault() throws Exception { + MessagePayload actual = MAPPER.readValue("{}", MessagePayload.class); + assertTrue(actual instanceof DefaultMessagePayload); + assertEquals(MessagePayload.Type.DEFAULT, actual.getType()); + } +} diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/controllers/FlowEngineController.java b/maestro-server/src/main/java/com/netflix/maestro/server/controllers/FlowEngineController.java index 4b19fe49..796ea23d 100644 --- a/maestro-server/src/main/java/com/netflix/maestro/server/controllers/FlowEngineController.java +++ b/maestro-server/src/main/java/com/netflix/maestro/server/controllers/FlowEngineController.java @@ -15,8 +15,10 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.netflix.maestro.engine.execution.WorkflowSummary; import com.netflix.maestro.flow.models.FlowDef; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.runtime.FlowOperation; import com.netflix.maestro.models.Constants; +import com.netflix.maestro.utils.IdHelper; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.tags.Tag; import jakarta.validation.Valid; @@ -44,6 +46,8 @@ produces = MediaType.APPLICATION_JSON_VALUE, consumes = MediaType.APPLICATION_JSON_VALUE) public class FlowEngineController { + private static final String GROUP_ID = "groupId"; + private static final String CODE = "code"; private final FlowOperation flowOperation; private final ObjectMapper objectMapper; @@ -63,7 +67,7 @@ public record StartFlowRequest(String flowId, FlowDef flowDef, Map, so the workflow summary @@ -85,10 +89,10 @@ public String startFlow( consumes = MediaType.ALL_VALUE) @Operation(summary = "Wake up a specific flow task in a group") public Boolean wakeUp( - @PathVariable("groupId") long groupId, + @PathVariable(GROUP_ID) long groupId, @Valid @NotNull @PathVariable("flowReference") String flowReference, @Valid @NotNull @PathVariable("taskReference") String taskReference, - @PathVariable("code") int code) { + @PathVariable(CODE) int code) { return flowOperation.wakeUp(groupId, flowReference, taskReference, code); } @@ -97,9 +101,40 @@ public Boolean wakeUp( consumes = MediaType.APPLICATION_JSON_VALUE) @Operation(summary = "Wake up all the tasks in a list of flows in a group") public Boolean wakeUp( - @PathVariable("groupId") long groupId, - @PathVariable("code") int code, + @PathVariable(GROUP_ID) long groupId, + @PathVariable(CODE) int code, @Valid @NotNull @RequestBody Set refs) { return flowOperation.wakeUp(groupId, refs, code); } + + @PostMapping( + value = "/{groupId}/flows/{flowReference}/tasks/{taskReference}/message/{code}", + consumes = MediaType.APPLICATION_JSON_VALUE) + @Operation(summary = "Send a message payload to a specific flow task in a group") + public Boolean wakeUp( + @PathVariable(GROUP_ID) long groupId, + @Valid @NotNull @PathVariable("flowReference") String flowReference, + @Valid @NotNull @PathVariable("taskReference") String taskReference, + @PathVariable(CODE) int code, + @Valid @NotNull @RequestBody MessagePayload payload) { + return flowOperation.wakeUp(groupId, flowReference, taskReference, code, payload); + } + + @PostMapping( + value = + "/workflows/{workflowId}/instances/{instanceId}/runs/{runId}/groups/{groupInfo}/steps/{stepId}/message/{code}", + consumes = MediaType.APPLICATION_JSON_VALUE) + @Operation(summary = "Send a message payload to a step by workflow reference") + public Boolean wakeUpWorkflowStep( + @Valid @NotNull @PathVariable("workflowId") String workflowId, + @PathVariable("instanceId") long instanceId, + @PathVariable("runId") long runId, + @PathVariable("groupInfo") long groupInfo, + @Valid @NotNull @PathVariable("stepId") String stepId, + @PathVariable(CODE) int code, + @Valid @NotNull @RequestBody MessagePayload payload) { + String flowReference = IdHelper.deriveFlowRef(workflowId, instanceId, runId); + long groupId = IdHelper.deriveGroupId(flowReference, groupInfo); + return flowOperation.wakeUp(groupId, flowReference, stepId, code, payload); + } } diff --git a/maestro-server/src/main/java/com/netflix/maestro/server/runtime/RestBasedFlowOperation.java b/maestro-server/src/main/java/com/netflix/maestro/server/runtime/RestBasedFlowOperation.java index 885d9e07..00fde265 100644 --- a/maestro-server/src/main/java/com/netflix/maestro/server/runtime/RestBasedFlowOperation.java +++ b/maestro-server/src/main/java/com/netflix/maestro/server/runtime/RestBasedFlowOperation.java @@ -17,6 +17,7 @@ import com.netflix.maestro.flow.engine.FlowExecutor; import com.netflix.maestro.flow.models.FlowDef; import com.netflix.maestro.flow.models.FlowGroup; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.properties.FlowEngineProperties; import com.netflix.maestro.flow.runtime.FlowOperation; import java.util.Map; @@ -100,6 +101,31 @@ public boolean wakeUp(long groupId, String flowReference, String taskReference, } } + @Override + public boolean wakeUp( + long groupId, String flowReference, String taskReference, int code, MessagePayload payload) { + try { + FlowGroup group = loadFlowGroup(groupId); + if (group == null || localAddress.equals(group.address())) { + return flowExecutor.wakeUp(groupId, flowReference, taskReference, code, payload); + } else { + return Boolean.TRUE.equals( + restTemplate.postForObject( + group.address() + + "/api/v3/groups/{groupId}/flows/{flowReference}/tasks/{taskReference}/message/{code}", + payload, + Boolean.class, + groupId, + flowReference, + taskReference, + code)); + } + } catch (MaestroRetryableError e) { + addressCache.remove(groupId); + throw e; + } + } + @Override public boolean wakeUp(long groupId, Set refs, int code) { try { diff --git a/maestro-server/src/test/java/com/netflix/maestro/server/controllers/FlowEngineControllerTest.java b/maestro-server/src/test/java/com/netflix/maestro/server/controllers/FlowEngineControllerTest.java index f0cf965a..c3eaefa4 100644 --- a/maestro-server/src/test/java/com/netflix/maestro/server/controllers/FlowEngineControllerTest.java +++ b/maestro-server/src/test/java/com/netflix/maestro/server/controllers/FlowEngineControllerTest.java @@ -12,6 +12,10 @@ */ package com.netflix.maestro.server.controllers; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.times; @@ -20,13 +24,15 @@ import com.netflix.maestro.MaestroBaseTest; import com.netflix.maestro.engine.execution.WorkflowSummary; +import com.netflix.maestro.flow.models.DefaultMessagePayload; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.runtime.FlowOperation; import com.netflix.maestro.models.Constants; import com.netflix.maestro.server.controllers.FlowEngineController.StartFlowRequest; +import com.netflix.maestro.utils.IdHelper; import java.util.HashMap; import java.util.Map; import java.util.Set; -import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentCaptor; @@ -60,7 +66,7 @@ public void testStartFlow() { verify(mockFlowOperation, times(1)) .startFlow(eq(groupId), eq(flowId), eq(flowReference), isNull(), eq(Map.of())); - Assert.assertEquals(expectedResult, result); + assertEquals(expectedResult, result); } @Test @@ -73,7 +79,7 @@ public void testStartFlowReTypesWorkflowSummaryFromMap() throws Exception { WorkflowSummary summary = loadObject("fixtures/parameters/sample-wf-summary-params.json", WorkflowSummary.class); Object summaryAsMap = MAPPER.convertValue(summary, Map.class); - Assert.assertFalse(summaryAsMap instanceof WorkflowSummary); + assertFalse(summaryAsMap instanceof WorkflowSummary); Map flowInput = new HashMap<>(); flowInput.put(Constants.WORKFLOW_SUMMARY_FIELD, summaryAsMap); @@ -85,8 +91,8 @@ public void testStartFlowReTypesWorkflowSummaryFromMap() throws Exception { verify(mockFlowOperation, times(1)) .startFlow(eq(groupId), eq(flowId), eq(flowReference), isNull(), captor.capture()); Object handed = captor.getValue().get(Constants.WORKFLOW_SUMMARY_FIELD); - Assert.assertTrue(handed instanceof WorkflowSummary); - Assert.assertEquals(MAPPER.writeValueAsString(summary), MAPPER.writeValueAsString(handed)); + assertTrue(handed instanceof WorkflowSummary); + assertEquals(MAPPER.writeValueAsString(summary), MAPPER.writeValueAsString(handed)); } @Test @@ -105,7 +111,7 @@ public void testStartFlowLeavesTypedWorkflowSummaryUntouched() { verify(mockFlowOperation, times(1)) .startFlow(eq(groupId), eq(flowId), eq(flowReference), isNull(), captor.capture()); // already typed, so the same instance passes through without a conversion. - Assert.assertSame(summary, captor.getValue().get(Constants.WORKFLOW_SUMMARY_FIELD)); + assertSame(summary, captor.getValue().get(Constants.WORKFLOW_SUMMARY_FIELD)); } @Test @@ -119,7 +125,7 @@ public void testWakeUpSingleTask() { verify(mockFlowOperation, times(1)) .wakeUp(eq(groupId), eq(flowReference), eq(taskReference), eq(code)); - assert result.equals(true); + assertTrue(result); } @Test @@ -130,6 +136,46 @@ public void testWakeUpMultipleFlows() { Boolean result = flowEngineController.wakeUp(groupId, code, refs); verify(mockFlowOperation, times(1)).wakeUp(eq(groupId), eq(refs), eq(code)); - assert result.equals(true); + assertTrue(result); + } + + @Test + public void testWakeUpSingleTaskWithPayload() { + String flowReference = "test-flow-ref"; + String taskReference = "test-task-ref"; + MessagePayload payload = new DefaultMessagePayload(); + when(mockFlowOperation.wakeUp( + eq(groupId), eq(flowReference), eq(taskReference), eq(code), eq(payload))) + .thenReturn(true); + + Boolean result = + flowEngineController.wakeUp(groupId, flowReference, taskReference, code, payload); + + verify(mockFlowOperation, times(1)) + .wakeUp(eq(groupId), eq(flowReference), eq(taskReference), eq(code), eq(payload)); + assertTrue(result); + } + + @Test + public void testWakeUpWorkflowStep() { + String workflowId = "test-wf"; + long instanceId = 1L; + long runId = 1L; + long groupInfo = 2L; + String stepId = "step1"; + MessagePayload payload = new DefaultMessagePayload(); + String flowReference = IdHelper.deriveFlowRef(workflowId, instanceId, runId); + long derivedGroupId = IdHelper.deriveGroupId(flowReference, groupInfo); + when(mockFlowOperation.wakeUp( + eq(derivedGroupId), eq(flowReference), eq(stepId), eq(code), eq(payload))) + .thenReturn(true); + + Boolean result = + flowEngineController.wakeUpWorkflowStep( + workflowId, instanceId, runId, groupInfo, stepId, code, payload); + + verify(mockFlowOperation, times(1)) + .wakeUp(eq(derivedGroupId), eq(flowReference), eq(stepId), eq(code), eq(payload)); + assertTrue(result); } } diff --git a/maestro-server/src/test/java/com/netflix/maestro/server/runtime/RestBasedFlowOperationTest.java b/maestro-server/src/test/java/com/netflix/maestro/server/runtime/RestBasedFlowOperationTest.java index b7c62fb3..d825856d 100644 --- a/maestro-server/src/test/java/com/netflix/maestro/server/runtime/RestBasedFlowOperationTest.java +++ b/maestro-server/src/test/java/com/netflix/maestro/server/runtime/RestBasedFlowOperationTest.java @@ -28,7 +28,9 @@ import com.netflix.maestro.MaestroBaseTest; import com.netflix.maestro.flow.dao.MaestroFlowDao; import com.netflix.maestro.flow.engine.FlowExecutor; +import com.netflix.maestro.flow.models.DefaultMessagePayload; import com.netflix.maestro.flow.models.FlowGroup; +import com.netflix.maestro.flow.models.MessagePayload; import com.netflix.maestro.flow.properties.FlowEngineProperties; import java.util.Set; import org.junit.Before; @@ -171,4 +173,46 @@ public void testWakeUpLocalWhenGroupAddressMatchesLocal() { verify(restTemplate, never()) .postForObject(anyString(), any(), eq(Boolean.class), (Object[]) any()); } + + @Test + public void testWakeUpWithPayloadLocalWhenGroupAddressMatchesLocal() { + FlowGroup localGroup = new FlowGroup(groupId, 1L, "localhost:8080", System.currentTimeMillis()); + when(flowDao.getGroup(groupId)).thenReturn(localGroup); + MessagePayload payload = new DefaultMessagePayload(); + when(flowExecutor.wakeUp(groupId, flowReference, taskReference, actionCode, payload)) + .thenReturn(true); + + boolean result = + flowOperation.wakeUp(groupId, flowReference, taskReference, actionCode, payload); + + assertTrue(result); + verify(flowExecutor, times(1)) + .wakeUp(groupId, flowReference, taskReference, actionCode, payload); + verify(restTemplate, never()) + .postForObject(anyString(), any(), eq(Boolean.class), (Object[]) any()); + } + + @Test + public void testWakeUpWithPayloadRoutesToRemotePod() { + String remoteAddress = "http://remote-pod:8080"; + FlowGroup remoteGroup = new FlowGroup(groupId, 1L, remoteAddress, System.currentTimeMillis()); + MessagePayload payload = new DefaultMessagePayload(); + when(flowDao.getGroup(groupId)).thenReturn(remoteGroup); + when(restTemplate.postForObject( + anyString(), eq(payload), eq(Boolean.class), any(), any(), any(), any())) + .thenReturn(Boolean.TRUE); + + boolean result = + flowOperation.wakeUp(groupId, flowReference, taskReference, actionCode, payload); + + assertTrue(result); + verify(flowExecutor, never()).wakeUp(anyLong(), anyString(), any(), anyInt(), any()); + ArgumentCaptor urlCaptor = ArgumentCaptor.forClass(String.class); + verify(restTemplate) + .postForObject( + urlCaptor.capture(), eq(payload), eq(Boolean.class), any(), any(), any(), any()); + assertTrue(urlCaptor.getValue().startsWith(remoteAddress)); + assertTrue(urlCaptor.getValue().contains("/tasks/")); + assertTrue(urlCaptor.getValue().contains("/message/")); + } }