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 644c9e6d..4b19fe49 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 @@ -12,8 +12,11 @@ */ 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; @@ -21,6 +24,7 @@ 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 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, 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 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( 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 dc7c07a6..f0cf965a 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 @@ -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, + // 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 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> 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 flowInput = new HashMap<>(); + flowInput.put(Constants.WORKFLOW_SUMMARY_FIELD, summary); + + flowEngineController.startFlow( + groupId, flowReference, new StartFlowRequest(flowId, null, flowInput)); + + ArgumentCaptor> 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";