Add payload-carrying wakeup to the flow engine via MessagePayload

.gitignore

@@ -40,6 +40,9 @@ bin/
 ### VS Code ###
 .vscode/
 
+### Claude Code ###
+.claude/
+
 # OS generated files #
 .DS_Store
 ehthumbs.db

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,26 +64,26 @@ 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 {}
 
   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 {}
 

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));
     }
   }
 

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();

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,16 +117,21 @@ 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;
     }
     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

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;

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;
+  }
+}

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
+  }
+}

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

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.
    *