private Map<String, Object> startWorkflow(
            Action action, Object payload, String event, String messageId) {
        StartWorkflow params = action.getStart_workflow();
        Map<String, Object> output = new HashMap<>();
        try {
            Map<String, Object> inputParams = params.getInput();
            Map<String, Object> workflowInput = parametersUtils.replace(inputParams, payload);

            Map<String, Object> paramsMap = new HashMap<>();
            Optional.ofNullable(params.getCorrelationId())
                    .ifPresent(value -> paramsMap.put("correlationId", value));
            Map<String, Object> replaced = parametersUtils.replace(paramsMap, payload);

            workflowInput.put("conductor.event.messageId", messageId);
            workflowInput.put("conductor.event.name", event);

            StartWorkflowInput startWorkflowInput = new StartWorkflowInput();
            startWorkflowInput.setName(params.getName());
            startWorkflowInput.setVersion(params.getVersion());
            startWorkflowInput.setCorrelationId(
                    Optional.ofNullable(replaced.get("correlationId"))
                            .map(Object::toString)
                            .orElse(params.getCorrelationId()));
            startWorkflowInput.setWorkflowInput(workflowInput);
            startWorkflowInput.setEvent(event);
            startWorkflowInput.setTaskToDomain(params.getTaskToDomain());

            String workflowId = startWorkflowOperation.execute(startWorkflowInput);

            output.put("workflowId", workflowId);
            LOGGER.debug(
                    "Started workflow: {}/{}/{} for event: {} for message:{}",
                    params.getName(),
                    params.getVersion(),
                    workflowId,
                    event,
                    messageId);

        } catch (RuntimeException e) {
            Monitors.recordEventActionError(action.getAction().name(), params.getName(), event);
            LOGGER.error(
                    "Error starting workflow: {}, version: {}, for event: {} for message: {}",
                    params.getName(),
                    params.getVersion(),
                    event,
                    messageId,
                    e);
            output.put("error", e.getMessage());
            throw e;
        }
        return output;
    }
Action Processor subscribes to the Event Actions queue and processes the actions (e.g. start
workflow etc)
core/src/main/java/com/netflix/conductor/core/events/SimpleActionProcessor.java