@Override
  public void setConfig(
      final String storeName,
      final Options options,
      final Map<String, Object> configs) {
    if (!configured.get()) {
      throw new IllegalStateException(
          "Cannot use KsqlBoundedMemoryRocksDBConfigSetter before it's been configured.");
    }

    final BlockBasedTableConfig tableConfig = (BlockBasedTableConfig)options.tableFormatConfig();

    tableConfig.setBlockCache(cache);
    tableConfig.setCacheIndexAndFilterBlocks(true);
    options.setWriteBufferManager(writeBufferManager);
    options.setCompactionStyle(compactionStyle);
    options.setCompressionType(compressionType);
    if (maxNumConcurrentJobs != -1) {
      options.setMaxBackgroundJobs(maxNumConcurrentJobs);
    }
    if (compactionStyle.equals(CompactionStyle.UNIVERSAL)) {
      if (options.compactionOptionsUniversal() == null) {
        final CompactionOptionsUniversal compactionOptionsUniversal =
            new CompactionOptionsUniversal();
        compactionOptionsUniversal.setAllowTrivialMove(allowTrivialMove);
        options.setCompactionOptionsUniversal(compactionOptionsUniversal);
      } else {
        options.compactionOptionsUniversal().setAllowTrivialMove(allowTrivialMove);
      }
    }

    tableConfig.setCacheIndexAndFilterBlocksWithHighPriority(true);
    tableConfig.setPinTopLevelIndexAndFilter(true);

    options.setStatsDumpPeriodSec(0);

    options.setTableFormatConfig(tableConfig);
  }
@Test
  public void shouldFailWithoutConfigure() {
    // Expect:
    // When:
    final Exception e = assertThrows(
        IllegalStateException.class,
        () -> rocksDBConfig.setConfig("store_name", rocksOptions, emptyMap())
    );

    // Then:
    assertThat(e.getMessage(), containsString("Cannot use KsqlBoundedMemoryRocksDBConfigSetter before it's been configured."));
  }