revert: going back to kstream to older version (#36)

* revert: going back to ksteram to older version

* Revert "Fix RocksDB native memory leak in stateful applications (#32)"

This reverts commit e2e26cf.

Author: kotharironak

kafka-streams-framework/build.gradle.kts

@@ -12,15 +12,15 @@ tasks.test {
 dependencies {
     api(project(":kafka-streams-serdes"))
     api("com.typesafe:config:1.4.1")
-    api("org.apache.kafka:kafka-streams:6.1.0-ccs")
-    api("io.confluent:kafka-streams-avro-serde:6.1.0")
+    api("org.apache.kafka:kafka-streams:6.0.1-ccs")
+    api("io.confluent:kafka-streams-avro-serde:6.0.1")
 
     implementation("com.google.guava:guava:30.1-jre")
     implementation("org.hypertrace.core.serviceframework:platform-metrics:0.1.23")
     implementation("org.hypertrace.core.serviceframework:platform-service-framework:0.1.23")
-    implementation("org.apache.kafka:kafka-clients:6.1.0-ccs")
+    implementation("org.apache.kafka:kafka-clients:6.0.1-ccs")
 
-    testImplementation("org.apache.kafka:kafka-streams-test-utils:6.1.0-ccs")
+    testImplementation("org.apache.kafka:kafka-streams-test-utils:6.0.1-ccs")
     testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
     testImplementation("org.junit-pioneer:junit-pioneer:1.1.0")
     testImplementation("org.mockito:mockito-core:3.6.28")

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

@@ -13,7 +13,6 @@
 import com.google.common.annotations.VisibleForTesting;
 import java.util.Map;
 import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.streams.state.internals.BlockBasedTableConfigWithAccessibleCache;
 import org.rocksdb.BlockBasedTableConfig;
 import org.rocksdb.Cache;
 import org.rocksdb.LRUCache;
@@ -108,7 +107,7 @@ protected synchronized void initCache(Options options, Map<String, Object> confi
           cacheTotalCapacity / (1024 * 1024), writeBuffersRatio, highPriorityPoolRatio);
     }
 
-    final BlockBasedTableConfigWithAccessibleCache tableConfig = (BlockBasedTableConfigWithAccessibleCache) options.tableFormatConfig();
+    final BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();
 
     // ######### Block cache (Read buffers) #########
     if (configs.containsKey(BLOCK_SIZE)) {
@@ -134,12 +133,7 @@ protected synchronized void initCache(Options options, Map<String, Object> confi
 
     options.setWriteBufferManager(writeBufferManager);
 
-    final Cache oldCache = tableConfig.blockCache();
-    if(oldCache != null) {
-      LOG.info("Releasing old rocksdb cache before setting shared cache.");
-      oldCache.close();
-    }
-    tableConfig.setBlockCache(this.cache);
+    tableConfig.setBlockCache(cache);
     options.setTableFormatConfig(tableConfig);
   }
 

kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/rocksdb/BoundedMemoryConfigSetterTest.java

@@ -17,7 +17,6 @@
 import java.util.stream.Stream;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.streams.state.RocksDBConfigSetter;
-import org.apache.kafka.streams.state.internals.BlockBasedTableConfigWithAccessibleCache;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -27,7 +26,6 @@
 import org.rocksdb.CompactionStyle;
 import org.rocksdb.CompressionType;
 import org.rocksdb.InfoLogLevel;
-import org.rocksdb.LRUCache;
 import org.rocksdb.Options;
 
 class BoundedMemoryConfigSetterTest {
@@ -43,9 +41,7 @@ public void setUp() {
     RocksDBCacheProvider.get().testDestroy();
     options = new Options();
     configs = new HashMap<>();
-    tableConfig = new BlockBasedTableConfigWithAccessibleCache();
-    // mimic kafka streams
-    tableConfig.setBlockCache(new LRUCache(32 * 1024));
+    tableConfig = new BlockBasedTableConfig();
     options.setTableFormatConfig(tableConfig);
     configSetter = new BoundedMemoryConfigSetter();
   }

kafka-streams-serdes/build.gradle.kts

@@ -11,9 +11,9 @@ tasks.test {
 }
 
 dependencies {
-    api("org.apache.kafka:kafka-streams:6.1.0-ccs")
+    api("org.apache.kafka:kafka-streams:6.0.1-ccs")
     implementation("org.apache.avro:avro:1.10.2")
-    implementation("org.apache.kafka:kafka-clients:6.1.0-ccs")
+    implementation("org.apache.kafka:kafka-clients:6.0.1-ccs")
     testImplementation("org.junit.jupiter:junit-jupiter:5.7.0")
     constraints {
         api("com.fasterxml.jackson.core:jackson-databind:2.11.0") {