NoBlockCacheRocksDbOptionsFactory.scala

import org.apache.flink.contrib.streaming.state.{ConfigurableRocksDBOptionsFactory, RocksDBOptionsFactory}
import org.rocksdb.DBOptions
import org.rocksdb.ColumnFamilyOptions
import org.rocksdb.BlockBasedTableConfig
import org.apache.flink.configuration.ReadableConfig
import java.util

class NoBlockCacheRocksDbOptionsFactory extends ConfigurableRocksDBOptionsFactory {
  override def createDBOptions(currentOptions: DBOptions, handlesToClose: util.Collection[AutoCloseable]): DBOptions = {
    currentOptions
  }

  override def createColumnOptions(
    currentOptions: ColumnFamilyOptions,
    handlesToClose: util.Collection[AutoCloseable]
  ): ColumnFamilyOptions = {
    val blockBasedTableConfig = new BlockBasedTableConfig()
    blockBasedTableConfig.setNoBlockCache(true)
    // Needed in order to disable block cache
    blockBasedTableConfig.setCacheIndexAndFilterBlocks(false)
    blockBasedTableConfig.setCacheIndexAndFilterBlocksWithHighPriority(false)
    blockBasedTableConfig.setPinL0FilterAndIndexBlocksInCache(false)
    currentOptions.setTableFormatConfig(blockBasedTableConfig)
    currentOptions
  }

  override def configure(configuration: ReadableConfig): RocksDBOptionsFactory = {
    this
  }
}