private RefStreamPurgeCounts purgeRefStreamIfEligible(
            final Predicate<ByteBuffer> processingInfoBufferPredicate,
            final ByteBuffer refStreamDefinitionBuf,
            final RefStreamDefinition refStreamDefinition) {
        final TaskContext taskContext = taskContextFactory.current();

        LOGGER.debug("Attempting to purge ref stream {} if eligible", refStreamDefinition);
        final AtomicReference<RefStreamPurgeCounts> refStreamPurgeCountsRef = new AtomicReference<>();

        taskContext.info(() -> "Acquiring lock for reference stream " +
                               refStreamDefinition.getStreamId() + ":" + refStreamDefinition.getPartIndex());
        updatePurgeTaskContextInfoSupplier(
                taskContext, refStreamDefinition, "Acquiring lock for reference stream", null);

        // now acquire a lock for this ref stream def, so we don't conflict with any load operations
        doWithRefStreamDefinitionLock(refStreamDefStripedReentrantLock, refStreamDefinition, () -> {

            updatePurgeTaskContextInfoSupplier(
                    taskContext,
                    refStreamDefinition,
                    "Acquired lock for reference stream",
                    null);

            try {
                try (final BatchingWriteTxn batchingWriteTxn = lmdbEnvironment.openBatchingWriteTxn(
                        referenceDataConfigProvider.get().getMaxPurgeDeletesBeforeCommit())) {

                    // We now hold an open write txn so re-fetch the processing info in case something has
                    // changed between our first read and now
                    final Optional<ByteBuffer> optRefDataProcInfoBuf = processingInfoDb.getAsBytes(
                            batchingWriteTxn.getTxn(), refStreamDefinitionBuf);

                    if (optRefDataProcInfoBuf.isPresent()) {
                        final ByteBuffer refDataProcessingInfoBuffer = optRefDataProcInfoBuf.get();

                        LOGGER.debug(() -> LogUtil.message("refStreamDefinition: {}",
                                processingInfoDb.deserializeKey(refDataProcessingInfoBuffer)));

                        // Test the predicate to ensure we can purge it
                        if (processingInfoBufferPredicate.test(refDataProcessingInfoBuffer)) {

                            final RefDataProcessingInfo refDataProcessingInfo = processingInfoDb.deserializeValue(
                                    refDataProcessingInfoBuffer);

                            final RefStreamPurgeCounts refStreamPurgeCounts = purgeRefStreamDef(
                                    taskContext,
                                    refStreamDefinition,
                                    refDataProcessingInfo,
                                    refStreamDefinitionBuf,
                                    batchingWriteTxn);

                            refStreamPurgeCountsRef.set(refStreamPurgeCounts);
                        } else {
                            LOGGER.debug(() -> LogUtil.message("Ref stream {} not eligible for purge, info {}",
                                    refStreamDefinition,
                                    processingInfoDb.deserializeValue(refDataProcessingInfoBuffer)));
                        }
                    } else {
                        LOGGER.info("Ref data processing info does not exist, another thread may have already " +
                                    "purged it {}",
                                refStreamDefinition);
                    }

                    // Force final commit
                    batchingWriteTxn.commit();
                }
            } catch (Exception e) {
                try {
                    LOGGER.error(LogUtil.message("Error purging ref stream {}", refStreamDefinition, e));
                    // We are still under the ref stream def lock here
                    lmdbEnvironment.doWithWriteTxn(writeTxn ->
                            processingInfoDb.updateProcessingState(writeTxn,
                                    refStreamDefinitionBuf,
                                    ProcessingState.PURGE_FAILED,
                                    false));
                } catch (Exception e2) {
                    LOGGER.error("Error setting processing state to PURGE_FAILED for {}. {}",
                            refStreamDefinition,
                            e2.getMessage(),
                            e2);
                }
            }
        });
        return refStreamPurgeCountsRef.get() != null
                ? refStreamPurgeCountsRef.get()
                : RefStreamPurgeCounts.zero();
    }
Synchronized to prevent to ensure concurrent purge jobs don't clash

@param now Allows the setting of the current time for testing purposes
stroom-pipeline/src/main/java/stroom/pipeline/refdata/store/offheapstore/RefDataOffHeapStore.java