private void populateMaps(final Pipeline pipeline,
                              final Meta meta,
                              final Source source,
                              final String feedName,
                              final RefStreamDefinition refStreamDefinition) {
        // Set the source meta.
        metaHolder.setMeta(meta);

        // Get the store specific to this load
        final RefDataStore refDataStore = refDataStoreFactory.getOffHeapStore(refStreamDefinition);

        taskContext.info(() -> LogUtil.message(
                "Loading reference data stream {}:{} - Acquiring stream lock",
                refStreamDefinition.getStreamId(),
                refStreamDefinition.getPartNumber()));

        refDataStore.doWithLoaderUnlessComplete(refStreamDefinition, meta.getEffectiveMs(), refDataLoader -> {
            // we are now blocking any other thread loading the same refStreamDefinition
            // and know that this stream has not already been loaded.
            taskContext.info(() -> LogUtil.message(
                    "Loading reference data stream {}:{} - Starting stream processing",
                    refStreamDefinition.getStreamId(),
                    refStreamDefinition.getPartNumber()));

            try {
                // set this loader in the holder so it is available to the pipeline filters
                refDataLoaderHolder.setRefDataLoader(refDataLoader);

                // Start processing.
                pipeline.startProcessing();

                // Get the appropriate encoding for the stream type.
                final String encoding = feedProperties.getEncoding(
                        feedName, meta.getTypeName(), null);
                LOGGER.debug("Using encoding '{}' for feed {}", encoding, feedName);

                final StreamLocationFactory streamLocationFactory = new StreamLocationFactory();
                locationFactory.setLocationFactory(streamLocationFactory);

                // Loop over the stream boundaries and process each sequentially.
                // Typically, ref data will only have a single partIndex so if there are
                // multiple then overrideExisting may be needed.
                final long count = source.count();
                for (long index = 0; index < count && !taskContext.isTerminated(); index++) {
                    metaHolder.setPartIndex(index);
                    streamLocationFactory.setPartIndex(index);

                    try (final InputStreamProvider inputStreamProvider = source.get(index)) {
                        // Get the stream providers.
                        metaHolder.setInputStreamProvider(inputStreamProvider);

                        // Get the stream.
                        final InputStream inputStream = inputStreamProvider.get();

                        // Process the boundary.
                        //process the pipeline, ref data will be loaded via the ReferenceDataFilter
                        try {
                            LOGGER.debug("Starting processing on pipeline: {}", pipeline);
                            pipeline.process(inputStream, encoding);
                        } finally {
                            // This calls endProcessing on the ReferenceDataFilter, which will initiate
                            // the transfer from staging to ref store
                            LOGGER.debug("Ending processing on pipeline: {}", pipeline);
                            pipeline.endProcessing();
                        }
                    }
                }

                if (taskContext.isTerminated()) {
                    final String msg = LogUtil.message("Load terminated while loading stream {} from feed '{}'",
                            refStreamDefinition.getStreamId(),
                            feedName);
                    logAndCompleteProcessing(refDataLoader, Severity.WARNING, ProcessingState.TERMINATED, msg);
                } else {
                    final String msg = LogUtil.message("Successfully loaded stream {} from feed: '{}'",
                            refStreamDefinition.getStreamId(),
                            feedName);
                    logAndCompleteProcessing(refDataLoader, Severity.INFO, ProcessingState.COMPLETE, msg);
                }
            } catch (UncheckedIOException | IOException e) {
                // Missing ref strm file or unable to read it
                final String msg = "Error reading reference stream "
                                   + refStreamDefinition.getStreamId()
                                   + ":"
                                   + refStreamDefinition.getPartNumber()
                                   + " - "
                                   + e.getMessage();
                logAndCompleteProcessing(refDataLoader, Severity.FATAL_ERROR, ProcessingState.FAILED, msg, e);
            } catch (Exception e) {
                handleException(refDataLoader, e);
            }
        });

        // clear the reference to the loader now we have finished with it
        refDataLoaderHolder.setRefDataLoader(null);
    }
Loads reference data that meets the supplied criteria into the current
reference data key, value maps.
stroom-pipeline/src/main/java/stroom/pipeline/refdata/ReferenceDataLoadTaskHandler.java