TaskPersisterJpa.java 文件源码

@Override
public void heartbeat(String nodeId, Set<String> runningTasksId, long failoverTimeout) {
    List<TaskStatusInfo> tsi = listTasks(null, runningTaskStatusFilter, 0, 0);
    if (tsi != null && !tsi.isEmpty()) {
        long ct = System.currentTimeMillis();
        long ft = ct - failoverTimeout;
        for (TaskStatusInfo task : tsi) {
            try {
                em.lock(task, LockModeType.PESSIMISTIC_WRITE);
                if (task.getTaskStatus() == TaskStatus.RUNNING) {
                    if (runningTasksId != null && runningTasksId.contains(task.getId())) {
                        task.setHeartbeat(ct);
                    } else {
                        if (task.getHeartbeat() < ft) {
                            changeTaskStatus(task.getId(), TaskStatus.FAILOVER, "Failover necessity detected by node '" + nodeId
                                    + "' at " + ct + " due heartbeat timestamp " + task.getHeartbeat());
                        }
                    }
                }
            } catch (LockTimeoutException e) {
                log.fine("Lock exception for task id=" + task.getId());
            }
        }
    }
}