private synchronized int queueNewTasks(final TaskContext taskContext,
                                           final boolean isEmptyReportRequired) {
        LOGGER.trace("queueNewTasks() - Starting");
        int totalAdded = 0;

        // We need to make sure that only 1 thread at a time is allowed to
        // create tasks. This should always be the case in production but some
        // tests will call this directly while scheduled execution could also be
        // running.
        info(taskContext, () -> "Starting");

        // Update the stream task store.
        final List<ProcessorFilter> filters = prioritisedFilters.get();
        final ProcessorConfig processorConfig = processorConfigProvider.get();
        final QueueProcessTasksState queueProcessTasksState =
                new QueueProcessTasksState(getTaskQueueSize(), processorConfig.getQueueSize());
        final ProgressMonitor progressMonitor = new ProgressMonitor(filters.size());

        final String nodeName = nodeInfo.getThisNodeName();
        if (nodeName == null) {
            throw new NullPointerException("Node is null");
        }

        // Now fill the stream task store with tasks for each filter.
        // The aim is to create N tasks in total where N is processorConfig.getQueueSize
        // Also need to ensure each filter queue has no more than N in it.
        try {
            if (processorConfig.isFillTaskQueue()) {
                for (final ProcessorFilter filter : filters) {
                    final ProcessorTaskQueue queue = queueMap.computeIfAbsent(
                            filter,
                            k -> new ProcessorTaskQueue());

                    // If we have enough tasks queued then stop trying to add more to the queues.
                    if (!queueProcessTasksState.keepAddingTasks()) {
                        break;

                    } else {
                        totalAdded += queueTasksForFilter(
                                taskContext,
                                nodeName,
                                filter,
                                progressMonitor,
                                queue,
                                queueProcessTasksState);
                    }
                }
            }
        } catch (final RuntimeException e) {
            LOGGER.error(e.getMessage(), e);
        }

        // Release items from the queue that no longer have an enabled filter
        info(taskContext, () -> "Releasing tasks for disabled filters");
        final Set<ProcessorFilter> enabledFilterSet = new HashSet<>(filters);
        for (final Entry<ProcessorFilter, ProcessorTaskQueue> entry : queueMap.entrySet()) {
            final ProcessorFilter filter = entry.getKey();
            final ProcessorTaskQueue queue = entry.getValue();
            if (!enabledFilterSet.contains(filter)) {
                final DurationTimer durationTimer = DurationTimer.start();
                final long count = releaseQueuedFilterTasks(filter);
                final FilterProgressMonitor filterProgressMonitor = progressMonitor.logFilter(filter, queue.size());
                filterProgressMonitor.logPhase(Phase.RELEASE_TASKS_FOR_DISABLED_FILTERS, durationTimer, count);
                filterProgressMonitor.complete();
            }
        }

        info(taskContext, () -> "Finished");

        // No point spamming the logs when this method is called MAX_ASSIGNMENT_ATTEMPTS in succession
        if (queueProcessTasksState.getTotalQueuedCount() > 0
            || isEmptyReportRequired
            || LOGGER.isDebugEnabled()) {
            progressMonitor.report("ADD TASKS TO QUEUE", queueProcessTasksState);
        }

        LOGGER.trace("queueNewTasks() - Finished");
        return totalAdded;
    }
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