private ProcessorTaskList assignTasks(final String nodeName,
                                          final int count,
                                          final TaskContext taskContext) {
        LOGGER.debug("assignTasks() called for node {}, count {}", nodeName, count);
        if (!processorConfigProvider.get().isAssignTasks()) {
            LOGGER.debug("assignTasks is disabled");
            return new ProcessorTaskList(nodeName, Collections.emptyList());
        }
        if (count <= 0) {
            LOGGER.warn("assignTasks called for {} tasks", count);
            return new ProcessorTaskList(nodeName, Collections.emptyList());
        }
        if (!securityContext.isProcessingUser()) {
            throw new PermissionException(securityContext.getUserRef(),
                    "Only the processing user is allowed to assign tasks");
        }

        final List<ProcessorTask> assignedStreamTasks = new ArrayList<>();
        final AtomicInteger attempt = new AtomicInteger();
        while (attempt.getAndIncrement() < MAX_ASSIGNMENT_ATTEMPTS) {
            try {
                info(taskContext, () -> "Attempting task assignment " +
                                        "(attempt=" +
                                        attempt.get() +
                                        ")");
                // Get local reference to list in case it is swapped out.
                final List<ProcessorFilter> filters = prioritisedFilters.get();
                // Try and get a bunch of tasks from the queue to assign to the requesting node.
                info(taskContext, () ->
                        "Attempting task assignment for " +
                        filters.size() +
                        " filters " +
                        "(attempt=" +
                        attempt.get() +
                        ")");
                final AtomicInteger filterCount = new AtomicInteger();
                for (final ProcessorFilter filter : filters) {
                    info(taskContext, () ->
                            "Attempting task assignment for " +
                            filterCount.incrementAndGet() +
                            "/" +
                            filters.size() +
                            " filters " +
                            " (filter=" +
                            filter +
                            ", attempt=" +
                            attempt.get() +
                            ")");

                    if (assignedStreamTasks.size() >= count) {
                        break;
                    }

                    // Get the queue for this filter.
                    final ProcessorTaskQueue queue = queueMap.get(filter);
                    if (queue != null) {
                        int filterTasksAssigned = 0;

                        // Maximum number of tasks to assign for this filter. If the filter task limit is
                        // unbounded, assign as many tasks up to the specified `count`. Otherwise, only assign
                        // tasks up to the filter's configured limit.
                        int maxFilterTasks = count - assignedStreamTasks.size();
                        if (filter.isProcessingTaskCountBounded()) {
                            final int maxFilterTasksToCreate = filter.getMaxProcessingTasks() -
                                                               processorTaskDao.countTasksForFilter(filter.getId(),
                                                                       TaskStatus.PROCESSING);
                            maxFilterTasks = Math.min(
                                    maxFilterTasks,
                                    Math.max(0, maxFilterTasksToCreate));
                        }

                        if (maxFilterTasks > 0) {
                            // Add as many tasks as we can for this filter.
                            ProcessorTask streamTask = queue.poll();
                            while (streamTask != null) {
                                assignedStreamTasks.add(streamTask);
                                filterTasksAssigned++;

                                if (assignedStreamTasks.size() < count && filterTasksAssigned < maxFilterTasks) {
                                    streamTask = queue.poll();
                                } else {
                                    streamTask = null;
                                }
                            }
                        }
                    }
                }
            } catch (final RuntimeException e) {
                LOGGER.error(e.getMessage(), e);
            }

            // Output some trace logging, so we can see where tasks go.
            taskStatusTraceLog.assignTasks(ProcessorTaskQueueManagerImpl.class, assignedStreamTasks, nodeName);

            if (LOGGER.isDebugEnabled()) {
                info(taskContext, () -> "Assigning " +
                                        assignedStreamTasks.size() +
                                        " tasks (" +
                                        count +
                                        " requested) to node " +
                                        nodeName);
                LOGGER.debug("Assigning " + assignedStreamTasks.size()
                             + " tasks (" + count + " requested) to node " + nodeName);
            }

            // If we don't get any tasks then force synchronous queueing of tasks.
            if (allowTaskQueueFill) {
                if (assignedStreamTasks.isEmpty()) {
                    info(taskContext, () -> "Assigned " +
                                            assignedStreamTasks.size() +
                                            " tasks, filling queue synchronously");
                    // Only want to see an empty progress report on the first attempt
                    fillTaskQueueSync(attempt.get() <= 1); // Has already been incremented
                } else {
                    // Kick off a queue fill.
                    info(taskContext, () -> "Assigned " +
                                            assignedStreamTasks.size() +
                                            " tasks, filling queue asynchronously");
                    fillTaskQueueAsync();
                    attempt.set(MAX_ASSIGNMENT_ATTEMPTS);
                }
            } else {
                attempt.set(MAX_ASSIGNMENT_ATTEMPTS);
            }
        }

        return new ProcessorTaskList(nodeName, assignedStreamTasks);
    }
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