Re-type workflow summary once at flow start ingress to avoid per-poll deserialization

maestro-server/src/main/java/com/netflix/maestro/server/controllers/FlowEngineController.java

@@ -12,15 +12,19 @@
  */
 package com.netflix.maestro.server.controllers;
 
+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.runtime.FlowOperation;
+import com.netflix.maestro.models.Constants;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
 import java.util.Map;
 import java.util.Set;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -42,10 +46,14 @@
 public class FlowEngineController {
 
   private final FlowOperation flowOperation;
+  private final ObjectMapper objectMapper;
 
   @Autowired
-  public FlowEngineController(FlowOperation flowOperation) {
+  public FlowEngineController(
+      FlowOperation flowOperation,
+      @Qualifier(Constants.MAESTRO_QUALIFIER) ObjectMapper objectMapper) {
     this.flowOperation = flowOperation;
+    this.objectMapper = objectMapper;
   }
 
   public record StartFlowRequest(String flowId, FlowDef flowDef, Map<String, Object> flowInput) {}
@@ -58,8 +66,18 @@ public String startFlow(
       @PathVariable("groupId") long groupId,
       @Valid @NotNull @PathVariable("flowReference") String flowReference,
       @Valid @NotNull @RequestBody StartFlowRequest request) {
+    // The request body was JSON-deserialized into a Map<String, Object>, so the workflow summary
+    // value is an untyped LinkedHashMap. Re-materialize it once here into a WorkflowSummary so the
+    // flow runs with the typed object and downstream reads cast instead of converting per poll.
+    Map<String, Object> flowInput = request.flowInput();
+    Object summary = flowInput.get(Constants.WORKFLOW_SUMMARY_FIELD);
+    if (summary != null && !(summary instanceof WorkflowSummary)) {
+      flowInput.put(
+          Constants.WORKFLOW_SUMMARY_FIELD,
+          objectMapper.convertValue(summary, WorkflowSummary.class));
+    }
     return flowOperation.startFlow(
-        groupId, request.flowId(), flowReference, request.flowDef(), request.flowInput());
+        groupId, request.flowId(), flowReference, request.flowDef(), flowInput);
   }
 
   @PostMapping(

maestro-server/src/test/java/com/netflix/maestro/server/controllers/FlowEngineControllerTest.java

@@ -19,13 +19,17 @@
 import static org.mockito.Mockito.when;
 
 import com.netflix.maestro.MaestroBaseTest;
+import com.netflix.maestro.engine.execution.WorkflowSummary;
 import com.netflix.maestro.flow.runtime.FlowOperation;
+import com.netflix.maestro.models.Constants;
 import com.netflix.maestro.server.controllers.FlowEngineController.StartFlowRequest;
+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;
 import org.mockito.Mock;
 
 public class FlowEngineControllerTest extends MaestroBaseTest {
@@ -38,7 +42,7 @@ public class FlowEngineControllerTest extends MaestroBaseTest {
 
   @Before
   public void before() {
-    this.flowEngineController = new FlowEngineController(mockFlowOperation);
+    this.flowEngineController = new FlowEngineController(mockFlowOperation, MAPPER);
   }
 
   @Test
@@ -59,6 +63,51 @@ public void testStartFlow() {
     Assert.assertEquals(expectedResult, result);
   }
 
+  @Test
+  public void testStartFlowReTypesWorkflowSummaryFromMap() throws Exception {
+    String flowReference = "test-flow-ref";
+    String flowId = "test-flow-id";
+
+    // Simulate the cross-node path: the request body was deserialized into a Map<String, Object>,
+    // so the workflow summary value arrives as an untyped Map, not a WorkflowSummary.
+    WorkflowSummary summary =
+        loadObject("fixtures/parameters/sample-wf-summary-params.json", WorkflowSummary.class);
+    Object summaryAsMap = MAPPER.convertValue(summary, Map.class);
+    Assert.assertFalse(summaryAsMap instanceof WorkflowSummary);
+    Map<String, Object> flowInput = new HashMap<>();
+    flowInput.put(Constants.WORKFLOW_SUMMARY_FIELD, summaryAsMap);
+
+    flowEngineController.startFlow(
+        groupId, flowReference, new StartFlowRequest(flowId, null, flowInput));
+
+    // The controller must hand a typed WorkflowSummary to the flow, not the raw map.
+    ArgumentCaptor<Map<String, Object>> captor = ArgumentCaptor.forClass(Map.class);
+    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));
+  }
+
+  @Test
+  public void testStartFlowLeavesTypedWorkflowSummaryUntouched() {
+    String flowReference = "test-flow-ref";
+    String flowId = "test-flow-id";
+
+    WorkflowSummary summary = new WorkflowSummary();
+    Map<String, Object> flowInput = new HashMap<>();
+    flowInput.put(Constants.WORKFLOW_SUMMARY_FIELD, summary);
+
+    flowEngineController.startFlow(
+        groupId, flowReference, new StartFlowRequest(flowId, null, flowInput));
+
+    ArgumentCaptor<Map<String, Object>> captor = ArgumentCaptor.forClass(Map.class);
+    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));
+  }
+
   @Test
   public void testWakeUpSingleTask() {
     String flowReference = "test-flow-ref";