@Override
  public void put(@NotNull K key,
                  @Nullable V value) throws IOException {
    int keyHash = keyEquality.getHashCode(key);
    int adjustedHash = HashUtils.adjustHash(keyHash);
    //Abstraction break: synchronize on keyHashToIdMap because we know keyHashToIdMap uses this-monitor to synchronize
    //    itself
    synchronized (keyHashToIdMap) {
      Ref<Boolean> valueIsSameRef = new Ref<>(Boolean.FALSE);
      int foundRecordId = keyHashToIdMap.lookup(
        adjustedHash,
        candidateRecordId -> {
          int logRecordId = convertStoredIdToLogId(candidateRecordId);
          if (valueEquality != null) {
            Entry<K, V> entryWithSameKey = readEntryIfKeyMatch(logRecordId, key);
            if (entryWithSameKey == null) {
              return false; // [record.key != key] => hash collision => look further
            }

            boolean valueIsSame = valuesEqualNullSafe(value, entryWithSameKey.value());
            valueIsSameRef.set(valueIsSame);
          }
          else {
            //without valueEquality we don't use .value at all -> don't need to read it then
            K candidateKey = readKey(logRecordId);
            if (candidateKey == null
                || !keyEquality.isEqual(key, candidateKey)) {
              return false; // [record.key != key] => hash collision => look further
            }

            valueIsSameRef.set(Boolean.FALSE);
          }

          return true;
        }
      );

      boolean keyRecordExists = (foundRecordId != DurableIntToMultiIntMap.NO_VALUE);
      boolean valueIsSame = valueIsSameRef.get();

      //Check is value differ from current value -- we don't need to append the log with the same (key,value)
      // again and again
      if (keyRecordExists && valueIsSame) {
        return; // [current value == new value] => nothing to do
      }

      long newLogRecordId = updateOrWriteEntry(foundRecordId, key, value);
      if (newLogRecordId == foundRecordId) {
        if (value == null) {//remove deleted mapping
          keyHashToIdMap.remove(adjustedHash, foundRecordId);
        }
        return;//new record written on the top of the old one => no need to update the map
      }

      int storedRecordId = convertLogIdToStoredId(newLogRecordId);

      if (keyRecordExists) {
        // (key) record exist, but with different value => replace recordId:
        if (value != null) {
          keyHashToIdMap.replace(adjustedHash, foundRecordId, storedRecordId);
        }
        else {//remove deleted mapping
          keyHashToIdMap.remove(adjustedHash, foundRecordId);
        }
      }
      else {
        // (key) record don't exist yet => put it:
        if (value != null) {
          keyHashToIdMap.put(adjustedHash, storedRecordId);
        }
      }
    }
  }
@Override
  public void put(@NotNull K key,
                  @Nullable V value)