void doCreateSstFiles(final Snapshot snapshot, final EnumMap<SstColumnFamily, File> sstFileTable,
                          final byte[] startKey, final byte[] endKey, final CompletableFuture<Void> future) {
        final Timer.Context timeCtx = getTimeContext("CREATE_SST_FILE");
        final Lock readLock = this.readWriteLock.readLock();
        readLock.lock();
        try {
            if (!this.shutdownLock.isAvailable()) {
                // KV store has shutdown, we do not release rocksdb's snapshot
                future.completeExceptionally(new StorageException("KV store has shutdown."));
                return;
            }
            try (final ReadOptions readOptions = new ReadOptions();
                    final EnvOptions envOptions = new EnvOptions();
                    final Options options = new Options().setMergeOperator(new StringAppendOperator())) {
                readOptions.setSnapshot(snapshot);
                for (final Map.Entry<SstColumnFamily, File> entry : sstFileTable.entrySet()) {
                    final SstColumnFamily sstColumnFamily = entry.getKey();
                    final File sstFile = entry.getValue();
                    final ColumnFamilyHandle columnFamilyHandle = findColumnFamilyHandle(sstColumnFamily);
                    try (final RocksIterator it = this.db.newIterator(columnFamilyHandle, readOptions);
                            final SstFileWriter sstFileWriter = new SstFileWriter(envOptions, options)) {
                        if (startKey == null) {
                            it.seekToFirst();
                        } else {
                            it.seek(startKey);
                        }
                        sstFileWriter.open(sstFile.getAbsolutePath());
                        long count = 0;
                        for (;;) {
                            if (!it.isValid()) {
                                break;
                            }
                            final byte[] key = it.key();
                            if (endKey != null && BytesUtil.compare(key, endKey) >= 0) {
                                break;
                            }
                            sstFileWriter.put(key, it.value());
                            ++count;
                            it.next();
                        }
                        if (count == 0) {
                            sstFileWriter.close();
                        } else {
                            sstFileWriter.finish();
                        }
                        LOG.info("Finish sst file {} with {} keys.", sstFile, count);
                    } catch (final RocksDBException e) {
                        throw new StorageException("Fail to create sst file at path: " + sstFile, e);
                    }
                }
                future.complete(null);
            } catch (final Throwable t) {
                future.completeExceptionally(t);
            } finally {
                // Nothing to release, rocksDB never own the pointer for a snapshot.
                snapshot.close();
                // The pointer to the snapshot is released by the database instance.
                this.db.releaseSnapshot(snapshot);
            }
        } finally {
            readLock.unlock();
            timeCtx.stop();
        }
    }
Local KV store based on RocksDB

@author dennis
@author jiachun.fjc
jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/storage/RocksRawKVStore.java