JNI get_helper code sharing / multiGet() use efficient batch C++ support (#12344)

Summary: Implement RAII-based helpers for JNIGet() and multiGet() Replace JNI C++ helpers `rocksdb_get_helper, rocksdb_get_helper_direct`, `multi_get_helper`, `multi_get_helper_direct`, `multi_get_helper_release_keys`, `txn_get_helper`, and `txn_multi_get_helper`. The model is to entirely do away with a single helper, instead a number of utility methods allow each separate JNI `Get()` and `MultiGet()` method to organise their parameters efficiently, then call the underlying C++ `db->Get()`, `db->MultiGet()`, `txn->Get()`, or `txn->MultiGet()` method itself, and use further utilities to retrieve results. Roughly speaking: * get keys into C++ form * Call C++ Get() * get results and status into Java form We achieve a useful performance gain as part of this work; by using the updated C++ multiGet we immediately pick up its performance gains (batch improvements to multiGet C++ were previously implemented, but not until now used by Java/JNI). multiGetBB already uses the batched C++ multiGet(), and all other benchmarks show consistent improvement after the changes: ## Before: ``` Benchmark (columnFamilyTestType) (keyCount) (keySize) (multiGetSize) (valueSize) Mode Cnt Score Error Units MultiGetNewBenchmarks.multiGetBB200 no_column_family 10000 1024 100 256 thrpt 25 5315.459 ± 20.465 ops/s MultiGetNewBenchmarks.multiGetBB200 no_column_family 10000 1024 100 1024 thrpt 25 5673.115 ± 78.299 ops/s MultiGetNewBenchmarks.multiGetBB200 no_column_family 10000 1024 100 4096 thrpt 25 2616.860 ± 46.994 ops/s MultiGetNewBenchmarks.multiGetBB200 no_column_family 10000 1024 100 16384 thrpt 25 1700.058 ± 24.034 ops/s MultiGetNewBenchmarks.multiGetBB200 no_column_family 10000 1024 100 65536 thrpt 25 791.171 ± 13.955 ops/s MultiGetNewBenchmarks.multiGetList10 no_column_family 10000 1024 100 256 thrpt 25 6129.929 ± 94.200 ops/s MultiGetNewBenchmarks.multiGetList10 no_column_family 10000 1024 100 1024 thrpt 25 7012.405 ± 97.886 ops/s MultiGetNewBenchmarks.multiGetList10 no_column_family 10000 1024 100 4096 thrpt 25 2799.014 ± 39.352 ops/s MultiGetNewBenchmarks.multiGetList10 no_column_family 10000 1024 100 16384 thrpt 25 1417.205 ± 22.272 ops/s MultiGetNewBenchmarks.multiGetList10 no_column_family 10000 1024 100 65536 thrpt 25 655.594 ± 13.050 ops/s MultiGetNewBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 256 thrpt 25 6147.247 ± 82.711 ops/s MultiGetNewBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 1024 thrpt 25 7004.213 ± 79.251 ops/s MultiGetNewBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 4096 thrpt 25 2715.154 ± 110.017 ops/s MultiGetNewBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 16384 thrpt 25 1408.070 ± 31.714 ops/s MultiGetNewBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 65536 thrpt 25 623.829 ± 57.374 ops/s MultiGetNewBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 256 thrpt 25 6119.243 ± 116.313 ops/s MultiGetNewBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 1024 thrpt 25 6931.873 ± 128.094 ops/s MultiGetNewBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 4096 thrpt 25 2678.253 ± 39.113 ops/s MultiGetNewBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 16384 thrpt 25 1337.384 ± 19.500 ops/s MultiGetNewBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 65536 thrpt 25 625.596 ± 14.525 ops/s ``` ## After: ``` Benchmark (columnFamilyTestType) (keyCount) (keySize) (multiGetSize) (valueSize) Mode Cnt Score Error Units MultiGetBenchmarks.multiGetBB200 no_column_family 10000 1024 100 256 thrpt 25 5191.074 ± 78.250 ops/s MultiGetBenchmarks.multiGetBB200 no_column_family 10000 1024 100 1024 thrpt 25 5378.692 ± 260.682 ops/s MultiGetBenchmarks.multiGetBB200 no_column_family 10000 1024 100 4096 thrpt 25 2590.183 ± 34.844 ops/s MultiGetBenchmarks.multiGetBB200 no_column_family 10000 1024 100 16384 thrpt 25 1634.793 ± 34.022 ops/s MultiGetBenchmarks.multiGetBB200 no_column_family 10000 1024 100 65536 thrpt 25 786.455 ± 8.462 ops/s MultiGetBenchmarks.multiGetBB200 1_column_family 10000 1024 100 256 thrpt 25 5285.055 ± 11.676 ops/s MultiGetBenchmarks.multiGetBB200 1_column_family 10000 1024 100 1024 thrpt 25 5586.758 ± 213.008 ops/s MultiGetBenchmarks.multiGetBB200 1_column_family 10000 1024 100 4096 thrpt 25 2527.172 ± 17.106 ops/s MultiGetBenchmarks.multiGetBB200 1_column_family 10000 1024 100 16384 thrpt 25 1819.547 ± 12.958 ops/s MultiGetBenchmarks.multiGetBB200 1_column_family 10000 1024 100 65536 thrpt 25 803.861 ± 9.963 ops/s MultiGetBenchmarks.multiGetBB200 20_column_families 10000 1024 100 256 thrpt 25 5253.793 ± 28.020 ops/s MultiGetBenchmarks.multiGetBB200 20_column_families 10000 1024 100 1024 thrpt 25 5705.591 ± 20.556 ops/s MultiGetBenchmarks.multiGetBB200 20_column_families 10000 1024 100 4096 thrpt 25 2523.377 ± 15.415 ops/s MultiGetBenchmarks.multiGetBB200 20_column_families 10000 1024 100 16384 thrpt 25 1815.344 ± 11.309 ops/s MultiGetBenchmarks.multiGetBB200 20_column_families 10000 1024 100 65536 thrpt 25 820.792 ± 3.192 ops/s MultiGetBenchmarks.multiGetBB200 100_column_families 10000 1024 100 256 thrpt 25 5262.184 ± 20.477 ops/s MultiGetBenchmarks.multiGetBB200 100_column_families 10000 1024 100 1024 thrpt 25 5706.959 ± 23.123 ops/s MultiGetBenchmarks.multiGetBB200 100_column_families 10000 1024 100 4096 thrpt 25 2520.362 ± 9.170 ops/s MultiGetBenchmarks.multiGetBB200 100_column_families 10000 1024 100 16384 thrpt 25 1789.185 ± 14.239 ops/s MultiGetBenchmarks.multiGetBB200 100_column_families 10000 1024 100 65536 thrpt 25 818.401 ± 12.132 ops/s MultiGetBenchmarks.multiGetList10 no_column_family 10000 1024 100 256 thrpt 25 6978.310 ± 14.084 ops/s MultiGetBenchmarks.multiGetList10 no_column_family 10000 1024 100 1024 thrpt 25 7664.242 ± 22.304 ops/s MultiGetBenchmarks.multiGetList10 no_column_family 10000 1024 100 4096 thrpt 25 2881.778 ± 81.054 ops/s MultiGetBenchmarks.multiGetList10 no_column_family 10000 1024 100 16384 thrpt 25 1599.826 ± 7.190 ops/s MultiGetBenchmarks.multiGetList10 no_column_family 10000 1024 100 65536 thrpt 25 737.520 ± 6.809 ops/s MultiGetBenchmarks.multiGetList10 1_column_family 10000 1024 100 256 thrpt 25 6974.376 ± 10.716 ops/s MultiGetBenchmarks.multiGetList10 1_column_family 10000 1024 100 1024 thrpt 25 7637.440 ± 45.877 ops/s MultiGetBenchmarks.multiGetList10 1_column_family 10000 1024 100 4096 thrpt 25 2820.472 ± 42.231 ops/s MultiGetBenchmarks.multiGetList10 1_column_family 10000 1024 100 16384 thrpt 25 1716.663 ± 8.527 ops/s MultiGetBenchmarks.multiGetList10 1_column_family 10000 1024 100 65536 thrpt 25 755.848 ± 7.514 ops/s MultiGetBenchmarks.multiGetList10 20_column_families 10000 1024 100 256 thrpt 25 6943.651 ± 20.040 ops/s MultiGetBenchmarks.multiGetList10 20_column_families 10000 1024 100 1024 thrpt 25 7679.415 ± 9.114 ops/s MultiGetBenchmarks.multiGetList10 20_column_families 10000 1024 100 4096 thrpt 25 2844.564 ± 13.388 ops/s MultiGetBenchmarks.multiGetList10 20_column_families 10000 1024 100 16384 thrpt 25 1729.545 ± 5.983 ops/s MultiGetBenchmarks.multiGetList10 20_column_families 10000 1024 100 65536 thrpt 25 783.218 ± 1.530 ops/s MultiGetBenchmarks.multiGetList10 100_column_families 10000 1024 100 256 thrpt 25 6944.276 ± 29.995 ops/s MultiGetBenchmarks.multiGetList10 100_column_families 10000 1024 100 1024 thrpt 25 7670.301 ± 8.986 ops/s MultiGetBenchmarks.multiGetList10 100_column_families 10000 1024 100 4096 thrpt 25 2839.828 ± 12.421 ops/s MultiGetBenchmarks.multiGetList10 100_column_families 10000 1024 100 16384 thrpt 25 1730.005 ± 9.209 ops/s MultiGetBenchmarks.multiGetList10 100_column_families 10000 1024 100 65536 thrpt 25 787.096 ± 1.977 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 256 thrpt 25 6896.944 ± 21.530 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 1024 thrpt 25 7622.407 ± 12.824 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 4096 thrpt 25 2927.538 ± 19.792 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 16384 thrpt 25 1598.041 ± 4.312 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 no_column_family 10000 1024 100 65536 thrpt 25 744.564 ± 9.236 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 1_column_family 10000 1024 100 256 thrpt 25 6853.760 ± 78.041 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 1_column_family 10000 1024 100 1024 thrpt 25 7360.917 ± 355.365 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 1_column_family 10000 1024 100 4096 thrpt 25 2848.774 ± 13.409 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 1_column_family 10000 1024 100 16384 thrpt 25 1727.688 ± 3.329 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 1_column_family 10000 1024 100 65536 thrpt 25 776.088 ± 7.517 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 20_column_families 10000 1024 100 256 thrpt 25 6910.339 ± 14.366 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 20_column_families 10000 1024 100 1024 thrpt 25 7633.660 ± 10.830 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 20_column_families 10000 1024 100 4096 thrpt 25 2787.799 ± 81.775 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 20_column_families 10000 1024 100 16384 thrpt 25 1726.517 ± 6.830 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 20_column_families 10000 1024 100 65536 thrpt 25 787.597 ± 3.362 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 100_column_families 10000 1024 100 256 thrpt 25 6922.445 ± 10.493 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 100_column_families 10000 1024 100 1024 thrpt 25 7604.710 ± 48.043 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 100_column_families 10000 1024 100 4096 thrpt 25 2848.788 ± 15.783 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 100_column_families 10000 1024 100 16384 thrpt 25 1730.837 ± 6.497 ops/s MultiGetBenchmarks.multiGetListExplicitCF20 100_column_families 10000 1024 100 65536 thrpt 25 794.557 ± 1.869 ops/s MultiGetBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 256 thrpt 25 6918.716 ± 15.766 ops/s MultiGetBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 1024 thrpt 25 7626.692 ± 9.394 ops/s MultiGetBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 4096 thrpt 25 2871.382 ± 72.155 ops/s MultiGetBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 16384 thrpt 25 1598.786 ± 4.819 ops/s MultiGetBenchmarks.multiGetListRandomCF30 no_column_family 10000 1024 100 65536 thrpt 25 748.469 ± 7.234 ops/s MultiGetBenchmarks.multiGetListRandomCF30 1_column_family 10000 1024 100 256 thrpt 25 6922.666 ± 17.131 ops/s MultiGetBenchmarks.multiGetListRandomCF30 1_column_family 10000 1024 100 1024 thrpt 25 7623.890 ± 8.805 ops/s MultiGetBenchmarks.multiGetListRandomCF30 1_column_family 10000 1024 100 4096 thrpt 25 2850.698 ± 18.004 ops/s MultiGetBenchmarks.multiGetListRandomCF30 1_column_family 10000 1024 100 16384 thrpt 25 1727.623 ± 4.868 ops/s MultiGetBenchmarks.multiGetListRandomCF30 1_column_family 10000 1024 100 65536 thrpt 25 774.534 ± 10.025 ops/s MultiGetBenchmarks.multiGetListRandomCF30 20_column_families 10000 1024 100 256 thrpt 25 5486.251 ± 13.582 ops/s MultiGetBenchmarks.multiGetListRandomCF30 20_column_families 10000 1024 100 1024 thrpt 25 4920.656 ± 44.557 ops/s MultiGetBenchmarks.multiGetListRandomCF30 20_column_families 10000 1024 100 4096 thrpt 25 3922.913 ± 25.686 ops/s MultiGetBenchmarks.multiGetListRandomCF30 20_column_families 10000 1024 100 16384 thrpt 25 2873.106 ± 4.336 ops/s MultiGetBenchmarks.multiGetListRandomCF30 20_column_families 10000 1024 100 65536 thrpt 25 802.404 ± 8.967 ops/s MultiGetBenchmarks.multiGetListRandomCF30 100_column_families 10000 1024 100 256 thrpt 25 4817.996 ± 18.042 ops/s MultiGetBenchmarks.multiGetListRandomCF30 100_column_families 10000 1024 100 1024 thrpt 25 4243.922 ± 13.929 ops/s MultiGetBenchmarks.multiGetListRandomCF30 100_column_families 10000 1024 100 4096 thrpt 25 3175.998 ± 7.773 ops/s MultiGetBenchmarks.multiGetListRandomCF30 100_column_families 10000 1024 100 16384 thrpt 25 2321.990 ± 12.501 ops/s MultiGetBenchmarks.multiGetListRandomCF30 100_column_families 10000 1024 100 65536 thrpt 25 1753.028 ± 7.130 ops/s ``` Closes https://github.com/facebook/rocksdb/issues/11518 Pull Request resolved: https://github.com/facebook/rocksdb/pull/12344 Reviewed By: cbi42 Differential Revision: D54809714 Pulled By: pdillinger fbshipit-source-id: bee3b949720abac073bce043b59ce976a11e99eb
2024-03-12 12:42:08 -07:00 · 2024-03-12 12:42:08 -07:00 · d9a441113e
parent 36c1b0aded
commit d9a441113e
21 changed files with 1686 additions and 896 deletions
--- a/include/rocksdb/utilities/transaction.h
+++ b/include/rocksdb/utilities/transaction.h
@ -417,6 +417,23 @@ class Transaction {
                              std::string* value, bool exclusive = true,
                              const bool do_validate = true) = 0;
  // An overload of the above method that receives a PinnableSlice
  // For backward compatibility a default implementation is provided
  virtual Status GetForUpdate(const ReadOptions& options, const Slice& key,
                              PinnableSlice* pinnable_val,
                              bool exclusive = true,
                              const bool do_validate = true) {
    if (pinnable_val == nullptr) {
      std::string* null_str = nullptr;
      return GetForUpdate(options, key, null_str, exclusive, do_validate);
    } else {
      auto s = GetForUpdate(options, key, pinnable_val->GetSelf(), exclusive,
                            do_validate);
      pinnable_val->PinSelf();
      return s;
    }
  }
  virtual std::vector<Status> MultiGetForUpdate(
      const ReadOptions& options,
      const std::vector<ColumnFamilyHandle*>& column_family,
--- a/java/CMakeLists.txt
+++ b/java/CMakeLists.txt
@ -46,6 +46,7 @@ set(JNI_NATIVE_SOURCES
        rocksjni/hyper_clock_cache.cc
        rocksjni/ingest_external_file_options.cc
        rocksjni/iterator.cc
        rocksjni/jni_multiget_helpers.cc
        rocksjni/jnicallback.cc
        rocksjni/loggerjnicallback.cc
        rocksjni/lru_cache.cc
--- a/java/jmh/README.md
+++ b/java/jmh/README.md
@ -6,10 +6,10 @@ These are micro-benchmarks for RocksJava functionality, using [JMH (Java Microbe
 **Note**: This uses a specific build of RocksDB that is set in the `<version>` element of the `dependencies` section of the `pom.xml` file. If you are testing local changes you should build and install a SNAPSHOT version of rocksdbjni, and update the `pom.xml` of rocksdbjni-jmh file to test with this.
-For instance, this is how to install the OSX jar you just built for 6.26.0
+For instance, this is how to install the OSX jar you just built for 8.11.0
 ```bash
-$ mvn install:install-file -Dfile=./java/target/rocksdbjni-6.26.0-SNAPSHOT-osx.jar -DgroupId=org.rocksdb -DartifactId=rocksdbjni -Dversion=6.26.0-SNAPSHOT -Dpackaging=jar
+$ mvn install:install-file -Dfile=./java/target/rocksdbjni-8.11.0-SNAPSHOT-osx.jar -DgroupId=org.rocksdb -DartifactId=rocksdbjni -Dversion=8.11.0-SNAPSHOT -Dpackaging=jar
 ```
 ```bash
--- a/java/jmh/pom.xml
+++ b/java/jmh/pom.xml
@ -38,8 +38,8 @@
    </scm>
    <properties>
-        <project.build.source>1.7</project.build.source>
+        <project.build.source>17</project.build.source>
-        <project.build.target>1.7</project.build.target>
+        <project.build.target>17</project.build.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <jmh.version>1.22</jmh.version>
@ -50,7 +50,7 @@
        <dependency>
            <groupId>org.rocksdb</groupId>
            <artifactId>rocksdbjni</artifactId>
-            <version>7.9.0-SNAPSHOT</version>
+            <version>9.0.0</version>
        </dependency>
        <dependency>
--- a/java/jmh/src/main/java/org/rocksdb/jmh/MultiGetBenchmarks.java
+++ b/java/jmh/src/main/java/org/rocksdb/jmh/MultiGetBenchmarks.java
@ -55,6 +55,9 @@ public class MultiGetBenchmarks {
  RocksDB db;
  private final AtomicInteger keyIndex = new AtomicInteger();
  private List<ColumnFamilyHandle> defaultCFHandles = new ArrayList<>();
  private List<ColumnFamilyHandle> randomCFHandles = new ArrayList<>();
  @Setup(Level.Trial)
  public void setup() throws IOException, RocksDBException {
    RocksDB.loadLibrary();
@ -88,6 +91,12 @@ public class MultiGetBenchmarks {
    cfHandles = cfHandlesList.toArray(new ColumnFamilyHandle[0]);
    // store initial data for retrieving via get
    for (int j = 0; j < keyCount; j++) {
      final byte[] paddedValue = Arrays.copyOf(ba("value" + j), valueSize);
      db.put(ba("key" + j), paddedValue);
    }
    // store initial data for retrieving via get - column families
    for (int i = 0; i < cfs; i++) {
      for (int j = 0; j < keyCount; j++) {
        final byte[] paddedValue = Arrays.copyOf(ba("value" + j), valueSize);
@ -95,6 +104,17 @@ public class MultiGetBenchmarks {
      }
    }
    // build a big list of default column families for efficient passing
    final ColumnFamilyHandle defaultCFH = db.getDefaultColumnFamily();
    for (int i = 0; i < keyCount; i++) {
      defaultCFHandles.add(defaultCFH);
    }
    // list of random cfs
    for (int i = 0; i < keyCount; i++) {
      randomCFHandles.add(cfHandlesList.get((int) (Math.random() * cfs)));
    }
    try (final FlushOptions flushOptions = new FlushOptions()
            .setWaitForFlush(true)) {
      db.flush(flushOptions);
@ -163,15 +183,13 @@ public class MultiGetBenchmarks {
  @Setup
  public void allocateSliceBuffers() {
-    keysBuffer = ByteBuffer.allocateDirect(keyCount * valueSize);
+    keysBuffer = ByteBuffer.allocateDirect(keyCount * keySize);
    valuesBuffer = ByteBuffer.allocateDirect(keyCount * valueSize);
    valueBuffersList = new ArrayList<>();
    keyBuffersList = new ArrayList<>();
    for (int i = 0; i < keyCount; i++) {
-      valueBuffersList.add(valuesBuffer.slice());
+      valueBuffersList.add(valuesBuffer.slice(i * valueSize, valueSize));
-      valuesBuffer.position(i * valueSize);
+      keyBuffersList.add(keysBuffer.slice(i * keySize, keySize));
      keyBuffersList.add(keysBuffer.slice());
      keysBuffer.position(i * keySize);
    }
  }
@ -181,7 +199,7 @@ public class MultiGetBenchmarks {
  }
  @Benchmark
-  public List<byte[]> multiGet10() throws RocksDBException {
+  public void multiGetList10() throws RocksDBException {
    final int fromKeyIdx = next(multiGetSize, keyCount);
    if (fromKeyIdx >= 0) {
      final List<byte[]> keys = keys(fromKeyIdx, fromKeyIdx + multiGetSize);
@ -191,6 +209,53 @@ public class MultiGetBenchmarks {
          throw new RuntimeException("Test valueSize assumption wrong");
      }
    }
  }
  @Benchmark
  public void multiGetListExplicitCF20() throws RocksDBException {
    final int fromKeyIdx = next(multiGetSize, keyCount);
    if (fromKeyIdx >= 0) {
      final List<byte[]> keys = keys(fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<ColumnFamilyHandle> columnFamilyHandles =
          defaultCFHandles.subList(fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<byte[]> valueResults = db.multiGetAsList(columnFamilyHandles, keys);
      for (final byte[] result : valueResults) {
        if (result.length != valueSize)
          throw new RuntimeException("Test valueSize assumption wrong");
      }
    }
  }
  @Benchmark
  public void multiGetListRandomCF30() throws RocksDBException {
    final int fromKeyIdx = next(multiGetSize, keyCount);
    if (fromKeyIdx >= 0) {
      final List<byte[]> keys = keys(fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<ColumnFamilyHandle> columnFamilyHandles =
          randomCFHandles.subList(fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<byte[]> valueResults = db.multiGetAsList(columnFamilyHandles, keys);
      for (final byte[] result : valueResults) {
        if (result.length != valueSize)
          throw new RuntimeException("Test valueSize assumption wrong");
      }
    }
  }
  @Benchmark
  public List<byte[]> multiGetBB200() throws RocksDBException {
    final int fromKeyIdx = next(multiGetSize, keyCount);
    if (fromKeyIdx >= 0) {
      final List<ByteBuffer> keys = keys(keyBuffersList, fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<ByteBuffer> values =
          valueBuffersList.subList(fromKeyIdx, fromKeyIdx + multiGetSize);
      final List<ByteBufferGetStatus> statusResults = db.multiGetByteBuffers(keys, values);
      for (final ByteBufferGetStatus result : statusResults) {
        if (result.status.getCode() != Status.Code.Ok)
          throw new RuntimeException("Test status not OK: " + result.status);
        if (result.value.limit() != valueSize)
          throw new RuntimeException("Test valueSize assumption wrong");
      }
    }
    return new ArrayList<>();
  }
--- a/java/rocksjni/jni_multiget_helpers.cc
+++ b/java/rocksjni/jni_multiget_helpers.cc
@ -0,0 +1,290 @@
 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
 //  This source code is licensed under both the GPLv2 (found in the
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 #include "rocksjni/jni_multiget_helpers.h"
 #include "jni_multiget_helpers.h"
 #include "rocksjni/portal.h"
 namespace ROCKSDB_NAMESPACE {
 bool MultiGetJNIKeys::fromByteArrays(JNIEnv* env, jobjectArray jkeys) {
  const jsize num_keys = env->GetArrayLength(jkeys);
  for (jsize i = 0; i < num_keys; i++) {
    jobject jkey = env->GetObjectArrayElement(jkeys, i);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      return false;
    }
    jbyteArray jkey_ba = reinterpret_cast<jbyteArray>(jkey);
    const jsize len_key = env->GetArrayLength(jkey_ba);
    std::unique_ptr<jbyte[]> key = std::make_unique<jbyte[]>(len_key);
    jbyte* raw_key = reinterpret_cast<jbyte*>(key.get());
    key_bufs.push_back(std::move(key));
    env->GetByteArrayRegion(jkey_ba, 0, len_key, raw_key);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      env->DeleteLocalRef(jkey);
      return false;
    }
    slices_.push_back(
        ROCKSDB_NAMESPACE::Slice(reinterpret_cast<char*>(raw_key), len_key));
    env->DeleteLocalRef(jkey);
  }
  return true;
 }
 bool MultiGetJNIKeys::fromByteArrays(JNIEnv* env, jobjectArray jkeys,
                                     jintArray jkey_offs, jintArray jkey_lens) {
  const jsize num_keys = env->GetArrayLength(jkeys);
  std::unique_ptr<jint[]> key_offs = std::make_unique<jint[]>(num_keys);
  env->GetIntArrayRegion(jkey_offs, 0, num_keys, key_offs.get());
  if (env->ExceptionCheck()) {
    return false;  // exception thrown: ArrayIndexOutOfBoundsException
  }
  std::unique_ptr<jint[]> key_lens = std::make_unique<jint[]>(num_keys);
  env->GetIntArrayRegion(jkey_lens, 0, num_keys, key_lens.get());
  if (env->ExceptionCheck()) {
    return false;  // exception thrown: ArrayIndexOutOfBoundsException
  }
  for (jsize i = 0; i < num_keys; i++) {
    jobject jkey = env->GetObjectArrayElement(jkeys, i);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      return false;
    }
    jbyteArray jkey_ba = reinterpret_cast<jbyteArray>(jkey);
    const jint len_key = key_lens[i];
    std::unique_ptr<jbyte[]> key = std::make_unique<jbyte[]>(len_key);
    jbyte* raw_key = reinterpret_cast<jbyte*>(key.get());
    key_bufs.push_back(std::move(key));
    env->GetByteArrayRegion(jkey_ba, key_offs[i], len_key, raw_key);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      env->DeleteLocalRef(jkey);
      return false;
    }
    slices_.push_back(
        ROCKSDB_NAMESPACE::Slice(reinterpret_cast<char*>(raw_key), len_key));
    env->DeleteLocalRef(jkey);
  }
  return true;
 }
 bool MultiGetJNIKeys::fromByteBuffers(JNIEnv* env, jobjectArray jkeys,
                                      jintArray jkey_offs,
                                      jintArray jkey_lens) {
  const jsize num_keys = env->GetArrayLength(jkeys);
  std::unique_ptr<jint[]> key_offs = std::make_unique<jint[]>(num_keys);
  env->GetIntArrayRegion(jkey_offs, 0, num_keys, key_offs.get());
  if (env->ExceptionCheck()) {
    return false;  // exception thrown: ArrayIndexOutOfBoundsException
  }
  std::unique_ptr<jint[]> key_lens = std::make_unique<jint[]>(num_keys);
  env->GetIntArrayRegion(jkey_lens, 0, num_keys, key_lens.get());
  if (env->ExceptionCheck()) {
    return false;  // exception thrown: ArrayIndexOutOfBoundsException
  }
  for (jsize i = 0; i < num_keys; i++) {
    jobject jkey = env->GetObjectArrayElement(jkeys, i);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      return false;
    }
    char* key = reinterpret_cast<char*>(env->GetDirectBufferAddress(jkey));
    ROCKSDB_NAMESPACE::Slice key_slice(key + key_offs[i], key_lens[i]);
    slices_.push_back(key_slice);
    env->DeleteLocalRef(jkey);
  }
  return true;
 }
 ROCKSDB_NAMESPACE::Slice* MultiGetJNIKeys::data() { return slices_.data(); }
 std::vector<ROCKSDB_NAMESPACE::Slice>::size_type MultiGetJNIKeys::size() {
  return slices_.size();
 }
 template <class TValue>
 jobjectArray MultiGetJNIValues::byteArrays(
    JNIEnv* env, std::vector<TValue>& values,
    std::vector<ROCKSDB_NAMESPACE::Status>& s) {
  jobjectArray jresults = ROCKSDB_NAMESPACE::ByteJni::new2dByteArray(
      env, static_cast<jsize>(s.size()));
  if (jresults == nullptr) {
    // exception occurred
    OutOfMemoryErrorJni::ThrowNew(env, "Insufficient Memory for results.");
    return nullptr;
  }
  // add to the jresults
  for (std::vector<ROCKSDB_NAMESPACE::Status>::size_type i = 0; i != s.size();
       i++) {
    if (s[i].ok()) {
      TValue* value = &values[i];
      jbyteArray jentry_value =
          ROCKSDB_NAMESPACE::JniUtil::createJavaByteArrayWithSizeCheck(
              env, value->data(), value->size());
      if (jentry_value == nullptr) {
        // exception set
        return nullptr;
      }
      env->SetObjectArrayElement(jresults, static_cast<jsize>(i), jentry_value);
      if (env->ExceptionCheck()) {
        // exception thrown:
        // ArrayIndexOutOfBoundsException
        env->DeleteLocalRef(jentry_value);
        return nullptr;
      }
      env->DeleteLocalRef(jentry_value);
    } else if (s[i].code() != ROCKSDB_NAMESPACE::Status::Code::kNotFound) {
      // The only way to return an error for a single key is to exception the
      // entire multiGet() Previous behaviour was to return a nullptr value for
      // this case and potentially succesfully return values for other keys; we
      // retain this behaviour. To change it, we need to do the following:
      // ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(env, s[i]);
      // return nullptr;
    }
  }
  return jresults;
 }
 template jobjectArray MultiGetJNIValues::byteArrays<std::string>(
    JNIEnv* env, std::vector<std::string>& values,
    std::vector<ROCKSDB_NAMESPACE::Status>& s);
 template jobjectArray
 MultiGetJNIValues::byteArrays<ROCKSDB_NAMESPACE::PinnableSlice>(
    JNIEnv* env, std::vector<ROCKSDB_NAMESPACE::PinnableSlice>& values,
    std::vector<ROCKSDB_NAMESPACE::Status>& s);
 template <class TValue>
 void MultiGetJNIValues::fillByteBuffersAndStatusObjects(
    JNIEnv* env, std::vector<TValue>& values,
    std::vector<ROCKSDB_NAMESPACE::Status>& s, jobjectArray jvalues,
    jintArray jvalue_sizes, jobjectArray jstatuses) {
  std::vector<jint> value_size;
  for (int i = 0; i < static_cast<jint>(values.size()); i++) {
    auto jstatus = ROCKSDB_NAMESPACE::StatusJni::construct(env, s[i]);
    if (jstatus == nullptr) {
      // exception in context
      return;
    }
    env->SetObjectArrayElement(jstatuses, i, jstatus);
    if (s[i].ok()) {
      jobject jvalue_bytebuf = env->GetObjectArrayElement(jvalues, i);
      if (env->ExceptionCheck()) {
        // ArrayIndexOutOfBoundsException is thrown
        return;
      }
      jlong jvalue_capacity = env->GetDirectBufferCapacity(jvalue_bytebuf);
      if (jvalue_capacity == -1) {
        ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(
            env,
            "Invalid value(s) argument (argument is not a valid direct "
            "ByteBuffer)");
        return;
      }
      void* jvalue_address = env->GetDirectBufferAddress(jvalue_bytebuf);
      if (jvalue_address == nullptr) {
        ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(
            env,
            "Invalid value(s) argument (argument is not a valid direct "
            "ByteBuffer)");
        return;
      }
      // record num returned, push back that number, which may be bigger then
      // the ByteBuffer supplied. then copy as much as fits in the ByteBuffer.
      static const size_t INTEGER_MAX_VALUE =
          ((static_cast<size_t>(1)) << 31) - 1;
      if (values[i].size() > INTEGER_MAX_VALUE) {
        // Indicate that the result size is bigger than can be represented in a
        // java integer by setting the status to incomplete and the size to -1
        env->SetObjectArrayElement(
            jstatuses, i,
            ROCKSDB_NAMESPACE::StatusJni::construct(
                env, Status::Incomplete("result too large to represent")));
        value_size.push_back(-1);
      } else {
        value_size.push_back(static_cast<jint>(values[i].size()));
      }
      auto copy_bytes =
          std::min(static_cast<jlong>(values[i].size()), jvalue_capacity);
      memcpy(jvalue_address, values[i].data(), copy_bytes);
    } else {
      // bad status for this
      value_size.push_back(0);
    }
  }
  env->SetIntArrayRegion(jvalue_sizes, 0, static_cast<jint>(values.size()),
                         value_size.data());
 }
 template void MultiGetJNIValues::fillByteBuffersAndStatusObjects<
    ROCKSDB_NAMESPACE::PinnableSlice>(
    JNIEnv* env, std::vector<ROCKSDB_NAMESPACE::PinnableSlice>& values,
    std::vector<ROCKSDB_NAMESPACE::Status>& s, jobjectArray jvalues,
    jintArray jvalue_sizes, jobjectArray jstatuses);
 std::unique_ptr<std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>>
 ColumnFamilyJNIHelpers::handlesFromJLongArray(
    JNIEnv* env, jlongArray jcolumn_family_handles) {
  if (jcolumn_family_handles == nullptr) return nullptr;
  const jsize num_cols = env->GetArrayLength(jcolumn_family_handles);
  std::unique_ptr<jlong[]> jcf_handles = std::make_unique<jlong[]>(num_cols);
  env->GetLongArrayRegion(jcolumn_family_handles, 0, num_cols,
                          jcf_handles.get());
  if (env->ExceptionCheck())
    // ArrayIndexOutOfBoundsException
    return nullptr;
  auto cf_handles =
      std::make_unique<std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>>();
  for (jsize i = 0; i < num_cols; i++) {
    auto* cf_handle = reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>(
        jcf_handles.get()[i]);
    cf_handles->push_back(cf_handle);
  }
  return cf_handles;
 }
 ROCKSDB_NAMESPACE::ColumnFamilyHandle* ColumnFamilyJNIHelpers::handleFromJLong(
    JNIEnv* env, jlong jcolumn_family_handle) {
  auto cf_handle = reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>(
      jcolumn_family_handle);
  if (cf_handle == nullptr) {
    ROCKSDB_NAMESPACE::RocksDBExceptionJni::ThrowNew(
        env, ROCKSDB_NAMESPACE::Status::InvalidArgument(
                 "Invalid ColumnFamilyHandle."));
    return nullptr;
  }
  return cf_handle;
 };
 };  // namespace ROCKSDB_NAMESPACE
--- a/java/rocksjni/jni_multiget_helpers.h
+++ b/java/rocksjni/jni_multiget_helpers.h
@ -0,0 +1,163 @@
 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
 //  This source code is licensed under both the GPLv2 (found in the
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).
 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 #pragma once
 #include <jni.h>
 #include <functional>
 #include "rocksdb/convenience.h"
 #include "rocksdb/db.h"
 namespace ROCKSDB_NAMESPACE {
 /**
 * @brief Encapsulate keys and key conversions from Java/JNI objects for
 * MultiGet
 *
 */
 class MultiGetJNIKeys {
 private:
  std::vector<ROCKSDB_NAMESPACE::Slice> slices_;
  std::vector<std::unique_ptr<jbyte[]>> key_bufs;
 public:
  /**
   * @brief Construct helper multiget keys object from array of java keys
   *
   * @param env JNI environment
   * @param jkeys Array of `byte[]`, each of which contains a key
   * @param jkey_offs array of offsets into keys, at which each key starts
   * @param jkey_lens array of key lengths
   * @return true if the keys were copied successfully from the parameters
   * @return false if a Java exception was raised (memory problem, or array
   * indexing problem)
   */
  bool fromByteArrays(JNIEnv* env, jobjectArray jkeys, jintArray jkey_offs,
                      jintArray jkey_lens);
  /**
   * @brief Construct helper multiget keys object from array of java keys
   *
   * @param env env JNI environment
   * @param jkeys jkeys Array of byte[], each of which is a key
   * @return true if the keys were copied successfully from the parameters
   * @return false if a Java exception was raised (memory problem, or array
   * indexing problem)
   */
  bool fromByteArrays(JNIEnv* env, jobjectArray jkeys);
  /**
   * @brief Construct helper multiget keys object from array of java ByteBuffers
   *
   * @param env JNI environment
   * @param jkeys Array of `java.nio.ByteBuffer`, each of which contains a key
   * @param jkey_offs array of offsets into buffers, at which each key starts
   * @param jkey_lens array of key lengths
   * @return `true` if the keys were copied successfully from the parameters
   * @return `false` if a Java exception was raised (memory problem, or array
   * indexing problem)
   */
  bool fromByteBuffers(JNIEnv* env, jobjectArray jkeys, jintArray jkey_offs,
                       jintArray jkey_lens);
  /**
   * @brief Used when the keys need to be passed to a RocksDB function which
   * takes keys as an array of slice pointers
   *
   * @return ROCKSDB_NAMESPACE::Slice* an array of slices, the n-th slice
   * contains the n-th key created by `fromByteArrays()` or `fromByteBuffers()`
   */
  ROCKSDB_NAMESPACE::Slice* data();
  /**
   * @brief Used when the keys need to be passed to a RocksDB function which
   * takes keys as a vector of slices
   *
   * @return std::vector<ROCKSDB_NAMESPACE::Slice>& a vector of slices, the n-th
   * slice contains the n-th key created by `fromByteArrays()` or
   * `fromByteBuffers()`
   */
  inline std::vector<ROCKSDB_NAMESPACE::Slice>& slices() { return slices_; }
  /**
   * @brief
   *
   * @return std::vector<ROCKSDB_NAMESPACE::Slice>::size_type the number of keys
   * in this object
   */
  std::vector<ROCKSDB_NAMESPACE::Slice>::size_type size();
 };
 /**
 * @brief Class with static helpers for returning java objects from RocksDB data
 * returned by MultiGet
 *
 */
 class MultiGetJNIValues {
 public:
  /**
   * @brief create an array of `byte[]` containing the result values from
   * `MultiGet`
   *
   * @tparam TValue a `std::string` or a `PinnableSlice` containing the result
   * for a single key
   * @return jobjectArray an array of `byte[]`, one per value in the input
   * vector
   */
  template <class TValue>
  static jobjectArray byteArrays(JNIEnv*, std::vector<TValue>&,
                                 std::vector<ROCKSDB_NAMESPACE::Status>&);
  /**
   * @brief fill a supplied array of `byte[]` with the result values from
   * `MultiGet`
   *
   * @tparam TValue a `std::string` or a `PinnableSlice` containing the result
   * for a single key
   * @param jvalues the array of `byte[]` to instantiate
   * @param jvalue_sizes the offsets at which to place the results in `jvalues`
   * @param jstatuses the status for every individual key/value get
   */
  template <class TValue>
  static void fillByteBuffersAndStatusObjects(
      JNIEnv*, std::vector<TValue>&, std::vector<ROCKSDB_NAMESPACE::Status>&,
      jobjectArray jvalues, jintArray jvalue_sizes, jobjectArray jstatuses);
 };
 /**
 * @brief class with static helper for arrays of column family handles
 *
 */
 class ColumnFamilyJNIHelpers {
 public:
  /**
   * @brief create a native array of cf handles from java handles
   *
   * @param env
   * @param jcolumn_family_handles
   * @return unique ptr to vector of handles on success, reset() unique ptr on
   * failure (and a JNI exception will be generated)
   */
  static std::unique_ptr<std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>>
  handlesFromJLongArray(JNIEnv* env, jlongArray jcolumn_family_handles);
  /**
   * @brief create a column family handle from a raw pointer, or raise an
   * appropriate JNI exception
   *
   * @param env
   * @param jcolumn_family_handle the raw pointer to convert
   * @return ROCKSDB_NAMESPACE::ColumnFamilyHandle* or raises a java exception
   */
  static ROCKSDB_NAMESPACE::ColumnFamilyHandle* handleFromJLong(
      JNIEnv* env, jlong jcolumn_family_handle);
 };
 };  // namespace ROCKSDB_NAMESPACE
--- a/java/rocksjni/kv_helper.h
+++ b/java/rocksjni/kv_helper.h
@ -213,14 +213,11 @@ class JByteArrayPinnableSlice {
   */
  jbyteArray NewByteArray() {
    const jint pinnable_len = static_cast<jint>(pinnable_slice_.size());
-    jbyteArray jbuffer = env_->NewByteArray(static_cast<jsize>(pinnable_len));
+    jbyteArray jbuffer =
        ROCKSDB_NAMESPACE::JniUtil::createJavaByteArrayWithSizeCheck(
            env_, pinnable_slice_.data(), pinnable_len);
    KVException::ThrowOnError(env_);  // OutOfMemoryError
    env_->SetByteArrayRegion(
        jbuffer, 0, pinnable_len,
        reinterpret_cast<const jbyte*>(pinnable_slice_.data()));
    KVException::ThrowOnError(env_);  // ArrayIndexOutOfBoundsException
    return jbuffer;
  }
--- a/java/rocksjni/portal.h
+++ b/java/rocksjni/portal.h
@ -222,6 +222,64 @@ class IllegalArgumentExceptionJni
  }
 };
 // The portal class for java.lang.IllegalArgumentException
 class OutOfMemoryErrorJni : public JavaException<OutOfMemoryErrorJni> {
 public:
  /**
   * Get the Java Class java.lang.OutOfMemoryError
   *
   * @param env A pointer to the Java environment
   *
   * @return The Java Class or nullptr if one of the
   *     ClassFormatError, ClassCircularityError, NoClassDefFoundError,
   *     OutOfMemoryError or ExceptionInInitializerError exceptions is thrown
   */
  static jclass getJClass(JNIEnv* env) {
    return JavaException::getJClass(env, "java/lang/OutOfMemoryError");
  }
  /**
   * Create and throw a Java OutOfMemoryError with the provided message
   *
   * @param env A pointer to the Java environment
   * @param msg The message for the exception
   *
   * @return true if an exception was thrown, false otherwise
   */
  static bool ThrowNew(JNIEnv* env, const std::string& msg) {
    return JavaException::ThrowNew(env, msg);
  }
  /**
   * Create and throw a Java OutOfMemoryError with the provided status
   *
   * If s.ok() == true, then this function will not throw any exception.
   *
   * @param env A pointer to the Java environment
   * @param s The status for the exception
   *
   * @return true if an exception was thrown, false otherwise
   */
  static bool ThrowNew(JNIEnv* env, const Status& s) {
    assert(!s.ok());
    if (s.ok()) {
      return false;
    }
    // get the OutOfMemoryError class
    jclass jclazz = getJClass(env);
    if (jclazz == nullptr) {
      // exception occurred accessing class
      std::cerr << "OutOfMemoryErrorJni::ThrowNew/class - Error: "
                   "unexpected exception!"
                << std::endl;
      return env->ExceptionCheck();
    }
    return JavaException::ThrowNew(env, s.ToString());
  }
 };
 // The portal class for org.rocksdb.Status.Code
 class CodeJni : public JavaClass {
 public:
--- a/java/rocksjni/rocksjni.cc
+++ b/java/rocksjni/rocksjni.cc
--- a/java/rocksjni/transaction.cc
+++ b/java/rocksjni/transaction.cc
@ -14,6 +14,7 @@
 #include "include/org_rocksdb_Transaction.h"
 #include "rocksjni/cplusplus_to_java_convert.h"
 #include "rocksjni/jni_multiget_helpers.h"
 #include "rocksjni/kv_helper.h"
 #include "rocksjni/portal.h"
@ -230,7 +231,7 @@ jint Java_org_rocksdb_Transaction_get__JJ_3BII_3BIIJ(
        env, txn->Get(*read_options, column_family_handle, key.slice(),
                      &value.pinnable_slice()));
    return value.Fetch();
-  } catch (const ROCKSDB_NAMESPACE::KVException& e) {
+  } catch (ROCKSDB_NAMESPACE::KVException& e) {
    return e.Code();
  }
 }
@ -267,31 +268,6 @@ jint Java_org_rocksdb_Transaction_getDirect(JNIEnv* env, jclass, jlong jhandle,
  }
 }
 // TODO(AR) consider refactoring to share this between here and rocksjni.cc
 // used by txn_multi_get_helper below
 std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*> txn_column_families_helper(
    JNIEnv* env, jlongArray jcolumn_family_handles, bool* has_exception) {
  std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*> cf_handles;
  if (jcolumn_family_handles != nullptr) {
    const jsize len_cols = env->GetArrayLength(jcolumn_family_handles);
    if (len_cols > 0) {
      jlong* jcfh = env->GetLongArrayElements(jcolumn_family_handles, nullptr);
      if (jcfh == nullptr) {
        // exception thrown: OutOfMemoryError
        *has_exception = JNI_TRUE;
        return std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>();
      }
      for (int i = 0; i < len_cols; i++) {
        auto* cf_handle =
            reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>(jcfh[i]);
        cf_handles.push_back(cf_handle);
      }
      env->ReleaseLongArrayElements(jcolumn_family_handles, jcfh, JNI_ABORT);
    }
  }
  return cf_handles;
 }
 typedef std::function<std::vector<ROCKSDB_NAMESPACE::Status>(
    const ROCKSDB_NAMESPACE::ReadOptions&,
    const std::vector<ROCKSDB_NAMESPACE::Slice>&, std::vector<std::string>*)>
@ -316,91 +292,6 @@ void free_key_values(std::vector<jbyte*>& keys_to_free) {
  }
 }
 // TODO(AR) consider refactoring to share this between here and rocksjni.cc
 // cf multi get
 jobjectArray txn_multi_get_helper(JNIEnv* env, const FnMultiGet& fn_multi_get,
                                  const jlong& jread_options_handle,
                                  const jobjectArray& jkey_parts) {
  const jsize len_key_parts = env->GetArrayLength(jkey_parts);
  std::vector<ROCKSDB_NAMESPACE::Slice> key_parts;
  std::vector<jbyte*> keys_to_free;
  for (int i = 0; i < len_key_parts; i++) {
    const jobject jk = env->GetObjectArrayElement(jkey_parts, i);
    if (env->ExceptionCheck()) {
      // exception thrown: ArrayIndexOutOfBoundsException
      free_key_values(keys_to_free);
      return nullptr;
    }
    jbyteArray jk_ba = reinterpret_cast<jbyteArray>(jk);
    const jsize len_key = env->GetArrayLength(jk_ba);
    jbyte* jk_val = new jbyte[len_key];
    if (jk_val == nullptr) {
      // exception thrown: OutOfMemoryError
      env->DeleteLocalRef(jk);
      free_key_values(keys_to_free);
      jclass exception_cls = (env)->FindClass("java/lang/OutOfMemoryError");
      (env)->ThrowNew(exception_cls,
                      "Insufficient Memory for CF handle array.");
      return nullptr;
    }
    env->GetByteArrayRegion(jk_ba, 0, len_key, jk_val);
    ROCKSDB_NAMESPACE::Slice key_slice(reinterpret_cast<char*>(jk_val),
                                       len_key);
    key_parts.push_back(key_slice);
    keys_to_free.push_back(jk_val);
    env->DeleteLocalRef(jk);
  }
  auto* read_options =
      reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle);
  std::vector<std::string> value_parts;
  std::vector<ROCKSDB_NAMESPACE::Status> s =
      fn_multi_get(*read_options, key_parts, &value_parts);
  // free up allocated byte arrays
  free_key_values(keys_to_free);
  // prepare the results
  const jclass jcls_ba = env->FindClass("[B");
  jobjectArray jresults =
      env->NewObjectArray(static_cast<jsize>(s.size()), jcls_ba, nullptr);
  if (jresults == nullptr) {
    // exception thrown: OutOfMemoryError
    return nullptr;
  }
  // add to the jresults
  for (std::vector<ROCKSDB_NAMESPACE::Status>::size_type i = 0; i != s.size();
       i++) {
    if (s[i].ok()) {
      jbyteArray jentry_value =
          env->NewByteArray(static_cast<jsize>(value_parts[i].size()));
      if (jentry_value == nullptr) {
        // exception thrown: OutOfMemoryError
        return nullptr;
      }
      env->SetByteArrayRegion(
          jentry_value, 0, static_cast<jsize>(value_parts[i].size()),
          const_cast<jbyte*>(
              reinterpret_cast<const jbyte*>(value_parts[i].c_str())));
      if (env->ExceptionCheck()) {
        // exception thrown: ArrayIndexOutOfBoundsException
        env->DeleteLocalRef(jentry_value);
        return nullptr;
      }
      env->SetObjectArrayElement(jresults, static_cast<jsize>(i), jentry_value);
      env->DeleteLocalRef(jentry_value);
    }
  }
  return jresults;
 }
 /*
 * Class:     org_rocksdb_Transaction
 * Method:    multiGet
@ -409,24 +300,22 @@ jobjectArray txn_multi_get_helper(JNIEnv* env, const FnMultiGet& fn_multi_get,
 jobjectArray Java_org_rocksdb_Transaction_multiGet__JJ_3_3B_3J(
    JNIEnv* env, jclass /*jobj*/, jlong jhandle, jlong jread_options_handle,
    jobjectArray jkey_parts, jlongArray jcolumn_family_handles) {
-  bool has_exception = false;
+  ROCKSDB_NAMESPACE::MultiGetJNIKeys keys;
-  const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>
+  if (!keys.fromByteArrays(env, jkey_parts)) {
      column_family_handles = txn_column_families_helper(
          env, jcolumn_family_handles, &has_exception);
  if (has_exception) {
    // exception thrown: OutOfMemoryError
    return nullptr;
  }
  auto cf_handles =
      ROCKSDB_NAMESPACE::ColumnFamilyJNIHelpers::handlesFromJLongArray(
          env, jcolumn_family_handles);
  if (!cf_handles) return nullptr;
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
-  FnMultiGet fn_multi_get = std::bind<std::vector<ROCKSDB_NAMESPACE::Status> (
+  std::vector<std::string> values(keys.size());
-      ROCKSDB_NAMESPACE::Transaction::*)(
+  std::vector<ROCKSDB_NAMESPACE::Status> statuses = txn->MultiGet(
-      const ROCKSDB_NAMESPACE::ReadOptions&,
+      *reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle),
-      const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>&,
+      *cf_handles, keys.slices(), &values);
-      const std::vector<ROCKSDB_NAMESPACE::Slice>&, std::vector<std::string>*)>(
+
-      &ROCKSDB_NAMESPACE::Transaction::MultiGet, txn, std::placeholders::_1,
+  return ROCKSDB_NAMESPACE::MultiGetJNIValues::byteArrays(env, values,
-      column_family_handles, std::placeholders::_2, std::placeholders::_3);
+                                                          statuses);
  return txn_multi_get_helper(env, fn_multi_get, jread_options_handle,
                              jkey_parts);
 }
 /*
@ -437,15 +326,19 @@ jobjectArray Java_org_rocksdb_Transaction_multiGet__JJ_3_3B_3J(
 jobjectArray Java_org_rocksdb_Transaction_multiGet__JJ_3_3B(
    JNIEnv* env, jclass /*jobj*/, jlong jhandle, jlong jread_options_handle,
    jobjectArray jkey_parts) {
  ROCKSDB_NAMESPACE::MultiGetJNIKeys keys;
  if (!keys.fromByteArrays(env, jkey_parts)) {
    return nullptr;
  }
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
-  FnMultiGet fn_multi_get = std::bind<std::vector<ROCKSDB_NAMESPACE::Status> (
+  std::vector<std::string> values(keys.size());
-      ROCKSDB_NAMESPACE::Transaction::*)(
+  std::vector<ROCKSDB_NAMESPACE::Status> statuses = txn->MultiGet(
-      const ROCKSDB_NAMESPACE::ReadOptions&,
+      *reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle),
-      const std::vector<ROCKSDB_NAMESPACE::Slice>&, std::vector<std::string>*)>(
+      keys.slices(), &values);
-      &ROCKSDB_NAMESPACE::Transaction::MultiGet, txn, std::placeholders::_1,
+
-      std::placeholders::_2, std::placeholders::_3);
+  return ROCKSDB_NAMESPACE::MultiGetJNIValues::byteArrays(env, values,
-  return txn_multi_get_helper(env, fn_multi_get, jread_options_handle,
+                                                          statuses);
                              jkey_parts);
 }
 /*
@ -486,12 +379,12 @@ jint Java_org_rocksdb_Transaction_getForUpdate__JJ_3BII_3BIIJZZ(
    jbyteArray jkey, jint jkey_off, jint jkey_part_len, jbyteArray jval,
    jint jval_off, jint jval_len, jlong jcolumn_family_handle,
    jboolean jexclusive, jboolean jdo_validate) {
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
  auto* read_options =
      reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle);
  auto* column_family_handle =
      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>(
          jcolumn_family_handle);
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
  try {
    ROCKSDB_NAMESPACE::JByteArraySlice key(env, jkey, jkey_off, jkey_part_len);
    ROCKSDB_NAMESPACE::JByteArrayPinnableSlice value(env, jval, jval_off,
@ -501,6 +394,36 @@ jint Java_org_rocksdb_Transaction_getForUpdate__JJ_3BII_3BIIJZZ(
        txn->GetForUpdate(*read_options, column_family_handle, key.slice(),
                          &value.pinnable_slice(), jexclusive, jdo_validate));
    return value.Fetch();
  } catch (ROCKSDB_NAMESPACE::KVException& e) {
    return e.Code();
  }
 }
 /*
 * Class:     org_rocksdb_Transaction
 * Method:    getDirect
 * Signature: (JJLjava/nio/ByteBuffer;IILjava/nio/ByteBuffer;IIJZZ)I
 */
 jint Java_org_rocksdb_Transaction_getDirect(
    JNIEnv* env, jobject /*jobj*/, jlong jhandle, jlong jread_options_handle,
    jobject jkey_bb, jint jkey_off, jint jkey_part_len, jobject jval_bb,
    jint jval_off, jint jval_len, jlong jcolumn_family_handle) {
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
  auto* read_options =
      reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle);
  auto* column_family_handle =
      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>(
          jcolumn_family_handle);
  try {
    ROCKSDB_NAMESPACE::JDirectBufferSlice key(env, jkey_bb, jkey_off,
                                              jkey_part_len);
    ROCKSDB_NAMESPACE::JDirectBufferPinnableSlice value(env, jval_bb, jval_off,
                                                        jval_len);
    ROCKSDB_NAMESPACE::KVException::ThrowOnError(
        env, txn->Get(*read_options, column_family_handle, key.slice(),
                      &value.pinnable_slice()));
    return value.Fetch();
  } catch (const ROCKSDB_NAMESPACE::KVException& e) {
    return e.Code();
  }
@ -546,25 +469,22 @@ jint Java_org_rocksdb_Transaction_getDirectForUpdate(
 jobjectArray Java_org_rocksdb_Transaction_multiGetForUpdate__JJ_3_3B_3J(
    JNIEnv* env, jclass, jlong jhandle, jlong jread_options_handle,
    jobjectArray jkey_parts, jlongArray jcolumn_family_handles) {
-  bool has_exception = false;
+  ROCKSDB_NAMESPACE::MultiGetJNIKeys keys;
-  const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>
+  if (!keys.fromByteArrays(env, jkey_parts)) {
      column_family_handles = txn_column_families_helper(
          env, jcolumn_family_handles, &has_exception);
  if (has_exception) {
    // exception thrown: OutOfMemoryError
    return nullptr;
  }
  auto cf_handles =
      ROCKSDB_NAMESPACE::ColumnFamilyJNIHelpers::handlesFromJLongArray(
          env, jcolumn_family_handles);
  if (!cf_handles) return nullptr;
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
-  FnMultiGet fn_multi_get_for_update = std::bind<std::vector<
+  std::vector<std::string> values(keys.size());
-      ROCKSDB_NAMESPACE::Status> (ROCKSDB_NAMESPACE::Transaction::*)(
+  std::vector<ROCKSDB_NAMESPACE::Status> statuses = txn->MultiGetForUpdate(
-      const ROCKSDB_NAMESPACE::ReadOptions&,
+      *reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle),
-      const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*>&,
+      *cf_handles, keys.slices(), &values);
-      const std::vector<ROCKSDB_NAMESPACE::Slice>&, std::vector<std::string>*)>(
+
-      &ROCKSDB_NAMESPACE::Transaction::MultiGetForUpdate, txn,
+  return ROCKSDB_NAMESPACE::MultiGetJNIValues::byteArrays(env, values,
-      std::placeholders::_1, column_family_handles, std::placeholders::_2,
+                                                          statuses);
      std::placeholders::_3);
  return txn_multi_get_helper(env, fn_multi_get_for_update,
                              jread_options_handle, jkey_parts);
 }
 /*
@ -575,15 +495,19 @@ jobjectArray Java_org_rocksdb_Transaction_multiGetForUpdate__JJ_3_3B_3J(
 jobjectArray Java_org_rocksdb_Transaction_multiGetForUpdate__JJ_3_3B(
    JNIEnv* env, jclass /*jobj*/, jlong jhandle, jlong jread_options_handle,
    jobjectArray jkey_parts) {
  ROCKSDB_NAMESPACE::MultiGetJNIKeys keys;
  if (!keys.fromByteArrays(env, jkey_parts)) {
    return nullptr;
  }
  auto* txn = reinterpret_cast<ROCKSDB_NAMESPACE::Transaction*>(jhandle);
-  FnMultiGet fn_multi_get_for_update = std::bind<std::vector<
+  std::vector<std::string> values(keys.size());
-      ROCKSDB_NAMESPACE::Status> (ROCKSDB_NAMESPACE::Transaction::*)(
+  std::vector<ROCKSDB_NAMESPACE::Status> statuses = txn->MultiGetForUpdate(
-      const ROCKSDB_NAMESPACE::ReadOptions&,
+      *reinterpret_cast<ROCKSDB_NAMESPACE::ReadOptions*>(jread_options_handle),
-      const std::vector<ROCKSDB_NAMESPACE::Slice>&, std::vector<std::string>*)>(
+      keys.slices(), &values);
-      &ROCKSDB_NAMESPACE::Transaction::MultiGetForUpdate, txn,
+
-      std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
+  return ROCKSDB_NAMESPACE::MultiGetJNIValues::byteArrays(env, values,
-  return txn_multi_get_helper(env, fn_multi_get_for_update,
+                                                          statuses);
                              jread_options_handle, jkey_parts);
 }
 /*
--- a/java/src/main/java/org/rocksdb/RocksDB.java
+++ b/java/src/main/java/org/rocksdb/RocksDB.java
@ -2315,7 +2315,7 @@ public class RocksDB extends RocksObject {
      final List<byte[]> keys) throws RocksDBException,
      IllegalArgumentException {
    assert (!keys.isEmpty());
-    // Check if key size equals cfList size. If not a exception must be
+    // Check if key size equals cfList size. If not an exception must be
    // thrown. If not a Segmentation fault happens.
    if (keys.size() != columnFamilyHandleList.size()) {
        throw new IllegalArgumentException(
@ -2505,7 +2505,8 @@ public class RocksDB extends RocksObject {
    // Check if key size equals cfList size. If not a exception must be
    // thrown. If not a Segmentation fault happens.
    if (values.size() != keys.size()) {
-      throw new IllegalArgumentException("For each key there must be a corresponding value.");
+      throw new IllegalArgumentException("For each key there must be a corresponding value. "
          + keys.size() + " keys were supplied, but " + values.size() + " values were supplied.");
    }
    // TODO (AP) support indirect buffers
@ -2555,6 +2556,12 @@ public class RocksDB extends RocksObject {
        value.position(Math.min(valuesSizeArray[i], value.capacity()));
        value.flip(); // prepare for read out
        results.add(new ByteBufferGetStatus(status, valuesSizeArray[i], value));
      } else if (status.getCode() == Status.Code.Incomplete) {
        assert valuesSizeArray[i] == -1;
        final ByteBuffer value = valuesArray[i];
        value.position(value.capacity());
        value.flip(); // prepare for read out
        results.add(new ByteBufferGetStatus(status, value.capacity(), value));
      } else {
        results.add(new ByteBufferGetStatus(status));
      }
--- a/java/src/test/java/org/rocksdb/ColumnFamilyTest.java
+++ b/java/src/test/java/org/rocksdb/ColumnFamilyTest.java
@ -391,13 +391,92 @@ public class ColumnFamilyTest {
      assertThat(getResult).isEqualTo(RocksDB.NOT_FOUND);
      // found value which fits in outValue
      getResult = db.get(columnFamilyHandleList.get(0), "key1".getBytes(), outValue);
-      assertThat(getResult).isNotEqualTo(RocksDB.NOT_FOUND);
+      assertThat(getResult).isEqualTo("value".getBytes().length);
      assertThat(outValue).isEqualTo("value".getBytes());
      // found value which fits partially
-      getResult =
+    }
-          db.get(columnFamilyHandleList.get(0), new ReadOptions(), "key2".getBytes(), outValue);
+  }
-      assertThat(getResult).isNotEqualTo(RocksDB.NOT_FOUND);
+
-      assertThat(outValue).isEqualTo("12345".getBytes());
+  @Test
  public void getWithOutValueAndCfPartial() throws RocksDBException {
    final List<ColumnFamilyDescriptor> cfDescriptors =
        Collections.singletonList(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    final List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<>();
    try (final DBOptions options =
             new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true);
         final RocksDB db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
             cfDescriptors, columnFamilyHandleList)) {
      db.put(columnFamilyHandleList.get(0), "key1".getBytes(), "value".getBytes());
      db.put("key2".getBytes(), "12345678".getBytes());
      final byte[] partialOutValue = new byte[5];
      int getResult = db.get(columnFamilyHandleList.get(0), "key2".getBytes(), partialOutValue);
      assertThat(getResult).isEqualTo("12345678".getBytes().length);
      assertThat(partialOutValue).isEqualTo("12345".getBytes());
      final byte[] offsetKeyValue = "abckey2hjk".getBytes();
      assertThat(offsetKeyValue.length).isEqualTo(10);
      final byte[] offsetOutValue = "abcdefghjk".getBytes();
      assertThat(offsetOutValue.length).isEqualTo(10);
      getResult = db.get(columnFamilyHandleList.get(0), offsetKeyValue, 3, 4, offsetOutValue, 2, 5);
      assertThat(getResult).isEqualTo("12345678".getBytes().length);
      assertThat(offsetOutValue).isEqualTo("ab12345hjk".getBytes());
    }
  }
  @Test
  public void getWithOutValueAndCfPartialAndOptions() throws RocksDBException {
    final List<ColumnFamilyDescriptor> cfDescriptors =
        Collections.singletonList(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    final List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<>();
    try (final DBOptions options =
             new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true);
         final RocksDB db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
             cfDescriptors, columnFamilyHandleList)) {
      db.put(
          columnFamilyHandleList.get(0), new WriteOptions(), "key1".getBytes(), "value".getBytes());
      db.put("key2".getBytes(), "12345678".getBytes());
      final byte[] partialOutValue = new byte[5];
      int getResult = db.get(
          columnFamilyHandleList.get(0), new ReadOptions(), "key2".getBytes(), partialOutValue);
      assertThat(getResult).isEqualTo("12345678".getBytes().length);
      assertThat(partialOutValue).isEqualTo("12345".getBytes());
      final byte[] offsetKeyValue = "abckey2hjk".getBytes();
      assertThat(offsetKeyValue.length).isEqualTo(10);
      final byte[] offsetOutValue = "abcdefghjk".getBytes();
      assertThat(offsetOutValue.length).isEqualTo(10);
      getResult = db.get(columnFamilyHandleList.get(0), new ReadOptions(), offsetKeyValue, 3, 4,
          offsetOutValue, 2, 5);
      assertThat(getResult).isEqualTo("12345678".getBytes().length);
      assertThat(offsetOutValue).isEqualTo("ab12345hjk".getBytes());
    }
  }
  @Test(expected = IndexOutOfBoundsException.class)
  public void getWithOutValueAndCfIndexOutOfBounds() throws RocksDBException {
    final List<ColumnFamilyDescriptor> cfDescriptors =
        Collections.singletonList(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
    final List<ColumnFamilyHandle> columnFamilyHandleList = new ArrayList<>();
    try (final DBOptions options =
             new DBOptions().setCreateIfMissing(true).setCreateMissingColumnFamilies(true);
         final RocksDB db = RocksDB.open(options, dbFolder.getRoot().getAbsolutePath(),
             cfDescriptors, columnFamilyHandleList)) {
      db.put(
          columnFamilyHandleList.get(0), new WriteOptions(), "key1".getBytes(), "value".getBytes());
      db.put("key2".getBytes(), "12345678".getBytes());
      final byte[] offsetKeyValue = "abckey2hjk".getBytes();
      final byte[] partialOutValue = new byte[5];
      int getResult = db.get(columnFamilyHandleList.get(0), new ReadOptions(), offsetKeyValue, 3, 4,
          partialOutValue, 2, 5);
    }
  }
@ -519,8 +598,14 @@ public class ColumnFamilyTest {
    }
  }
-  @Test
+  @FunctionalInterface
-  public void multiGet() throws RocksDBException {
+  public interface RocksDBTriFunction<T1, T2, T3, R> {
    R apply(T1 t1, T2 t2, T3 t3) throws IllegalArgumentException, RocksDBException;
  }
  private void multiGetHelper(
      RocksDBTriFunction<RocksDB, List<ColumnFamilyHandle>, List<byte[]>, List<byte[]>> multiGetter)
      throws RocksDBException {
    final List<ColumnFamilyDescriptor> cfDescriptors = Arrays.asList(
        new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY),
        new ColumnFamilyDescriptor("new_cf".getBytes()));
@ -536,17 +621,24 @@ public class ColumnFamilyTest {
      final List<byte[]> keys = Arrays.asList("key".getBytes(), "newcfkey".getBytes());
-      List<byte[]> retValues = db.multiGetAsList(columnFamilyHandleList, keys);
+      List<byte[]> retValues = multiGetter.apply(db, columnFamilyHandleList, keys);
      assertThat(retValues.size()).isEqualTo(2);
      assertThat(new String(retValues.get(0))).isEqualTo("value");
      assertThat(new String(retValues.get(1))).isEqualTo("value");
      retValues = db.multiGetAsList(new ReadOptions(), columnFamilyHandleList, keys);
      assertThat(retValues.size()).isEqualTo(2);
      assertThat(new String(retValues.get(0))).isEqualTo("value");
      assertThat(new String(retValues.get(1))).isEqualTo("value");
    }
  }
  @Test
  public void multiGet() throws RocksDBException {
    multiGetHelper(RocksDB::multiGetAsList);
  }
  @Test
  public void multiGetReadOptions() throws RocksDBException {
    multiGetHelper(
        (db, columnFamilies, keys) -> db.multiGetAsList(new ReadOptions(), columnFamilies, keys));
  }
  @Test
  public void multiGetAsList() throws RocksDBException {
    final List<ColumnFamilyDescriptor> cfDescriptors = Arrays.asList(
--- a/java/src/test/java/org/rocksdb/MultiGetTest.java
+++ b/java/src/test/java/org/rocksdb/MultiGetTest.java
@ -4,14 +4,14 @@
 //  (found in the LICENSE.Apache file in the root directory).
 package org.rocksdb;
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 import static org.assertj.core.api.Assertions.fail;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import org.junit.ClassRule;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@ -24,8 +24,13 @@ public class MultiGetTest {
  @Rule public TemporaryFolder dbFolder = new TemporaryFolder();
-  @Test
+  @FunctionalInterface
-  public void putNThenMultiGet() throws RocksDBException {
+  public interface RocksDBBiFunction<T1, T2, R> {
    R apply(T1 t1, T2 t2) throws RocksDBException;
  }
  private void putNThenMultiGetHelper(
      RocksDBBiFunction<RocksDB, List<byte[]>, List<byte[]>> multiGetter) throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      db.put("key1".getBytes(), "value1ForKey1".getBytes());
@ -33,7 +38,7 @@ public class MultiGetTest {
      db.put("key3".getBytes(), "value3ForKey3".getBytes());
      final List<byte[]> keys =
          Arrays.asList("key1".getBytes(), "key2".getBytes(), "key3".getBytes());
-      final List<byte[]> values = db.multiGetAsList(keys);
+      final List<byte[]> values = multiGetter.apply(db, keys);
      assertThat(values.size()).isEqualTo(keys.size());
      assertThat(values.get(0)).isEqualTo("value1ForKey1".getBytes());
      assertThat(values.get(1)).isEqualTo("value2ForKey2".getBytes());
@ -41,6 +46,42 @@ public class MultiGetTest {
    }
  }
  private void putNThenMultiGetHelperWithMissing(
      RocksDBBiFunction<RocksDB, List<byte[]>, List<byte[]>> multiGetter) throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      db.put("key1".getBytes(), "value1ForKey1".getBytes());
      db.put("key3".getBytes(), "value3ForKey3".getBytes());
      final List<byte[]> keys =
          Arrays.asList("key1".getBytes(), "key2".getBytes(), "key3".getBytes());
      final List<byte[]> values = multiGetter.apply(db, keys);
      assertThat(values.size()).isEqualTo(keys.size());
      assertThat(values.get(0)).isEqualTo("value1ForKey1".getBytes());
      assertThat(values.get(1)).isEqualTo(null);
      assertThat(values.get(2)).isEqualTo("value3ForKey3".getBytes());
    }
  }
  @Test
  public void putNThenMultiGet() throws RocksDBException {
    putNThenMultiGetHelper(RocksDB::multiGetAsList);
  }
  @Test
  public void putNThenMultiGetWithMissing() throws RocksDBException {
    putNThenMultiGetHelperWithMissing(RocksDB::multiGetAsList);
  }
  @Test
  public void putNThenMultiGetReadOptions() throws RocksDBException {
    putNThenMultiGetHelper((db, keys) -> db.multiGetAsList(new ReadOptions(), keys));
  }
  @Test
  public void putNThenMultiGetReadOptionsWithMissing() throws RocksDBException {
    putNThenMultiGetHelperWithMissing((db, keys) -> db.multiGetAsList(new ReadOptions(), keys));
  }
  @Test
  public void putNThenMultiGetDirect() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
@ -103,6 +144,65 @@ public class MultiGetTest {
    }
  }
  @Test
  public void putNThenMultiGetDirectWithMissing() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      db.put("key1".getBytes(), "value1ForKey1".getBytes());
      db.put("key3".getBytes(), "value3ForKey3".getBytes());
      final List<ByteBuffer> keys = new ArrayList<>();
      keys.add(ByteBuffer.allocateDirect(12).put("key1".getBytes()));
      keys.add(ByteBuffer.allocateDirect(12).put("key2".getBytes()));
      keys.add(ByteBuffer.allocateDirect(12).put("key3".getBytes()));
      // Java8 and lower flip() returns Buffer not ByteBuffer, so can't chain above /\/\
      for (final ByteBuffer key : keys) {
        key.flip();
      }
      final List<ByteBuffer> values = new ArrayList<>();
      for (int i = 0; i < keys.size(); i++) {
        values.add(ByteBuffer.allocateDirect(24));
      }
      {
        final List<ByteBufferGetStatus> results = db.multiGetByteBuffers(keys, values);
        assertThat(results.get(0).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(1).status.getCode()).isEqualTo(Status.Code.NotFound);
        assertThat(results.get(2).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(0).requiredSize).isEqualTo("value1ForKey1".getBytes().length);
        assertThat(results.get(1).requiredSize).isEqualTo(0);
        assertThat(results.get(2).requiredSize).isEqualTo("value3ForKey3".getBytes().length);
        assertThat(TestUtil.bufferBytes(results.get(0).value))
            .isEqualTo("value1ForKey1".getBytes());
        assertThat(results.get(1).value).isNull();
        assertThat(TestUtil.bufferBytes(results.get(2).value))
            .isEqualTo("value3ForKey3".getBytes());
      }
      {
        final List<ByteBufferGetStatus> results =
            db.multiGetByteBuffers(new ReadOptions(), keys, values);
        assertThat(results.get(0).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(1).status.getCode()).isEqualTo(Status.Code.NotFound);
        assertThat(results.get(2).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(0).requiredSize).isEqualTo("value1ForKey1".getBytes().length);
        assertThat(results.get(1).requiredSize).isEqualTo(0);
        assertThat(results.get(2).requiredSize).isEqualTo("value3ForKey3".getBytes().length);
        assertThat(TestUtil.bufferBytes(results.get(0).value))
            .isEqualTo("value1ForKey1".getBytes());
        assertThat(results.get(1).value).isNull();
        assertThat(TestUtil.bufferBytes(results.get(2).value))
            .isEqualTo("value3ForKey3".getBytes());
      }
    }
  }
  @Test
  public void putNThenMultiGetDirectSliced() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
@ -146,6 +246,47 @@ public class MultiGetTest {
    }
  }
  @Test
  public void putNThenMultiGetDirectSlicedWithMissing() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      db.put("key1".getBytes(), "value1ForKey1".getBytes());
      db.put("key3".getBytes(), "value3ForKey3".getBytes());
      final List<ByteBuffer> keys = new ArrayList<>();
      keys.add(ByteBuffer.allocateDirect(12).put("key2".getBytes()));
      keys.add(ByteBuffer.allocateDirect(12).put("key3".getBytes()));
      keys.add(
          ByteBuffer.allocateDirect(12).put("prefix1".getBytes()).slice().put("key1".getBytes()));
      // Java8 and lower flip() returns Buffer not ByteBuffer, so can't chain above /\/\
      for (final ByteBuffer key : keys) {
        key.flip();
      }
      final List<ByteBuffer> values = new ArrayList<>();
      for (int i = 0; i < keys.size(); i++) {
        values.add(ByteBuffer.allocateDirect(24));
      }
      {
        final List<ByteBufferGetStatus> results = db.multiGetByteBuffers(keys, values);
        assertThat(results.get(0).status.getCode()).isEqualTo(Status.Code.NotFound);
        assertThat(results.get(1).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(2).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(1).requiredSize).isEqualTo("value3ForKey3".getBytes().length);
        assertThat(results.get(2).requiredSize).isEqualTo("value1ForKey1".getBytes().length);
        assertThat(results.get(0).requiredSize).isEqualTo(0);
        assertThat(results.get(0).value).isNull();
        assertThat(TestUtil.bufferBytes(results.get(1).value))
            .isEqualTo("value3ForKey3".getBytes());
        assertThat(TestUtil.bufferBytes(results.get(2).value))
            .isEqualTo("value1ForKey1".getBytes());
      }
    }
  }
  @Test
  public void putNThenMultiGetDirectBadValuesArray() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
@ -315,6 +456,39 @@ public class MultiGetTest {
        assertThat(TestUtil.bufferBytes(results.get(2).value))
            .isEqualTo("value3ForKey3".getBytes());
      }
      {
        final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
        columnFamilyHandles.add(cf.get(0));
        columnFamilyHandles.add(cf.get(0));
        columnFamilyHandles.add(cf.get(0));
        final List<ByteBuffer> keysWithMissing = new ArrayList<>();
        keysWithMissing.add(ByteBuffer.allocateDirect(12).put("key1".getBytes()));
        keysWithMissing.add(ByteBuffer.allocateDirect(12).put("key3Bad".getBytes()));
        keysWithMissing.add(ByteBuffer.allocateDirect(12).put("key3".getBytes()));
        // Java8 and lower flip() returns Buffer not ByteBuffer, so can't chain above /\/\
        for (final ByteBuffer key : keysWithMissing) {
          key.flip();
        }
        final List<ByteBufferGetStatus> results =
            db.multiGetByteBuffers(columnFamilyHandles, keysWithMissing, values);
        assertThat(results.get(0).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(1).status.getCode()).isEqualTo(Status.Code.NotFound);
        assertThat(results.get(2).status.getCode()).isEqualTo(Status.Code.Ok);
        assertThat(results.get(0).requiredSize).isEqualTo("value1ForKey1".getBytes().length);
        assertThat(results.get(1).requiredSize).isEqualTo(0);
        assertThat(results.get(2).requiredSize).isEqualTo("value3ForKey3".getBytes().length);
        assertThat(TestUtil.bufferBytes(results.get(0).value))
            .isEqualTo("value1ForKey1".getBytes());
        assertThat(results.get(1).value).isNull();
        assertThat(TestUtil.bufferBytes(results.get(2).value))
            .isEqualTo("value3ForKey3".getBytes());
      }
    }
  }
@ -527,4 +701,321 @@ public class MultiGetTest {
      }
    }
  }
  /**
   *
   * @param db database to write to
   * @param key key to write
   * @return expected size of data written
   * @throws RocksDBException if {@code put} or {@code merge} fail
   */
  private long createIntOverflowValue(
      final RocksDB db, final ColumnFamilyHandle cf, final String key) throws RocksDBException {
    final int BUFSIZE = 100000000;
    final int BUFCOUNT = 30;
    final byte[] wbuf = new byte[BUFSIZE];
    Arrays.fill(wbuf, (byte) 10);
    for (int i = 0; i < BUFCOUNT; i++) {
      final byte[] vals = ("value" + i + "ForKey" + key).getBytes();
      System.arraycopy(vals, 0, wbuf, 0, vals.length);
      db.merge(cf, "key1".getBytes(), wbuf);
    }
    return ((long) BUFSIZE + 1) * BUFCOUNT - 1;
  }
  private void checkIntOVerflowValue(final ByteBuffer byteBuffer, final String key) {
    final int BUFSIZE = 100000000;
    final int BUFCOUNT = 30;
    for (int i = 0; i < BUFCOUNT; i++) {
      final byte[] vals = ("value" + i + "ForKey" + key).getBytes();
      final long position = (long) i * (BUFSIZE + 1);
      if (position > Integer.MAX_VALUE)
        break;
      byteBuffer.position((int) position);
      for (byte b : vals) {
        assertThat(byteBuffer.get()).isEqualTo(b);
      }
    }
  }
  private static ByteBuffer bbDirect(final String s) {
    final byte[] bytes = s.getBytes();
    final ByteBuffer byteBuffer = ByteBuffer.allocateDirect(bytes.length);
    byteBuffer.put(bytes);
    byteBuffer.flip();
    return byteBuffer;
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetDirect() throws RocksDBException {
    try (final Options opt =
             new Options().setCreateIfMissing(true).setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final long length = createIntOverflowValue(db, db.getDefaultColumnFamily(), "key1");
      db.put("key2".getBytes(), "value2ForKey2".getBytes());
      final List<ByteBuffer> byteBufferValues = new ArrayList<>();
      byteBufferValues.add(ByteBuffer.allocateDirect(Integer.MAX_VALUE));
      final List<ByteBuffer> byteBufferKeys = new ArrayList<>();
      byteBufferKeys.add(bbDirect("key1"));
      final List<ByteBufferGetStatus> statusList =
          db.multiGetByteBuffers(new ReadOptions(), byteBufferKeys, byteBufferValues);
      assertThat(statusList.size()).isEqualTo(1);
      final ByteBufferGetStatus status = statusList.get(0);
      assertThat(status.status.getCode()).isEqualTo(Status.Code.Incomplete);
      checkIntOVerflowValue(status.value, "key1");
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetDirectCF() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final ColumnFamilyOptions cfOptions =
             new ColumnFamilyOptions().setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>(0);
      cfDescriptors.add(new ColumnFamilyDescriptor("cf0".getBytes(), cfOptions));
      final List<ColumnFamilyHandle> cf = db.createColumnFamilies(cfDescriptors);
      final long length = createIntOverflowValue(db, cf.get(0), "key1");
      db.put(cf.get(0), "key2".getBytes(), "value2ForKey2".getBytes());
      final List<ByteBuffer> byteBufferValues = new ArrayList<>();
      byteBufferValues.add(ByteBuffer.allocateDirect(Integer.MAX_VALUE));
      final List<ByteBuffer> byteBufferKeys = new ArrayList<>();
      byteBufferKeys.add(bbDirect("key1"));
      final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
      columnFamilyHandles.add(cf.get(0));
      final List<ByteBufferGetStatus> statusList = db.multiGetByteBuffers(
          new ReadOptions(), columnFamilyHandles, byteBufferKeys, byteBufferValues);
      assertThat(statusList.size()).isEqualTo(1);
      final ByteBufferGetStatus status = statusList.get(0);
      assertThat(status.status.getCode()).isEqualTo(Status.Code.Incomplete);
      checkIntOVerflowValue(status.value, "key1");
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetDirect2Keys() throws RocksDBException {
    try (final Options opt =
             new Options().setCreateIfMissing(true).setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final long length = createIntOverflowValue(db, db.getDefaultColumnFamily(), "key1");
      db.put("key2".getBytes(), "value2ForKey2".getBytes());
      final List<ByteBuffer> byteBufferValues = new ArrayList<>();
      byteBufferValues.add(ByteBuffer.allocateDirect(Integer.MAX_VALUE));
      byteBufferValues.add(ByteBuffer.allocateDirect(12));
      final List<ByteBuffer> byteBufferKeys = new ArrayList<>();
      byteBufferKeys.add(bbDirect("key1"));
      byteBufferKeys.add(bbDirect("key2"));
      final List<ByteBufferGetStatus> statusList =
          db.multiGetByteBuffers(new ReadOptions(), byteBufferKeys, byteBufferValues);
      assertThat(statusList.size()).isEqualTo(2);
      assertThat(statusList.get(0).status.getCode()).isEqualTo(Status.Code.Incomplete);
      checkIntOVerflowValue(statusList.get(0).value, "key1");
      assertThat(statusList.get(1).status.getCode()).isEqualTo(Status.Code.Ok);
      final ByteBuffer bbKey2 = statusList.get(1).value;
      final byte[] bytes = new byte[bbKey2.capacity()];
      bbKey2.get(bytes);
      assertThat(bytes).isEqualTo("value2ForKey".getBytes());
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetDirect2KeysCF() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final ColumnFamilyOptions cfOptions =
             new ColumnFamilyOptions().setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>(0);
      cfDescriptors.add(new ColumnFamilyDescriptor("cf0".getBytes(), cfOptions));
      final List<ColumnFamilyHandle> cf = db.createColumnFamilies(cfDescriptors);
      final long length = createIntOverflowValue(db, cf.get(0), "key1");
      db.put(cf.get(0), "key2".getBytes(), "value2ForKey2".getBytes());
      final List<ByteBuffer> byteBufferValues = new ArrayList<>();
      byteBufferValues.add(ByteBuffer.allocateDirect(Integer.MAX_VALUE));
      byteBufferValues.add(ByteBuffer.allocateDirect(12));
      final List<ByteBuffer> byteBufferKeys = new ArrayList<>();
      byteBufferKeys.add(bbDirect("key1"));
      byteBufferKeys.add(bbDirect("key2"));
      final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
      columnFamilyHandles.add(cf.get(0));
      final List<ByteBufferGetStatus> statusList = db.multiGetByteBuffers(
          new ReadOptions(), columnFamilyHandles, byteBufferKeys, byteBufferValues);
      assertThat(statusList.size()).isEqualTo(2);
      assertThat(statusList.get(0).status.getCode()).isEqualTo(Status.Code.Incomplete);
      checkIntOVerflowValue(statusList.get(0).value, "key1");
      assertThat(statusList.get(1).status.getCode()).isEqualTo(Status.Code.Ok);
      final ByteBuffer bbKey2 = statusList.get(1).value;
      final byte[] bytes = new byte[bbKey2.capacity()];
      bbKey2.get(bytes);
      assertThat(bytes).isEqualTo("value2ForKey".getBytes());
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetAsList() throws RocksDBException {
    try (final Options opt =
             new Options().setCreateIfMissing(true).setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final long length = createIntOverflowValue(db, db.getDefaultColumnFamily(), "key1");
      db.put("key2".getBytes(), "value2ForKey2".getBytes());
      final List<byte[]> keys = new ArrayList<>();
      keys.add("key1".getBytes());
      assertThatThrownBy(() -> { db.multiGetAsList(keys); })
          .isInstanceOf(RocksDBException.class)
          .hasMessageContaining("Requested array size exceeds VM limit");
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetAsListCF() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final ColumnFamilyOptions cfOptions =
             new ColumnFamilyOptions().setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>(0);
      cfDescriptors.add(new ColumnFamilyDescriptor("cf0".getBytes(), cfOptions));
      final List<ColumnFamilyHandle> cf = db.createColumnFamilies(cfDescriptors);
      final long length = createIntOverflowValue(db, cf.get(0), "key1");
      db.put(cf.get(0), "key2".getBytes(), "value2ForKey2".getBytes());
      final List<byte[]> keys = new ArrayList<>();
      keys.add("key1".getBytes());
      assertThatThrownBy(() -> { db.multiGetAsList(cf, keys); })
          .isInstanceOf(RocksDBException.class)
          .hasMessageContaining("Requested array size exceeds VM limit");
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetAsList2Keys() throws RocksDBException {
    try (final Options opt =
             new Options().setCreateIfMissing(true).setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final long length = createIntOverflowValue(db, db.getDefaultColumnFamily(), "key1");
      db.put("key2".getBytes(), "value2ForKey2".getBytes());
      final List<byte[]> keys = new ArrayList<>();
      keys.add("key2".getBytes());
      keys.add("key1".getBytes());
      assertThatThrownBy(() -> { db.multiGetAsList(keys); })
          .isInstanceOf(RocksDBException.class)
          .hasMessageContaining("Requested array size exceeds VM limit");
    }
  }
  /**
   * Too slow/disk space dependent for CI
   * @throws RocksDBException
   */
  @Ignore
  @Test
  public void putBigMultiGetAsList2KeysCF() throws RocksDBException {
    try (final Options opt = new Options().setCreateIfMissing(true);
         final ColumnFamilyOptions cfOptions =
             new ColumnFamilyOptions().setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>(0);
      cfDescriptors.add(new ColumnFamilyDescriptor("cf0".getBytes(), cfOptions));
      final List<ColumnFamilyHandle> cf = db.createColumnFamilies(cfDescriptors);
      final long length = createIntOverflowValue(db, cf.get(0), "key1");
      db.put(cf.get(0), "key2".getBytes(), "value2ForKey2".getBytes());
      final List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
      columnFamilyHandles.add(cf.get(0));
      columnFamilyHandles.add(cf.get(0));
      final List<byte[]> keys = new ArrayList<>();
      keys.add("key2".getBytes());
      keys.add("key1".getBytes());
      assertThatThrownBy(() -> { db.multiGetAsList(columnFamilyHandles, keys); })
          .isInstanceOf(RocksDBException.class)
          .hasMessageContaining("Requested array size exceeds VM limit");
    }
  }
  /**
   * This eventually doesn't throw as expected
   * At about 3rd loop of asking (on a 64GB M1 Max Mac)
   * I presume it's a legitimate space exhaustion error in RocksDB,
   * but I think it worth having this here as a record.
   *
   * @throws RocksDBException
   */
  @Test
  @Ignore
  public void putBigMultiGetAsListRepeat() throws RocksDBException {
    try (final Options opt =
             new Options().setCreateIfMissing(true).setMergeOperatorName("stringappend");
         final RocksDB db = RocksDB.open(opt, dbFolder.getRoot().getAbsolutePath())) {
      final long length = createIntOverflowValue(db, db.getDefaultColumnFamily(), "key1");
      db.put("key2".getBytes(), "value2ForKey2".getBytes());
      final int REPEAT = 10;
      for (int i = 0; i < REPEAT; i++) {
        final List<byte[]> keys = new ArrayList<>();
        keys.add("key1".getBytes());
        assertThatThrownBy(() -> { db.multiGetAsList(keys); })
            .isInstanceOf(RocksDBException.class)
            .hasMessageContaining("Requested array size exceeds VM limit");
      }
    }
  }
 }
--- a/java/src/test/java/org/rocksdb/RocksDBTest.java
+++ b/java/src/test/java/org/rocksdb/RocksDBTest.java
@ -219,15 +219,31 @@ public class RocksDBTest {
      key.position(4);
      final ByteBuffer result2 = ByteBuffer.allocateDirect(12);
      result2.put("abcdefghijkl".getBytes());
      result2.flip().position(3);
      assertThat(db.get(optr, key, result2)).isEqualTo(4);
      assertThat(result2.position()).isEqualTo(3);
      assertThat(result2.limit()).isEqualTo(7);
      assertThat(key.position()).isEqualTo(8);
      assertThat(key.limit()).isEqualTo(8);
      final byte[] tmp2 = new byte[12];
      result2.position(0).limit(12);
      result2.get(tmp2);
      assertThat(tmp2).isEqualTo("abcval3hijkl".getBytes());
      key.position(4);
      result.clear().position(9);
      assertThat(db.get(optr, key, result)).isEqualTo(4);
      assertThat(result.position()).isEqualTo(9);
      assertThat(result.limit()).isEqualTo(12);
      assertThat(key.position()).isEqualTo(8);
      assertThat(key.limit()).isEqualTo(8);
-      final byte[] tmp2 = new byte[3];
+      final byte[] tmp3 = new byte[3];
-      result.get(tmp2);
+      result.get(tmp3);
-      assertThat(tmp2).isEqualTo("val".getBytes());
+      assertThat(tmp3).isEqualTo("val".getBytes());
      // put
      final Segment key3 = sliceSegment("key3");
--- a/java/src/test/java/org/rocksdb/VerifyChecksumsTest.java
+++ b/java/src/test/java/org/rocksdb/VerifyChecksumsTest.java
@ -12,6 +12,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import org.junit.ClassRule;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
@ -127,9 +128,9 @@ public class VerifyChecksumsTest {
  /**
   * Run some operations and count the TickerType.BLOCK_CHECKSUM_COMPUTE_COUNT before and after
   * It should GO UP when the read options have checksum verification turned on.
-   * It shoulld REMAIN UNCHANGED when the read options have checksum verification turned off.
+   * It should REMAIN UNCHANGED when the read options have checksum verification turned off.
   * As the read options refer only to the read operations, there are still a few checksums
-   * performed outside this (blocks are getting loaded for lots of reasons, not aways directly due
+   * performed outside this (blocks are getting loaded for lots of reasons, not always directly due
   * to reads) but this test provides a good enough proxy for whether the flag is being noticed.
   *
   * @param operations the DB reading operations to perform which affect the checksum stats
@ -201,8 +202,11 @@ public class VerifyChecksumsTest {
    });
  }
  @Ignore(
      "The block checksum count looks as if it is not updated when a more optimized C++ multiGet is used.")
  @Test
-  public void verifyChecksumsMultiGet() throws RocksDBException {
+  public void
  verifyChecksumsMultiGet() throws RocksDBException {
    // noinspection AnonymousInnerClassMayBeStatic
    verifyChecksums(new Operations(KV_COUNT) {
      @Override
--- a/src.mk
+++ b/src.mk
@ -675,6 +675,7 @@ JNI_NATIVE_SOURCES =                                          \
  java/rocksjni/hyper_clock_cache.cc                          \
  java/rocksjni/iterator.cc                                   \
  java/rocksjni/jni_perf_context.cc                           \
  java/rocksjni/jni_multiget_helpers.cc                       \
  java/rocksjni/jnicallback.cc                                \
  java/rocksjni/loggerjnicallback.cc                          \
  java/rocksjni/lru_cache.cc                                  \
--- a/unreleased_history/performance_improvements/java_api_multiget.md
+++ b/unreleased_history/performance_improvements/java_api_multiget.md
@ -0,0 +1,14 @@
 Java API `multiGet()` variants now take advantage of the underlying batched `multiGet()` performance improvements.
 Before
 ```
 Benchmark (columnFamilyTestType) (keyCount) (keySize) (multiGetSize) (valueSize) Mode Cnt Score Error Units
 MultiGetBenchmarks.multiGetList10 no_column_family 10000 16 100 64 thrpt 25 6315.541 ± 8.106 ops/s
 MultiGetBenchmarks.multiGetList10 no_column_family 10000 16 100 1024 thrpt 25 6975.468 ± 68.964 ops/s
 ```
 After
 ```
 Benchmark (columnFamilyTestType) (keyCount) (keySize) (multiGetSize) (valueSize) Mode Cnt Score Error Units
 MultiGetBenchmarks.multiGetList10 no_column_family 10000 16 100 64 thrpt 25 7046.739 ± 13.299 ops/s
 MultiGetBenchmarks.multiGetList10 no_column_family 10000 16 100 1024 thrpt 25 7654.521 ± 60.121 ops/s
 ```
--- a/utilities/transactions/optimistic_transaction_test.cc
+++ b/utilities/transactions/optimistic_transaction_test.cc
@ -1217,7 +1217,8 @@ TEST_P(OptimisticTransactionTest, IteratorTest) {
    ASSERT_TRUE(iter->Valid());
    ASSERT_EQ(results[i], iter->value().ToString());
-    ASSERT_OK(txn->GetForUpdate(read_options, iter->key(), nullptr));
+    ASSERT_OK(
        txn->GetForUpdate(read_options, iter->key(), (std::string*)nullptr));
    iter->Next();
  }
--- a/utilities/transactions/transaction_base.h
+++ b/utilities/transactions/transaction_base.h
@ -84,6 +84,13 @@ class TransactionBaseImpl : public Transaction {
                        exclusive, do_validate);
  }
  Status GetForUpdate(const ReadOptions& options, const Slice& key,
                      PinnableSlice* pinnable_val, bool exclusive,
                      const bool do_validate) override {
    return GetForUpdate(options, db_->DefaultColumnFamily(), key, pinnable_val,
                        exclusive, do_validate);
  }
  using Transaction::MultiGet;
  std::vector<Status> MultiGet(
      const ReadOptions& _read_options,
--- a/utilities/transactions/transaction_test.cc
+++ b/utilities/transactions/transaction_test.cc
@ -250,6 +250,42 @@ TEST_P(TransactionTest, SuccessTest) {
  delete txn;
 }
 // Test the basic API of the pinnable slice overload of GetForUpdate()
 TEST_P(TransactionTest, SuccessTestPinnable) {
  ASSERT_OK(db->ResetStats());
  WriteOptions write_options;
  ReadOptions read_options;
  PinnableSlice pinnable_val;
  ASSERT_OK(db->Put(write_options, Slice("foo"), Slice("bar")));
  ASSERT_OK(db->Put(write_options, Slice("foo2"), Slice("bar")));
  Transaction* txn = db->BeginTransaction(write_options, TransactionOptions());
  ASSERT_TRUE(txn);
  ASSERT_EQ(0, txn->GetNumPuts());
  ASSERT_LE(0, txn->GetID());
  ASSERT_OK(txn->GetForUpdate(read_options, "foo", &pinnable_val));
  ASSERT_EQ(*pinnable_val.GetSelf(), std::string("bar"));
  ASSERT_OK(txn->Put(Slice("foo"), Slice("bar2")));
  ASSERT_EQ(1, txn->GetNumPuts());
  ASSERT_OK(txn->GetForUpdate(read_options, "foo", &pinnable_val));
  ASSERT_EQ(*pinnable_val.GetSelf(), std::string("bar2"));
  ASSERT_OK(txn->Commit());
  ASSERT_OK(
      db->Get(read_options, db->DefaultColumnFamily(), "foo", &pinnable_val));
  ASSERT_EQ(*pinnable_val.GetSelf(), std::string("bar2"));
  delete txn;
 }
 TEST_P(TransactionTest, SwitchMemtableDuringPrepareAndCommit_WC) {
  const TxnDBWritePolicy write_policy = std::get<2>(GetParam());
@ -543,13 +579,16 @@ TEST_P(TransactionTest, SharedLocks) {
  ASSERT_TRUE(txn3);
  // Test shared access between txns
-  s = txn1->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn3->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn3->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
  auto lock_data = db->GetLockStatusData();
@ -572,23 +611,25 @@ TEST_P(TransactionTest, SharedLocks) {
  ASSERT_OK(txn3->Rollback());
  // Test txn1 and txn2 sharing a lock and txn3 trying to obtain it.
-  s = txn1->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn3->GetForUpdate(read_options, "foo", nullptr);
+  s = txn3->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
  txn1->UndoGetForUpdate("foo");
-  s = txn3->GetForUpdate(read_options, "foo", nullptr);
+  s = txn3->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
  txn2->UndoGetForUpdate("foo");
-  s = txn3->GetForUpdate(read_options, "foo", nullptr);
+  s = txn3->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_OK(s);
  ASSERT_OK(txn1->Rollback());
@ -596,36 +637,42 @@ TEST_P(TransactionTest, SharedLocks) {
  ASSERT_OK(txn3->Rollback());
  // Test txn1 and txn2 sharing a lock and txn2 trying to upgrade lock.
-  s = txn1->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
  txn1->UndoGetForUpdate("foo");
-  s = txn2->GetForUpdate(read_options, "foo", nullptr);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_OK(s);
  ASSERT_OK(txn1->Rollback());
  ASSERT_OK(txn2->Rollback());
  // Test txn1 trying to downgrade its lock.
-  s = txn1->GetForUpdate(read_options, "foo", nullptr, true /* exclusive */);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         true /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
  // Should still fail after "downgrading".
-  s = txn1->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
@ -634,15 +681,17 @@ TEST_P(TransactionTest, SharedLocks) {
  // Test txn1 holding an exclusive lock and txn2 trying to obtain shared
  // access.
-  s = txn1->GetForUpdate(read_options, "foo", nullptr);
+  s = txn1->GetForUpdate(read_options, "foo", (std::string*)nullptr);
  ASSERT_OK(s);
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_TRUE(s.IsTimedOut());
  ASSERT_EQ(s.ToString(), "Operation timed out: Timeout waiting to lock key");
  txn1->UndoGetForUpdate("foo");
-  s = txn2->GetForUpdate(read_options, "foo", nullptr, false /* exclusive */);
+  s = txn2->GetForUpdate(read_options, "foo", (std::string*)nullptr,
                         false /* exclusive */);
  ASSERT_OK(s);
  delete txn1;
@ -676,8 +725,9 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  for (uint32_t i = 0; i < 31; i++) {
    txns[i] = db->BeginTransaction(write_options, txn_options);
    ASSERT_TRUE(txns[i]);
-    auto s = txns[i]->GetForUpdate(read_options, std::to_string((i + 1) / 2),
+    auto s =
-                                   nullptr, false /* exclusive */);
+        txns[i]->GetForUpdate(read_options, std::to_string((i + 1) / 2),
                              (std::string*)nullptr, false /* exclusive */);
    ASSERT_OK(s);
  }
@ -691,8 +741,9 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  std::vector<port::Thread> threads;
  for (uint32_t i = 0; i < 15; i++) {
    std::function<void()> blocking_thread = [&, i] {
-      auto s = txns[i]->GetForUpdate(read_options, std::to_string(i + 1),
+      auto s =
-                                     nullptr, true /* exclusive */);
+          txns[i]->GetForUpdate(read_options, std::to_string(i + 1),
                                (std::string*)nullptr, true /* exclusive */);
      ASSERT_OK(s);
      ASSERT_OK(txns[i]->Rollback());
      delete txns[i];
@ -710,8 +761,8 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  // Complete the cycle T[16 - 31] -> T1
  for (uint32_t i = 15; i < 31; i++) {
-    auto s =
+    auto s = txns[i]->GetForUpdate(read_options, "0", (std::string*)nullptr,
-        txns[i]->GetForUpdate(read_options, "0", nullptr, true /* exclusive */);
+                                   true /* exclusive */);
    ASSERT_TRUE(s.IsDeadlock());
    // Calculate next buffer len, plateau at 5 when 5 records are inserted.
@ -810,8 +861,8 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  for (uint32_t i = 0; i < 2; i++) {
    txns_shared[i] = db->BeginTransaction(write_options, txn_options);
    ASSERT_TRUE(txns_shared[i]);
-    auto s =
+    auto s = txns_shared[i]->GetForUpdate(read_options, std::to_string(i),
-        txns_shared[i]->GetForUpdate(read_options, std::to_string(i), nullptr);
+                                          (std::string*)nullptr);
    ASSERT_OK(s);
  }
@ -825,7 +876,7 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  for (uint32_t i = 0; i < 1; i++) {
    std::function<void()> blocking_thread = [&, i] {
      auto s = txns_shared[i]->GetForUpdate(read_options, std::to_string(i + 1),
-                                            nullptr);
+                                            (std::string*)nullptr);
      ASSERT_OK(s);
      ASSERT_OK(txns_shared[i]->Rollback());
      delete txns_shared[i];
@ -842,7 +893,8 @@ TEST_P(TransactionTest, DeadlockCycleShared) {
  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
  // Complete the cycle T2 -> T1 with a shared lock.
-  auto s = txns_shared[1]->GetForUpdate(read_options, "0", nullptr, false);
+  auto s = txns_shared[1]->GetForUpdate(read_options, "0",
                                        (std::string*)nullptr, false);
  ASSERT_TRUE(s.IsDeadlock());
  auto dlock_buffer = db->GetDeadlockInfoBuffer();
@ -884,7 +936,8 @@ TEST_P(TransactionStressTest, DeadlockCycle) {
    for (uint32_t i = 0; i < len; i++) {
      txns[i] = db->BeginTransaction(write_options, txn_options);
      ASSERT_TRUE(txns[i]);
-      auto s = txns[i]->GetForUpdate(read_options, std::to_string(i), nullptr);
+      auto s = txns[i]->GetForUpdate(read_options, std::to_string(i),
                                     (std::string*)nullptr);
      ASSERT_OK(s);
    }
@ -899,8 +952,8 @@ TEST_P(TransactionStressTest, DeadlockCycle) {
    std::vector<port::Thread> threads;
    for (uint32_t i = 0; i + 1 < len; i++) {
      std::function<void()> blocking_thread = [&, i] {
-        auto s =
+        auto s = txns[i]->GetForUpdate(read_options, std::to_string(i + 1),
-            txns[i]->GetForUpdate(read_options, std::to_string(i + 1), nullptr);
+                                       (std::string*)nullptr);
        ASSERT_OK(s);
        ASSERT_OK(txns[i]->Rollback());
        delete txns[i];
@ -917,7 +970,8 @@ TEST_P(TransactionStressTest, DeadlockCycle) {
    ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
    // Complete the cycle Tlen -> T1
-    auto s = txns[len - 1]->GetForUpdate(read_options, "0", nullptr);
+    auto s =
        txns[len - 1]->GetForUpdate(read_options, "0", (std::string*)nullptr);
    ASSERT_TRUE(s.IsDeadlock());
    const uint32_t dlock_buffer_size_ = (len - 1 > 5) ? 5 : (len - 1);
@ -1004,8 +1058,8 @@ TEST_P(TransactionStressTest, DeadlockStress) {
      // Lock keys in random order.
      for (const auto& k : random_keys) {
        // Lock mostly for shared access, but exclusive 1/4 of the time.
-        auto s =
+        auto s = txn->GetForUpdate(read_options, k, (std::string*)nullptr,
-            txn->GetForUpdate(read_options, k, nullptr, txn->GetID() % 4 == 0);
+                                   txn->GetID() % 4 == 0);
        if (!s.ok()) {
          ASSERT_TRUE(s.IsDeadlock());
          ASSERT_OK(txn->Rollback());
@ -3872,7 +3926,7 @@ TEST_P(TransactionTest, IteratorTest) {
    ASSERT_TRUE(iter->Valid());
    ASSERT_EQ(results[i], iter->value().ToString());
-    s = txn->GetForUpdate(read_options, iter->key(), nullptr);
+    s = txn->GetForUpdate(read_options, iter->key(), (std::string*)nullptr);
    if (i == 2) {
      // "C" was modified after txn's snapshot
      ASSERT_TRUE(s.IsBusy());
@ -4795,7 +4849,7 @@ TEST_P(TransactionTest, TimeoutTest) {
  txn_options0.lock_timeout = 50;  // txn timeout no longer infinite
  Transaction* txn1 = db->BeginTransaction(write_options, txn_options0);
-  s = txn1->GetForUpdate(read_options, "aaa", nullptr);
+  s = txn1->GetForUpdate(read_options, "aaa", (std::string*)nullptr);
  ASSERT_OK(s);
  // Conflicts with previous GetForUpdate.
@ -4832,7 +4886,7 @@ TEST_P(TransactionTest, TimeoutTest) {
  txn_options.expiration = 100;  // 100ms
  txn1 = db->BeginTransaction(write_options, txn_options);
-  s = txn1->GetForUpdate(read_options, "aaa", nullptr);
+  s = txn1->GetForUpdate(read_options, "aaa", (std::string*)nullptr);
  ASSERT_OK(s);
  // Conflicts with previous GetForUpdate.