diff --git a/java/org/rocksdb/BlockBasedTableConfig.java b/java/org/rocksdb/BlockBasedTableConfig.java
index 76e9302046..0798be2b87 100644
--- a/java/org/rocksdb/BlockBasedTableConfig.java
+++ b/java/org/rocksdb/BlockBasedTableConfig.java
@@ -26,6 +26,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
     blockCacheCompressedNumShardBits_ = 0;
     checksumType_ = ChecksumType.kCRC32c;
     indexType_ = IndexType.kBinarySearch;
+    formatVersion_ = 0;
   }
 
   /**
@@ -335,6 +336,46 @@ public class BlockBasedTableConfig extends TableFormatConfig {
     return indexType_;
   }
 
+  /**
+   * <p>For more details on BlockBasedTable's formats, see FORMAT-CHANGES.md
+   * We currently have three versions:</p>
+   *
+   * <ul>
+   * <li><strong>0</strong> - This version is currently written
+   * out by all RocksDB's versions by default. Can be read by really old
+   * RocksDB's. Doesn't support changing checksum (default is CRC32).</li>
+   * <li><strong>1</strong> - Can be read by RocksDB's versions since 3.0.
+   * Supports non-default checksum, like xxHash. It is written by RocksDB when
+   * BlockBasedTableOptions::checksum is something other than kCRC32c. (version
+   * 0 is silently upconverted)</li>
+   * <li><strong>2</strong> - Can be read by RocksDB's versions since 3.10.
+   * Changes the way we encode compressed blocks with LZ4, BZip2 and Zlib
+   * compression. If you don't plan to run RocksDB before version 3.10,
+   * you should probably use this.</li>
+   * </ul>
+   * <p> This option only affects newly written tables. When reading existing
+   * tables, the information about version is read from the footer.</p>
+   *
+   * @param formatVersion integer representing the version to be used.
+   * @return the reference to the current option.
+   */
+  public BlockBasedTableConfig setFormatVersion(int formatVersion) {
+    assert(formatVersion>=0 && formatVersion <=2);
+    formatVersion_ = formatVersion;
+    return this;
+  }
+
+  /**
+   *
+   * @return the currently configured format version.
+   * See also: {@link #setFormatVersion(int)}.
+   */
+  public int formatVersion() {
+    return formatVersion_;
+  }
+
+
+
   @Override protected long newTableFactoryHandle() {
     long filterHandle = 0;
     if (filter_ != null) {
@@ -347,7 +388,8 @@ public class BlockBasedTableConfig extends TableFormatConfig {
         filterHandle, cacheIndexAndFilterBlocks_,
         hashIndexAllowCollision_, blockCacheCompressedSize_,
         blockCacheCompressedNumShardBits_,
-        checksumType_.getValue(), indexType_.getValue());
+        checksumType_.getValue(), indexType_.getValue(),
+        formatVersion_);
   }
 
   private native long newTableFactoryHandle(
@@ -356,7 +398,7 @@ public class BlockBasedTableConfig extends TableFormatConfig {
       boolean wholeKeyFiltering, long filterPolicyHandle,
       boolean cacheIndexAndFilterBlocks, boolean hashIndexAllowCollision,
       long blockCacheCompressedSize, int blockCacheCompressedNumShardBits,
-      byte checkSumType, byte indexType);
+      byte checkSumType, byte indexType, int formatVersion);
 
   private boolean cacheIndexAndFilterBlocks_;
   private IndexType indexType_;
@@ -372,4 +414,5 @@ public class BlockBasedTableConfig extends TableFormatConfig {
   private int blockRestartInterval_;
   private Filter filter_;
   private boolean wholeKeyFiltering_;
+  private int formatVersion_;
 }
diff --git a/java/org/rocksdb/test/BlockBasedTableConfigTest.java b/java/org/rocksdb/test/BlockBasedTableConfigTest.java
index 5e0b96f29d..1172effc8e 100644
--- a/java/org/rocksdb/test/BlockBasedTableConfigTest.java
+++ b/java/org/rocksdb/test/BlockBasedTableConfigTest.java
@@ -162,4 +162,25 @@ public class BlockBasedTableConfigTest {
       }
     }
   }
+
+  @Test
+  public void blockBasedTableFormatVersion() {
+    BlockBasedTableConfig config = new BlockBasedTableConfig();
+    for (int version=0; version<=2; version++) {
+      config.setFormatVersion(version);
+      assertThat(config.formatVersion()).isEqualTo(version);
+    }
+  }
+
+  @Test(expected = AssertionError.class)
+  public void blockBasedTableFormatVersionFailNegative() {
+    BlockBasedTableConfig config = new BlockBasedTableConfig();
+    config.setFormatVersion(-1);
+  }
+
+  @Test(expected = AssertionError.class)
+  public void blockBasedTableFormatVersionFailIllegalVersion() {
+    BlockBasedTableConfig config = new BlockBasedTableConfig();
+    config.setFormatVersion(3);
+  }
 }
diff --git a/java/rocksjni/table.cc b/java/rocksjni/table.cc
index 1b576a7540..e78e7e0d7d 100644
--- a/java/rocksjni/table.cc
+++ b/java/rocksjni/table.cc
@@ -38,7 +38,7 @@ jlong Java_org_rocksdb_PlainTableConfig_newTableFactoryHandle(
 /*
  * Class:     org_rocksdb_BlockBasedTableConfig
  * Method:    newTableFactoryHandle
- * Signature: (ZJIJIIZIZZJIBB)J
+ * Signature: (ZJIJIIZIZZJIBBI)J
  */
 jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
     JNIEnv* env, jobject jobj, jboolean no_block_cache, jlong block_cache_size,
@@ -47,7 +47,7 @@ jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
     jlong jfilterPolicy, jboolean cache_index_and_filter_blocks,
     jboolean hash_index_allow_collision, jlong block_cache_compressed_size,
     jint block_cache_compressd_num_shard_bits, jbyte jchecksum_type,
-    jbyte jindex_type) {
+    jbyte jindex_type, jint jformat_version) {
   rocksdb::BlockBasedTableOptions options;
   options.no_block_cache = no_block_cache;
 
@@ -83,6 +83,7 @@ jlong Java_org_rocksdb_BlockBasedTableConfig_newTableFactoryHandle(
   options.checksum = static_cast<rocksdb::ChecksumType>(jchecksum_type);
   options.index_type = static_cast<
       rocksdb::BlockBasedTableOptions::IndexType>(jindex_type);
+  options.format_version = jformat_version;
 
   return reinterpret_cast<jlong>(rocksdb::NewBlockBasedTableFactory(options));
 }