third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: Repeated execution after broadcast worker node down #1003

Browse Source
This commit is contained in:
tjq 2024-11-22 21:32:13 +08:00
parent 4e84bc60d7
commit 57627305fa
1 changed files with 4 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
powerjob-worker/src/main/java/tech/powerjob/worker/core/tracker/task/heavy/CommonTaskTracker.java
View File

@ -296,8 +296,10 @@ public class CommonTaskTracker extends HeavyTaskTracker {
// 6.2 定期检查 -> 重新执行被派发到宕机ProcessorTracker上的任务 // 6.2 定期检查 -> 重新执行被派发到宕机ProcessorTracker上的任务
List<String> disconnectedPTs = ptStatusHolder.getAllDisconnectedProcessorTrackers(); List<String> disconnectedPTs = ptStatusHolder.getAllDisconnectedProcessorTrackers();
if (!disconnectedPTs.isEmpty()) { if (!disconnectedPTs.isEmpty()) {
log.warn("[TaskTracker-{}] some ProcessorTracker disconnected from TaskTracker,their address is {}.", instanceId, disconnectedPTs); // 广播任务节点丢失后若直接移除 IP 重试，后续会派发到其他节点，导致重复执行，因此此处不能重试 https://github.com/PowerJob/PowerJob/issues/1003
if (taskPersistenceService.updateLostTasks(instanceId, disconnectedPTs, true)) { boolean needRetry = !ExecuteType.BROADCAST.equals(executeType);
log.warn("[TaskTracker-{}] some ProcessorTracker disconnected from TaskTracker,their address is {}, needRetry: {}.", instanceId, disconnectedPTs, needRetry);
if (taskPersistenceService.updateLostTasks(instanceId, disconnectedPTs, needRetry)) {
ptStatusHolder.remove(disconnectedPTs); ptStatusHolder.remove(disconnectedPTs);
log.warn("[TaskTracker-{}] removed these ProcessorTracker from StatusHolder: {}", instanceId, disconnectedPTs); log.warn("[TaskTracker-{}] removed these ProcessorTracker from StatusHolder: {}", instanceId, disconnectedPTs);
} }