public ByteBuffer getOrCreateKey(final Txn<ByteBuffer> writeTxn,
                                     final StagingValue refDataValue,
                                     final PooledByteBuffer valueStoreKeyPooledBuffer,
                                     final EntryConsumer onExistingEntryAction,
                                     final EntryConsumer onNewEntryAction) {

        Preconditions.checkArgument(!writeTxn.isReadOnly(), "A write transaction is required");

        LOGGER.trace("getOrCreate called for refDataValue: {}", refDataValue);

        final ByteBuffer valueBuffer = refDataValue.getValueBuffer();

        LAMBDA_LOGGER.trace(() ->
                LogUtil.message("valueBuffer: {}", ByteBufferUtils.byteBufferInfo(valueBuffer)));

        // Use atomics so they can be mutated and then used in lambdas
        final AtomicBoolean isValueInDb = new AtomicBoolean(false);
        final AtomicInteger valuesCount = new AtomicInteger(0);
        short firstUnusedKeyId = -1;
        short firstUsedKeyId = -1;

        // We have to allocate a new ByteBuffer here as we may/may not return it
        final ByteBuffer startKey = buildStartKeyBuffer(refDataValue, valueStoreKeyPooledBuffer);
        ByteBuffer lastKeyBufferClone = null;

        try (Cursor<ByteBuffer> cursor = getLmdbDbi().openCursor(writeTxn)) {
            // get this key or one greater than it
            boolean isFound = cursor.get(startKey, GetOp.MDB_SET_RANGE);

            short lastKeyId = -1;
            while (isFound) {
                if (ValueStoreKeySerde.compareValueHashCode(startKey, cursor.key()) != 0) {
                    // cursor key has a different hashcode to ours so we can stop looping
                    break;
                }
                valuesCount.incrementAndGet();

                final ByteBuffer valueFromDbBuf = cursor.val();
                final ByteBuffer keyFromDbBuf = cursor.key();
                short thisKeyId = ValueStoreKeySerde.extractId(keyFromDbBuf);

                // Because we have removal of entries we can end up with sparse id sequences
                // therefore capture the first used and unused IDs so we can use it if we need to put a
                // new key/value.
                if (firstUsedKeyId == -1) {
                    // Capture the first id we find for this hash
                    firstUsedKeyId = thisKeyId;
                }
                if (firstUnusedKeyId == -1 && firstUsedKeyId > ValueStoreKey.MIN_UNIQUE_ID) {
                    // There is a gap before the first used key so use the lowest id
                    // e.g. 2,3,7, so use 0
                    firstUnusedKeyId = ValueStoreKey.MIN_UNIQUE_ID;
                } else if (firstUnusedKeyId == -1 && lastKeyId != -1) {
                    if (thisKeyId <= lastKeyId) {
                        throw new RuntimeException(LogUtil.message(
                                "thisKeyId [{}] should be greater than lastId [{}]", thisKeyId, lastKeyId));
                    }
                    if ((thisKeyId - lastKeyId) > 1) {
                        // There is a gap between this id and the last so use one after the last
                        // e.g. 0,1,2,3,7, so use 4
                        firstUnusedKeyId = (short) (lastKeyId + 1);
                    }
                }
                lastKeyId = thisKeyId;

                LAMBDA_LOGGER.trace(() -> LogUtil.message("Our value {}, db value {}",
                        LmdbUtils.byteBufferToHex(valueBuffer),
                        LmdbUtils.byteBufferToHex(valueFromDbBuf)));

                // we cannot use keyVal outside of the cursor loop so have to copy the content
                if (lastKeyBufferClone == null) {
                    // make a new buffer from the cursor key content
                    lastKeyBufferClone = valueStoreKeyPooledBuffer.getByteBuffer();
                }
                lastKeyBufferClone.clear();

                // copy the cursor key content out of the LMDB managed buffer and into our passed in one
                ByteBufferUtils.copy(keyFromDbBuf, lastKeyBufferClone);

                // see if the found value is identical to the value passed in
                if (valueBuffer.equals(valueFromDbBuf)) {
                    isValueInDb.set(true);
                    LAMBDA_LOGGER.trace(() ->
                            "Found our value so incrementing its ref count and breaking out");

                    // perform any entry found actions
                    onExistingEntryAction.accept(writeTxn, keyFromDbBuf, valueFromDbBuf);

                    break;
                } else {
                    LAMBDA_LOGGER.trace(() -> "Values are not equal, keep looking");
                }
                // advance cursor
                isFound = cursor.next();
            }
        }

        LAMBDA_LOGGER.trace(() -> LogUtil.message("isValueInMap: {}, valuesCount {}",
                isValueInDb.get(),
                valuesCount.get()));

        ByteBuffer valueStoreKeyBuffer;

        if (isValueInDb.get()) {
            // value is already in the map, so use its key
            LOGGER.trace("Found value");
            valueStoreKeyBuffer = lastKeyBufferClone;
        } else {
            // value is not in the map so we need to add it
            final ByteBuffer keyBuffer;
            if (lastKeyBufferClone == null) {
                LOGGER.trace("no existing entry for this valueHashCode so use first uniqueId");
                // the start key is the first key for that valueHashcode so use that
                keyBuffer = startKey;
            } else {
                // One or more entries share our valueHashCode (but with different values)
                // so give it a new ID value
                if (firstUnusedKeyId != -1) {
                    ValueStoreKeySerde.updateId(lastKeyBufferClone, firstUnusedKeyId);
                } else {
                    // No gaps in the ids so just use one more than last one
                    ValueStoreKeySerde.incrementId(lastKeyBufferClone);
                }
                keyBuffer = lastKeyBufferClone;
//                LOGGER.trace("Incrementing key, valueStoreKey {}", valueStoreKey);
            }
            valueStoreKeyBuffer = keyBuffer;
            final PutOutcome putOutcome = put(writeTxn, keyBuffer, valueBuffer, false);

            // perform any new entry created actions
            onNewEntryAction.accept(writeTxn, keyBuffer, valueBuffer);

            if (!putOutcome.isSuccess()) {
                throw new RuntimeException(LogUtil.message("Put failed for key: {}, value {}",
                        ByteBufferUtils.byteBufferInfo(keyBuffer),
                        ByteBufferUtils.byteBufferInfo(valueBuffer)));
            }
        }
        return valueStoreKeyBuffer;
    }
Either gets the {@link ValueStoreKey} corresponding to the passed refDataValue
from the database if we already hold it or creates the entry in the database
and returns the generated key. To determine if we already hold it, the value
will be hashed and that hash will be looked up and any matches tested for equality
using the serialised bytes.
<p>
onExistingValueAction Action to perform when the value is found to already exist

@param valueStoreKeyPooledBuffer A pooled buffer to use for the return value
@return A clone of the {@link ByteBuffer} containing the database key.
stroom-pipeline/src/main/java/stroom/pipeline/refdata/store/offheapstore/databases/ValueStoreDb.java