@Override
    public long retainOwnedTasks(final Set<String> retainForNodes, final Instant statusOlderThan) {
        LOGGER.info(() -> "Retaining owned tasks");
        final List<Condition> conditions = new ArrayList<>();

        // Keep tasks ownership for active nodes.
        final Set<Integer> nodeIdSet = getNodeIdSet(retainForNodes);
        conditions.add(PROCESSOR_TASK.FK_PROCESSOR_NODE_ID.notIn(nodeIdSet));

        // Only change tasks that have not been changed for a certain amount of time.
        conditions.add(PROCESSOR_TASK.STATUS_TIME_MS.lt(statusOlderThan.toEpochMilli()));

        // Only alter tasks that are marked as queued, assigned or processing,
        // i.e. ignore complete and failed tasks.
        conditions.add(ACTIVE_TASKS_STATUS_CONDITION);

        // Release tasks.
        final long count = releaseTasks(conditions);

        LOGGER.info(() -> "Set " + count + " tasks back to CREATED that were " +
                          "QUEUED, ASSIGNED, PROCESSING");

        return count;
    }
Retain task ownership

@param retainForNodes  A set of nodes to retain task ownership for.
@param statusOlderThan Change task ownership for tasks that have a status older than this.
@return The number of tasks released.
stroom-processor/stroom-processor-impl-db/src/main/java/stroom/processor/impl/db/ProcessorTaskDaoImpl.java