private int queueCreatedTasks(final TaskContext taskContext,
                                  final String nodeName,
                                  final ProcessorFilter filter,
                                  final ProcessorTaskQueue queue,
                                  final QueueProcessTasksState queueProcessTasksState,
                                  final FilterProgressMonitor filterProgressMonitor) {
        // Queue tasks for this filter.
        final int initialQueueSize = queue.size();
        queueProcessTasksState.addCurrentlyQueuedTasks(initialQueueSize);

        int totalTasks = 0;
        int totalAddedTasks = 0;
        int tasksToAdd = queueProcessTasksState.getRequiredTaskCount() - initialQueueSize;
        final int batchSize = Math.max(BATCH_SIZE, tasksToAdd);
        long lastTaskId = 0;

        try {
            // Keep adding tasks until we have reached the requested number.
            while (tasksToAdd > 0) {

                // Look for any existing tasks we have created.
                DurationTimer durationTimer = DurationTimer.start();
                final List<ExistingCreatedTask> existingCreatedTasks = processorTaskDao
                        .findExistingCreatedTasks(lastTaskId, filter.getId(), batchSize);
                filterProgressMonitor.logPhase(Phase.QUEUE_CREATED_TASKS_FETCH_TASKS,
                        durationTimer,
                        existingCreatedTasks.size());

                // If we got fewer tasks returned than we asked for then we won't need to ask for more.
                if (existingCreatedTasks.size() < batchSize) {
                    tasksToAdd = 0;
                }

                // If we have some existing tasks then queue them unless they belong to locked meta.
                if (!existingCreatedTasks.isEmpty()) {
                    try {
                        // Increment the total number of unowned tasks.
                        totalTasks += existingCreatedTasks.size();

                        final List<Long> metaIdList = new ArrayList<>(existingCreatedTasks.size());
                        for (final ExistingCreatedTask task : existingCreatedTasks) {
                            metaIdList.add(task.getMetaId());
                            // Ensure we don't see this task again in the next attempt.
                            lastTaskId = Math.max(lastTaskId, task.getTaskId());
                        }

                        // Find all locked meta entries for the selected processor tasks.
                        durationTimer = DurationTimer.start();
                        final Set<Long> lockedMetaIdSet = metaService.findLockedMeta(metaIdList);
                        filterProgressMonitor.logPhase(Phase.QUEUE_CREATED_TASKS_FETCH_META,
                                durationTimer,
                                lockedMetaIdSet.size());

                        // Filter out tasks associated with meta that is still locked.
                        final Set<Long> processorTaskIdSet = existingCreatedTasks
                                .stream()
                                .filter(processorTask -> !lockedMetaIdSet.contains(processorTask.getMetaId()))
                                .map(ExistingCreatedTask::getTaskId)
                                .collect(Collectors.toSet());

                        durationTimer = DurationTimer.start();
                        final List<ProcessorTask> existingTasks = processorTaskDao.queueTasks(
                                processorTaskIdSet,
                                nodeName);
                        filterProgressMonitor.logPhase(Phase.QUEUE_CREATED_TASKS_QUEUE_TASKS,
                                durationTimer,
                                existingTasks.size());

                        queue.addAll(existingTasks);
                        tasksToAdd -= existingTasks.size();
                        totalAddedTasks += existingTasks.size();

                        final int finalTotalAddedTasks = totalAddedTasks;
                        final int finalTotalTasks = totalTasks;
                        info(taskContext, () ->
                                LogUtil.message("Adding {}/{} non owned Tasks",
                                        finalTotalAddedTasks,
                                        finalTotalTasks));

                        if (Thread.currentThread().isInterrupted()) {
                            // Stop trying to add tasks.
                            tasksToAdd = 0;
                        }
                    } catch (final RuntimeException e) {
                        LOGGER.error("doCreateTasks() - Failed to grab non owned tasks", e);
                    }
                }
            }

            if (totalAddedTasks > 0) {
                filterProgressMonitor.add(totalAddedTasks);
                LOGGER.debug("doCreateTasks() - Added {} tasks that are no longer locked", totalAddedTasks);
            }

        } catch (final RuntimeException e) {
            LOGGER.error(e.getMessage(), e);
        }

        if (filter.isProcessingTaskCountBounded()) {
            // If the filter specifies a maximum number of concurrent processing tasks then limit the number we
            // report as being added to the queue otherwise we might stop adding other tasks early.
            queueProcessTasksState
                    .addTotalQueuedTasks(Math.min(filter.getMaxProcessingTasks(), initialQueueSize + totalAddedTasks));
        } else {
            queueProcessTasksState.addTotalQueuedTasks(initialQueueSize + totalAddedTasks);
        }

        return totalAddedTasks;
    }
Return the next task to perform. Called by worker threads. Also assigns
the task to the node asking for the job
stroom-processor/stroom-processor-impl/src/main/java/stroom/processor/impl/ProcessorTaskQueueManagerImpl.java