private void transfer() {
        SimpleMetrics.measure("Transfer", () -> {
            transferState.setThread(Thread.currentThread());
            try {
                env.write(writeTxn -> {
                    CurrentDbState currentDbState = getCurrentDbState();
                    long lastCommitMs = System.currentTimeMillis();
                    long uncommittedCount = 0;

                    try {
                        while (!transferState.isTerminated()) {
                            LOGGER.trace(() -> "Transferring");
                            SearchProgressLog.increment(queryKey, SearchPhase.LMDB_DATA_STORE_QUEUE_POLL);
                            final LmdbQueueItem queueItem = queue.poll(1, TimeUnit.SECONDS);

                            if (queueItem != null) {
                                if (queueItem instanceof final LmdbKV lmdbKV) {
                                    currentDbState = lmdbKV.getCurrentDbState();
                                    insert(writeTxn, db, lmdbKV);
                                    uncommittedCount++;
                                } else if (queueItem instanceof
                                        final CurrentDbStateLmdbQueueItem currentDbStateLmdbQueueItem) {
                                    currentDbState = currentDbStateLmdbQueueItem.getCurrentDbState()
                                            .mergeExisting(currentDbState);
                                } else if (queueItem instanceof final Sync sync) {
                                    commit(writeTxn, currentDbState);
                                    sync.sync();
                                } else if (queueItem instanceof final DeleteCommand deleteCommand) {
                                    delete(writeTxn, deleteCommand);
                                }
                            }

                            if (producePayloads && payloadCreator.isEmpty()) {
                                // Commit
                                LOGGER.debug(() -> "Committing for new payload");

                                commit(writeTxn, currentDbState);
                                lastCommitMs = System.currentTimeMillis();
                                uncommittedCount = 0;

                                // Create payload and clear the DB.
                                payloadCreator.addPayload(writeTxn, db, false);

                            } else if (uncommittedCount > 0) {
                                final long count = uncommittedCount;
                                if (count >= maxPutsBeforeCommit ||
                                    lastCommitMs < System.currentTimeMillis() - COMMIT_FREQUENCY_MS) {

                                    // Commit
                                    LOGGER.trace(() -> {
                                        if (count >= maxPutsBeforeCommit) {
                                            return "Committing for max puts " + maxPutsBeforeCommit;
                                        } else {
                                            return "Committing for elapsed time";
                                        }
                                    });
                                    commit(writeTxn, currentDbState);
                                    lastCommitMs = System.currentTimeMillis();
                                    uncommittedCount = 0;
                                }
                            }
                        }
                    } catch (final InterruptedException e) {
                        LOGGER.trace(e::getMessage, e);
                        // Keep interrupting this thread.
                        Thread.currentThread().interrupt();
                    } catch (final CompleteException e) {
                        LOGGER.debug(() -> "Complete");
                        LOGGER.trace(e::getMessage, e);
                    } catch (final RuntimeException e) {
                        LOGGER.error(e::getMessage, e);
                        error(e);
                    }

                    if (!transferState.isTerminated() && uncommittedCount > 0) {
                        LOGGER.debug(() -> "Final commit");
                        commit(writeTxn, currentDbState);
                    }

                    // Create final payloads and ensure they are all delivered before we complete.
                    if (!transferState.isTerminated() && producePayloads) {
                        LOGGER.debug(() -> "Producing final payloads");
                        // Create payload and clear the DB.
                        boolean finalPayload = false;
                        while (!finalPayload) {
                            finalPayload = payloadCreator.addPayload(writeTxn, db, true);
                        }
                        // Make sure we end with an empty payload to indicate completion.
                        // Adding a final empty payload to the queue ensures that a consuming node will have to request
                        // the payload from the queue before we complete.
                        LOGGER.debug(() -> "Final payload");
                        payloadCreator.finalPayload();
                    }
                });
            } catch (final Throwable e) {
                LOGGER.error(e::getMessage, e);
                error(e);
            } finally {
                // Ensure we complete.
                queue.terminate();
                // The LMDB environment will be closed after we complete.
                complete.countDown();
                LOGGER.debug(() -> "Finished transfer while loop");
                transferState.setThread(null);
            }
        });
    }
Add some values to the data store.

@param values The values to add to the store.
stroom-query/stroom-query-common/src/main/java/stroom/query/common/v2/LmdbDataStore.java