@NotNull
    private Optional<RefStreamDefinition> findNextRefStreamDef(
            final PooledByteBufferPair procInfoPooledBufferPair,
            final AtomicReference<ByteBuffer> currRefStreamDefBufRef,
            final Predicate<ByteBuffer> accessTimePredicate,
            final Txn<ByteBuffer> readTxn) {

        // ensure the buffers are cleared as we are using them in a loop
        procInfoPooledBufferPair.clear();

        final Optional<PooledByteBufferPair> optProcInfoBufferPair = processingInfoDb.getNextEntryAsBytes(
                readTxn,
                currRefStreamDefBufRef.get(),
                accessTimePredicate,
                procInfoPooledBufferPair);

        return optProcInfoBufferPair.map(procInfoBufferPair -> {
            RefStreamDefinition refStreamDefinition = processingInfoDb.deserializeKey(
                    procInfoBufferPair.getKeyBuffer());

            // update the current key buffer so we can search from here next time
            currRefStreamDefBufRef.set(procInfoBufferPair.getKeyBuffer());
            return refStreamDefinition;
        });
    }
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