Conductor EVENT 系统任务实战指南Sink 语法、消息构成与 asyncComplete 完成语义【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor本篇以 Conductor事件驱动的 Agent 工作流引擎的EVENT系统任务为核心系统讲解其任务参数、sink目标寻址与conductor命名空间展开规则、发布消息的字段构成以及asyncComplete决定的同步/异步完成行为。读完本文你可以独立编写并发布可靠的EVENT任务定义理解它与事件处理器event handler的衔接方式并能在源码层面解释任务状态机、消息标识与失败重试行为。EVENT 任务定位通用的“发布器”而非 Kafka 专属EVENT任务通过已注册的 event-queue provider 发布一条 JSON 消息是 Conductor 工作流对外广播的通用发布任务。官方参考文档为 event-task.md。选择任务时官方 how-to Publish events 给出的判断标准是使用场景适用任务提供的能力需要向已启用的 event-queue provider 发送与具体中间件无关的消息EVENT统一的 sink 模型、自动附加工作流元数据、稳定的消息标识、与 event handler 的兼容性消息契约需要 Kafka 专属的 key、header、序列化器或生产者控制KAFKA_PUBLISH直接发布到 Kafka topic支持kafka_request专属配置需要强调的一条工程约定不要仅仅因为目标恰好是 Kafka 就使用KAFKA_PUBLISH。除非确实需要 Kafka 特有的生产者控制key、header、序列化器、超时等否则一律优先EVENT这样消息契约与中间件解耦未来更换 broker 只需修改sink。任务参数EVENT任务的完整参数如下继承自参考文档并结合仓库实现补充参数是否必填行为sink是格式为provider:provider-specific destination表达式如${...}在运行时解析inputParameters否用户自定义的消息 payload 字段asyncComplete否默认false为true时发布成功后任务仍保持IN_PROGRESSOSS 中的 provider 标识在 Conductor OSS 中已注册的 provider 标识包括conductor、kafka、sqs、nats、jsm、nats_stream、amqp_queue、amqp_exchange前提是服务器上启用了相应模块本仓库中可对应kafka/、nats/、nats-streaming/、amqp/等子模块。冒号之后的目标语法由各 provider 自己拥有——它可以是 Kafka topic、SQS queue URL、NATS subject或 AMQP 的 queue/exchange 名称等。调度阶段的参数解析源码佐证从源码结构看sink与asyncComplete在任务被映射schedule时就会与inputParameters一起进入输入数据并完成表达式解析EventTaskMapper 先执行workflowTask.getInputParameters().put(sink, workflowTask.getSink()); workflowTask.getInputParameters().put(asyncComplete, workflowTask.isAsyncComplete());再调用ParametersUtils.getTaskInputV2(...)统一解析所有输入含sink中的${...}表达式创建的任务初始状态为SCHEDULED。这解释了文档中“expressions resolve at runtime”的含义sink本身也支持引用工作流输入、上游任务输出等动态表达式。Sink 寻址与 conductor 命名空间展开非conductorprovider 的 sink 原样作为队列名使用queueName sinkValue这一点由 EventSpec 中 “event task with sqs as sink” 用例sqs:testEvent→ 队列名仍为sqs:testEvent验证。conductorprovider 则会把短 sink 按工作流做命名空间展开实现逻辑位于 Event.computeQueueName定义中的 sink展开后的队列名以工作流order_workflow为例conductorconductor:order_workflow:taskReferenceNameconductor:order-statusconductor:order_workflow:order-status规则概括为conductor展开为conductor:workflowName:taskReferenceNameconductor:suffix展开为conductor:workflowName:suffix。事件处理器event handler必须监听展开后的名称否则消息无人消费。测试用例 “event task with conductor as sink” 与 “event task with conductor: as sink” 分别断言了上述两种展开结果见 EventSpec.groovy 第 85-140 行。源码同时暴露了一个边界行为如果 sink 以conductor开头但既不是conductor本身、也不是conductor:前缀例如conductorinvalidsinkcomputeQueueName会抛出IllegalStateException(Invalid / Unsupported sink specified: ...)任务在start阶段即被标记为FAILED并记录reasonForIncompletion——对应测试 “event task with incorrect conductor sink value”。发布消息的构成与输出EVENT任务以解析后的输入参数为起点附加父工作流的元数据。start阶段构建的字段如下实现见 Event.start字段值workflowInstanceId父工作流执行 IDworkflowType父工作流名称workflowVersion父工作流版本correlationId父工作流的 correlation IDtaskToDomain父工作流的 task-domain 映射这些元数据字段由task.addOutput(payload)写入任务输出EventSpec.verifyOutputData对这些键值逐一做了断言EventSpec.groovy 第 333-340 行。event_produced 与消息标识任务输出中还会包含event_produced字段其值为展开后的 sink队列名由start中的task.addOutput(EVENT_PRODUCED, computeQueueName(workflow, task))写入。参考文档与 publish-events 均声明发布到 broker 的消息体是“任务输出去掉event_produced”的形式。需要注意从当前仓库源码结构看getPopulatedMessage 是对整个task.getOutputData()做 JSON 序列化因此消费端若收到event_produced字段直接忽略即可——它只是描述“消息发往何处”的元信息不影响业务字段。另一条关键契约是消息标识getPopulatedMessage构造的Message以任务 ID 同时作为消息 ID 与 receipt IDnew Message(task.getTaskId(), payloadJson, task.getTaskId())。也就是说Event 任务把自己的 task ID 作为 broker 消息身份消费方可以把这个稳定值用作去重键重试重发不会凭空产生新的随机 key。EventSpec.verifyMessage对id与receipt均等于 task ID 做了断言第 342-347 行。完成行为与状态机asyncComplete 的两种语义EVENT的执行状态流转由asyncComplete参数决定源码见 Event.executequeue.publish(List.of(message)); if (!isAsyncComplete(task)) { task.setStatus(TaskModel.Status.COMPLETED); return true; }asyncComplete: false默认发布成功即COMPLETED任务生命周期结束。asyncComplete: true发布成功但任务保持IN_PROGRESS并返回“无需立即更新”的信号后续必须由外部任务更新Task API 的 update 调用或事件处理器中的complete_task/fail_task动作来终结它。对应测试 “event task with conductor as sink and async complete” 断言了execute返回false且状态仍为IN_PROGRESSEventSpec.groovy 同目录第 173-199 行。isAsyncComplete的解析逻辑位于 WorkflowSystemTask即从任务输入中读取该参数。失败路径与取消行为从源码与测试可以确认以下失败语义编写工作流与 event handler 时应纳入设计触发条件任务结果依据sink 表达式非法如conductorinvalidsinkstart阶段即FAILED原因含 “Invalid / Unsupported sink specified”EventSpec.groovysink 解析不到任何已注册队列如未启用 provider 的rabbitmq:abc_123FAILED同上第 217-240 行broker 发布抛出异常瞬态或非瞬态FAILEDreasonForIncompletion记录异常信息同上第 242-287 行payload 无法序列化为 JSONFAILED原因前缀 “Error serializing JSON payload”同上第 289-309 行任务被取消cancel不修改任务状态而是向目标队列对以 task ID 为标识的消息执行ackEvent.cancel实现 “event cancel calls ack on the queue” 用例第 66-83 行发布失败后任务进入FAILED若任务定义TaskDef配置了重试策略执行器会按常规任务重试机制重新调度——发布这一步是普通执行步骤可像其他任务一样被记录与重试。完整配置示例参考文档给出的标准示例发布订单状态事件如下{ name: publish_order_status, taskReferenceName: publish_order_status, type: EVENT, sink: conductor:order-status, inputParameters: { orderId: ${workflow.input.orderId}, status: READY }, asyncComplete: false }解读要点sink为conductor:order-status若所在工作流名为order_workflow实际发布到conductor:order_workflow:order-status事件处理器需监听该展开名orderId通过表达式${workflow.input.orderId}从工作流输入解析演示了inputParameters的动态取值能力asyncComplete: false表示发布成功即完成无需事件处理器回调终结任务。端到端验证可参考集成测试 EventTaskSpec.groovy它覆盖了真实队列上的 EVENT 任务发布与完成流程。生产环境建议结合参考文档与 publish-events 的生产指导投递语义将 broker 投递视为 at-least-once消费端动作与任何副作用必须幂等可观测性监控event_queue_depth与event_queue_messages_*计数器再检查下游工作流或任务结果定位积压标识稳定性保留 broker 消息 ID 并用 Event 任务 ID 做去重重试时不要发明新的随机 key。延伸阅读KAFKA_PUBLISH 任务参考需要 Kafka 专属控制时的替代方案Publish events how-to首次使用的实操走查Event-Driven Orchestrationprovider 矩阵、路由、webhook、signal 与投递可观测性的完整视图核心实现与测试Event.java、EventTaskMapper.java、EventSpec.groovy。【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
