@Override
    public ProcessorTask changeTaskStatus(final ProcessorTask processorTask,
                                          final String nodeName,
                                          final TaskStatus status,
                                          final Long startTime,
                                          final Long endTime) {
        LOGGER.debug(() -> LogUtil.message(
                "changeTaskStatus({}) - Changing task status on node {}, {}",
                status, nodeName, processorTask));
        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 ProcessorTaskRecord result = JooqUtil.transactionResultWithOptimisticLocking(
                processorDbConnProvider, context -> {
                    ProcessorTaskRecord record = context.newRecord(PROCESSOR_TASK);

                    try {
                        try {
                            record.from(processorTask);
                            record.setFkProcessorNodeId(nodeId);
                            record.setStatus(status.getPrimitiveValue());
                            record.setStatusTimeMs(now);
                            record.setStartTimeMs(startTime);
                            record.setEndTimeMs(endTime);
                            record.update();

                        } catch (final RuntimeException e) {
                            // Try this operation a few times.
                            boolean success = false;
                            RuntimeException lastError = null;

                            // Try and do this up to 100 times.
                            for (int tries = 0; tries < 100 && !success; tries++) {
                                success = true;

                                try {
                                    if (LOGGER.isDebugEnabled()) {
                                        LOGGER.warn(() -> LogUtil.message(
                                                "changeTaskStatus({}) - {} - Task has changed, attempting reload {}",
                                                status, e.getMessage(), processorTask), e);
                                    } else {
                                        LOGGER.warn(() -> LogUtil.message(
                                                "changeTaskStatus({}) - Task has changed, attempting reload {}",
                                                status, processorTask));
                                    }

                                    final Optional<ProcessorTaskRecord> optTaskRec = context
                                            .selectFrom(PROCESSOR_TASK)
                                            .where(PROCESSOR_TASK.ID.eq(record.getId()))
                                            .fetchOptional();
                                    LOGGER.debug("Actual DB record {}", optTaskRec);

                                    if (optTaskRec.isEmpty()) {
                                        LOGGER.warn(() -> LogUtil.message(
                                                "changeTaskStatus({}) - Task does not exist, " +
                                                "task may have been physically deleted {}",
                                                processorTask));
                                        record = null;
                                    } else if (TaskStatus.DELETED.getPrimitiveValue() == optTaskRec.get().getStatus()) {
                                        LOGGER.warn(() -> LogUtil.message(
                                                "changeTaskStatus({}) - Task has been logically deleted {}",
                                                status,
                                                processorTask));
                                        record = null;
                                    } else {
                                        LOGGER.warn(() -> LogUtil.message(
                                                "changeTaskStatus({}) - Re-loaded stream task {}",
                                                status,
                                                optTaskRec.get()));
                                        record = optTaskRec.get();
                                        record.setFkProcessorNodeId(nodeId);
                                        record.setStatus(status.getPrimitiveValue());
                                        record.setStatusTimeMs(now);
                                        record.setStartTimeMs(startTime);
                                        record.setEndTimeMs(endTime);
                                        record.update();
                                    }
                                } catch (final RuntimeException e2) {
                                    success = false;
                                    lastError = e2;
                                    // Wait before trying this operation again.
                                    Thread.sleep(1000);
                                }
                            }

                            if (!success) {
                                LOGGER.error("Error changing task status to {} for task '{}': {}",
                                        status, processorTask, lastError.getMessage(), lastError);
                            }
                        }
                    } catch (final InterruptedException e) {
                        LOGGER.error(e::getMessage, e);

                        // Continue to interrupt this thread.
                        Thread.currentThread().interrupt();
                    }

                    return record;
                });

        return convert(result,
                nodeName,
                processorTask.getFeedName(),
                processorTask.getProcessorFilter());
    }
Count the current number of tasks for a filter matching the specified status.

@param filterId The filter to count tasks for.
@param status   Task status.
@return The number of tasks matching the specified status.
stroom-processor/stroom-processor-impl-db/src/main/java/stroom/processor/impl/db/ProcessorTaskDaoImpl.java