code: synchronized void loadState(long lastAppliedIndex, InputStream snapshotInputStream) throws IOException {
          try {
              // read values from the snapshot into a temporary map
              Map<String, String> snapshotEntries = Maps.newHashMap();
  
              ObjectMapper mapper = new ObjectMapper();
              ObjectReader reader = mapper.reader(KeyValue.class);
              MappingIterator<KeyValue> it = reader.readValues(snapshotInputStream);
              while (it.hasNext()) {
                  KeyValue keyValue = it.next();
                  snapshotEntries.put(keyValue.getKey(), keyValue.getValue()); // values are read into this temporary map!
              }
  
              // update the index
              // NOTE: AFAIU, we should never get a snapshot which contains fewer entries than we've applied
              updateLastAppliedIndex(lastAppliedIndex);
  
              // replace the old entries map with the new one generated from the snapshot
              entries = snapshotEntries;
          } finally {
              Closeables.close(snapshotInputStream, true);
          }
      }
docs: Flushes all {@code key=>value} pairs and replaces them with
  the values read from the given {@code snapshotInputStream}.
  This implementation expects the {@code InputStream} to contain
  a sequence of JSON {@link KeyValue} instances.
  
  @param lastAppliedIndex log index >= 0 of the last change contained in this snapshot
  @param snapshotInputStream {@code InputStream} from which all {@code key=> value} pairs should be read
  @throws IOException if valid {@code key=>value} pairs cannot be read from {@code snapshotInputStream}.
  Since no internal state is modified until the {@code snapshotInputStream} is completely read and
  parsed, {@code LocalStore} can still be used safely after an {@code IOException} is thrown
  @throws RuntimeException if the {@code key=>value} pairs in {@code snapshotInputStream} cannot be converted to JSON