void migrate() {
        LOGGER.debug(() -> LogUtil.message("Migrating stream: {} into  destinationStore: {}",
                refStreamId, destinationStore.getName()));
        updateTaskContextInfoSupplier("Start");
        final DurationTimer timer = DurationTimer.start();
        try {
            final Predicate<RefStreamDefinition> refStreamDefinitionPredicate = refStreamDef ->
                    refStreamDef.getStreamId() == refStreamId;

            // Initially we scan the whole range but as we find stuff we move the start key
            final AtomicReference<KeyRange<RefStreamDefinition>> keyRangeRef = new AtomicReference<>(KeyRange.all());
            boolean wasMatchFound;
            do {
                // Allow for task termination
                if (Thread.currentThread().isInterrupted() || taskContext.isTerminated()) {
                    // As we are outside of a txn the interruption is ok and everything will be in
                    // valid state for when purge is run again. Thus we don't n
                    throw new TaskTerminatedException();
                }

                // RefStreamDefinition has variable width serialisation, so we have to scan to find what we are after
                final Optional<Entry<RefStreamDefinition, RefDataProcessingInfo>> optEntry =
                        sourceStore.getLmdbEnvironment().getWithReadTxn(readTxn ->
                                processingInfoDb.findFirstMatchingKey(
                                        readTxn,
                                        keyRangeRef.get(),
                                        refStreamDefinitionPredicate));

                if (optEntry.isPresent()) {
                    wasMatchFound = true;
                    final RefStreamDefinition refStreamDefinition = optEntry.get().getKey();
                    LOGGER.debug("refStreamDefinition: {}", refStreamDefinition);

                    // Make the next iteration start just after this entry
                    keyRangeRef.set(KeyRange.greaterThan(refStreamDefinition));

                    migrateRefStreamDef(refStreamDefinition, destinationStore);
                } else {
                    LOGGER.debug("No matching ref stream found");
                    wasMatchFound = false;
                }
            } while (wasMatchFound);

            LOGGER.info("Completed migration of ref stream {} into store {} in {}",
                    refStreamId, destinationStore.getName(), timer);

        } catch (TaskTerminatedException e) {
            // Expected behaviour so just rethrow, stopping it being picked up by the other
            // catch block
            LOGGER.debug("Migration terminated, refStreamId: " + refStreamId, e);
            LOGGER.warn("Migration terminated, refStreamId: " + refStreamId);
            throw e;
        } catch (Exception e) {
            LOGGER.error(() -> "Migration of refStreamId " + refStreamId + " failed due to " + e.getMessage());
            throw e;
        }
    }
Migrate the data for refStreamId into destinationStore
stroom-pipeline/src/main/java/stroom/pipeline/refdata/store/offheapstore/StoreMigrator.java