Add Disable WAL support

Author: anujgoyal1

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/rocksdb/BoundedMemoryConfigSetter.java

@@ -8,6 +8,7 @@
 import static org.hypertrace.core.kafkastreams.framework.rocksdb.RocksDBConfigs.MAX_SIZE_AMPLIFICATION_PERCENT;
 import static org.hypertrace.core.kafkastreams.framework.rocksdb.RocksDBConfigs.OPTIMIZE_FOR_POINT_LOOKUPS;
 import static org.hypertrace.core.kafkastreams.framework.rocksdb.RocksDBConfigs.PERIODIC_COMPACTION_SECONDS;
+import static org.hypertrace.core.kafkastreams.framework.rocksdb.RocksDBConfigs.WAL_DISABLED;
 
 import java.util.Map;
 import org.apache.kafka.streams.state.RocksDBConfigSetter;
@@ -17,6 +18,7 @@
 import org.rocksdb.CompressionType;
 import org.rocksdb.InfoLogLevel;
 import org.rocksdb.Options;
+import org.rocksdb.WriteOptions;
 
 public class BoundedMemoryConfigSetter implements RocksDBConfigSetter {
 
@@ -52,6 +54,12 @@ public void setConfig(String storeName, Options options, Map<String, Object> con
       options.setUseDirectReads(Boolean.valueOf(String.valueOf(configs.get(DIRECT_READS_ENABLED))));
     }
 
+    if (configs.containsKey(WAL_DISABLED)) {
+      options.setWalDir("");
+      WriteOptions writeOptions = new WriteOptions();
+      writeOptions.setDisableWAL(true);
+    }
+
     if (configs.containsKey(OPTIMIZE_FOR_POINT_LOOKUPS)) {
       Boolean optimizeForPointLookups =
           Boolean.valueOf(String.valueOf(configs.get(OPTIMIZE_FOR_POINT_LOOKUPS)));

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/rocksdb/RocksDBConfigs.java

@@ -36,6 +36,7 @@ public class RocksDBConfigs {
   public static final String DIRECT_READS_ENABLED = rocksdbPrefix("direct.reads.enabled");
   public static final String OPTIMIZE_FOR_POINT_LOOKUPS = rocksdbPrefix("optimize.point.lookups");
   public static final String LOG_LEVEL_CONFIG = rocksdbPrefix("log.level");
+  public static final String WAL_DISABLED = rocksdbPrefix("optimize.point.lookups");
 
   public static String rocksdbPrefix(String configKey) {
     return ROCKS_DB_PREFIX + configKey;