private RefStreamPurgeCounts purgeRefStreamData(final TaskContext taskContext,
                                                    final BatchingWriteTxn batchingWriteTxn,
                                                    final RefStreamDefinition refStreamDefinition) {

        LOGGER.debug("purgeRefStreamData({}), store: '{}'", refStreamDefinition, storeName);
        final PurgeCounter purgeCounter = new PurgeCounter();
        RefStreamPurgeCounts summaryInfo = RefStreamPurgeCounts.zero();
        Optional<UID> optMapUid;
        updatePurgeTaskContextInfoSupplier(taskContext, refStreamDefinition, "Finding maps", purgeCounter);

        try (PooledByteBuffer pooledUidBuffer = byteBufferPool.getPooledByteBuffer(UID.UID_ARRAY_LENGTH)) {
            do {
                // open a ranged cursor on the map forward table to scan all map defs for that stream def
                // for each map def get the map uid
                optMapUid = mapDefinitionUIDStore.getNextMapDefinition(
                        batchingWriteTxn.getTxn(), refStreamDefinition, pooledUidBuffer::getByteBuffer);

                if (optMapUid.isPresent()) {
                    final UID mapUid = optMapUid.get();
                    final MapDefinition mapDefinition = mapDefinitionUIDStore.get(
                                    batchingWriteTxn.getTxn(), mapUid)
                            .orElseThrow(() ->
                                    new RuntimeException(LogUtil.message(
                                            "We should be a mapDefinition if we have a UID, uid: {}",
                                            mapUid)));

                    LOGGER.debug("Found mapUid {} for refStreamDefinition {}", mapUid, refStreamDefinition);

                    // Partially reset the counter, so we can log the counts for this map def
                    purgeCounter.resetTransientCounts();
                    purgeMapData(taskContext, refStreamDefinition, batchingWriteTxn, mapUid, purgeCounter);

                    summaryInfo = summaryInfo.increment(
                            1,
                            purgeCounter.transientValuesPurged,
                            purgeCounter.transientValuesDeReferenced);

                    LOGGER.info("  Purged map {}, {} values deleted, {} values de-referenced, store: '{}'",
                            mapDefinition.getMapName(),
                            purgeCounter.transientValuesPurged,
                            purgeCounter.transientValuesDeReferenced,
                            storeName);
                } else {
                    LOGGER.debug("No more map definitions to purge for refStreamDefinition {}", refStreamDefinition);
                }
            } while (optMapUid.isPresent());
        }

//        LAMBDA_LOGGER.info("Purged data for {} map(s) for {}", cnt, refStreamDefinition);
        return summaryInfo;
    }
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