其中Master 节点负责管理作业级状态Worker 节点负责执行 Task Group。当用户发起 Cancel 操作时Master 需要向对应的 Worker 发送取消请求。任务结束或取消完成后Worker 再将最终状态回传给 Master。简化后的流程如下用户发起 Cancel ↓ Master 向 Worker 发送取消请求 ↓ Worker 取消或完成任务 ↓ Worker 向 Master 汇报最终状态 ↓ Master 更新 Job 状态乍看之下这个流程并不复杂。但在分布式系统中每一个环节都可能受到节点故障、集群成员变化以及 Master 恢复过程等因素的影响。而这些因素都可能引发竞态条件Race Condition。问题现象问题的表现很直接Job 一直停留在 CANCELING 状态用户已经执行了取消操作但任务始终无法进CANCELED等最终状态。最开始我把排查重点放在Master → Worker这条取消请求链路上。第一个怀疑对象取消请求链路在 Master 端SeaTunnel 会向 Worker 发送CancelTaskOperation。这里有一个关键逻辑在发送取消请求之前系统会先检查当前任务所在的执行节点是否仍然存在于集群成员列表中。核心逻辑如下while (!taskFuture.isDone() nodeEngine .getClusterService() .getMember(executionAddress getCurrentExecutionAddress()) ! null) { try { nodeEngine .getOperationService() .createInvocationBuilder( Constant.SEATUNNEL_SERVICE_NAME, new CancelTaskOperation(taskGroupLocation), executionAddress) .invoke() .get(); return; } catch (Exception e) { Thread.sleep(2000); } }这一点立刻引起了我的注意。如果由于心跳异常等原因Worker 节点暂时从集群视图中消失那么循环可能会直接退出甚至连 Cancel 请求都没有发送出去。因此我最初的猜测是Master 认为取消流程已经处理完毕但 Worker 实际上从未收到 Cancel 请求。这个方向确实值得关注。不过后来我发现仅凭这一点还不足以解释为什么 Job 会一直停留在 CANCELING 状态。因为即便错过了 Cancel 请求任务最终仍有可能正常结束并将最终状态上报给 Master。于是我开始把视线转向另一个方向当 Worker 向 Master 汇报最终任务状态时究竟会发生什么更关键的链路Worker → Master 的状态通知当任务进入最终状态后Worker 会调用notifyTaskStatusToMaster()将状态上报给 Master。该方法的设计思路是在通知成功之前持续进行重试。while (isRunning !notifyStateSuccess) { InvocationFutureObject invoke nodeEngine .getOperationService() .createInvocationBuilder( SeaTunnelServer.SERVICE_NAME, new NotifyTaskStatusOperation( taskGroupLocation, taskExecutionState), nodeEngine.getMasterAddress()) .invoke(); try { invoke.get(); notifyStateSuccess true; } catch (JobNotFoundException e) { notifyStateSuccess true; } catch (ExecutionException e) { if (e.getCause() instanceof JobNotFoundException) { notifyStateSuccess true; } else { Thread.sleep(sleepTime); } } }乍看之下这套重试机制本身并没有什么问题。但其中对JobNotFoundException的处理却至关重要。如果 Worker 收到JobNotFoundException它会将notifyStateSuccess设为true并停止后续重试。在很多情况下这样的处理是合理的。因为如果 Master 上已经找不到对应的 Job往往意味着该 Job 已经结束并且已经从运行中的 Job 列表中移除。但在 Master Failover 场景下这种假设就可能带来问题。JobNotFoundException 的特殊处理逻辑在 Master 端任务状态更新时会先检查runningJobMasterMappublic void updateTaskExecutionState(TaskExecutionState taskExecutionState) { TaskGroupLocation taskGroupLocation taskExecutionState.getTaskGroupLocation(); JobMaster runningJobMaster runningJobMasterMap.get(taskGroupLocation.getJobId()); if (runningJobMaster null) { throw new JobNotFoundException( String.format(Job %s not running, taskGroupLocation.getJobId())); } runningJobMaster.updateTaskExecutionState(taskExecutionState); }通常情况下runningJobMaster null表示该 Job 已经不再运行。然而还有另一种可能的时间窗口。在 Master Failover 期间新的 Master 可能尚未完全恢复runningJobMasterMap。如果 Worker 恰好在这段时间内上报最终任务状态新 Master 就可能无法找到对应的JobMaster并抛出JobNotFoundException。随后Worker 会将这个异常视为成功处理并停止重试。问题发生的过程可能如下一个 Job 正在执行取消操作。Master 发生切换。Worker 完成任务并上报最终任务状态。新 Master 尚未完全恢复runningJobMasterMap。Master 抛出JobNotFoundException。Worker 将其视为成功处理并停止重试。Master 在恢复完成后始终没有收到最终任务状态。Job 一直停留在CANCELING状态。这正是我发现问题的关键。问题不仅仅在于 Cancel RPC 可能发送失败更重要的是Worker 的最终状态通知有可能在 Master 恢复期间丢失。为什么我选择新增 Force Stop而不是直接修改 Cancel定位到问题后我考虑过几种不同的解决方案调整 JobNotFoundException 的处理逻辑等待 runningJobMasterMap 完全恢复后再处理状态通知将更多运行时状态持久化到分布式存储重构现有的 Cancel 状态机。但这个问题具有偶发性而且高度依赖特定的时序条件。与此同时正常的 Cancel 流程又是整个执行生命周期中非常敏感的一部分。直接修改这部分逻辑可能会引入新的行为变化或者带来额外的性能开销。因此我首先选择了一种更务实的方案。我没有改变现有 Cancel 的语义而是新增了一套独立的 Force Stop 机制。思路其实很简单如果优雅取消Graceful Cancellation无法继续推进那么运维人员就需要一种明确的手段来最终确定 Job 的状态。Cancel 与 Force Stop 的区别让我来解释下 Cancel 与 Force Stop 的区别。CancelCancel 属于一种优雅终止Graceful Shutdown机制。它会向正在运行的任务发送停止请求并依赖正常的任务生命周期以及 Worker 的状态通知链路来完成整个过程。Force StopForce Stop 则是一种面向运维恢复的机制。它不应该依赖远端 Worker 是否仍然能够正常响应。Master 会根据自身的判断直接完成 Job 状态的终结和相关资源的清理。简单来说Cancel 尝试以优雅的方式停止 Job Force Stop 当 Cancel 无法继续推进时直接终结 JobForce Stop 的目的并不是取代 Cancel。它是一条专门用于处理卡死场景的兜底路径。我的收获这次问题让我意识到一个 Job 状态卡住并不一定是负责更新该状态的那段代码出了问题。在这个案例中一开始最可疑的是 Cancel 请求链路。但最终发现更关键的问题其实出在 Worker 到 Master 的状态通知链路上。在正常情况下JobNotFoundException看起来是一个合理的终止条件但在 Master Failover 场景下它也可能意味着新的 Master 尚未完成 Job 的恢复。这两种含义有着本质区别。而这样一个细微的差别恰恰决定了 Worker 应该停止重试还是继续重试。这也再次提醒我在分布式系统中异常处理本身就是状态机的一部分而不仅仅是错误处理逻辑。总结