@Override
    public List<ProcessorTask> queueTasks(final Set<Long> idSet,
                                          final String nodeName) {
        final long now = System.currentTimeMillis();
        final Integer nodeId;
        if (nodeName != null) {
            nodeId = processorNodeCache.getOrCreate(nodeName);
        } else {
            nodeId = null;
        }

        // Do everything within a single transaction.
        final Result<Record> result = JooqUtil.transactionResult(
                processorDbConnProvider, context -> {
                    // Change created tasks to queued.
                    final Condition updateCondition = PROCESSOR_TASK.ID.in(idSet)
                            .and(PROCESSOR_TASK.STATUS.eq(TaskStatus.CREATED.getPrimitiveValue()));
                    final int count = changeStatus(
                            context,
                            nodeId,
                            TaskStatus.QUEUED,
                            now,
                            updateCondition);

                    // Select back the updated records.
                    final Condition selectCondition = PROCESSOR_TASK.ID.in(idSet)
                            .and(PROCESSOR_TASK.STATUS_TIME_MS.eq(now))
                            .and(PROCESSOR_TASK.FK_PROCESSOR_NODE_ID.eq(nodeId))
                            .and(PROCESSOR_TASK.STATUS.eq(TaskStatus.QUEUED.getPrimitiveValue()));
                    final Result<Record> r = select(context, selectCondition);

                    if (r.size() != count) {
                        throw new RuntimeException(
                                "Unexpected number of stream tasks selected back after update.");
                    }

                    return r;
                });

        return convert(result);
    }
Change the node ownership of the tasks in the id set and select them back to include in the queue.

@param idSet    The ids of the tasks to take ownership of.
@param nodeName This node name.
@return A list of tasks to queue.
stroom-processor/stroom-processor-impl-db/src/main/java/stroom/processor/impl/db/ProcessorTaskDaoImpl.java