public static void maybeConfigureRocksDBConfigSetter(final KsqlConfig ksqlConfig) {
    final Map<String, Object> streamsProps = ksqlConfig.getKsqlStreamConfigProps();
    final Class<?> clazz =
        (Class) streamsProps.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG);

    if (clazz != null && org.apache.kafka.common.Configurable.class.isAssignableFrom(clazz)) {
      try {
        ((org.apache.kafka.common.Configurable) Utils.newInstance(clazz))
            .configure(ksqlConfig.originals());
      } catch (final Exception e) {
        throw new ConfigException(
            "Failed to configure Configurable RocksDBConfigSetter. "
                + StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG + ": " + clazz.getName(),
            e);
      }
    }
  }
@Test
  public void shouldStartWithNonConfigurableRocksDBConfigSetter() throws Exception {
    // Given:
    when(ksqlConfig.getKsqlStreamConfigProps()).thenReturn(
        ImmutableMap.of(
            StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG,
            Class.forName("io.confluent.ksql.rest.util.RocksDBConfigSetterHandlerTest$NonConfigurableTestRocksDBConfigSetter"))
    );

    // No error when:
    RocksDBConfigSetterHandler.maybeConfigureRocksDBConfigSetter(ksqlConfig);
  }