Persistent Cache Implementation using RocksDB #198
+434
−69
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…ementation, and refactor `InMemoryCache` to implement interface. (cherry picked from commit 14e284a)
(cherry picked from commit 3fb06b5)
…e connector (cherry picked from commit 6a61963)
…ching (cherry picked from commit 75d91f7)
(cherry picked from commit b1a617f)
(cherry picked from commit dc72d73)
(cherry picked from commit 1d1a1f7)
…ksDB (cherry picked from commit 618c993)
(cherry picked from commit e3d7973)
(cherry picked from commit 9788b47)
(cherry picked from commit b86be91)
15 tasks
srinath-ctds
approved these changes
Jul 29, 2025
sandeep-ctds
requested changes
Jul 29, 2025
| @@ -33,7 +33,7 @@ jobs: | |||
| needs: build | |||
| name: Test | |||
| runs-on: ubuntu-latest | |||
| timeout-minutes: 120 | |||
| timeout-minutes: 360 | |||
There was a problem hiding this comment.
360 minutes is too high. Why don't we make it 180.
cc: @srinath-ctds
| @@ -24,6 +24,7 @@ kafkaVersion=3.4.0 | |||
| vavrVersion=0.10.3 | |||
| testContainersVersion=1.19.1 | |||
| caffeineVersion=2.8.8 | |||
| rocksdbVersion=9.4.0 | |||
There was a problem hiding this comment.
Just for Info, Why specifically 9.4.0?
| @@ -34,7 +34,7 @@ public class ChaosNetworkContainer<SELF extends ChaosNetworkContainer<SELF>> ext | |||
|
|
|||
| public ChaosNetworkContainer(String targetContainer, String pause) { | |||
| super(PUMBA_IMAGE); | |||
| setCommand("--log-level debug netem --tc-image gaiadocker/iproute2 --duration " + pause + " loss --percent 100 " + targetContainer); | |||
| setCommand("--log-level debug netem --tc-image ghcr.io/alexei-led/pumba-debian-nettools --duration " + pause + " loss --percent 100 " + targetContainer); | |||
There was a problem hiding this comment.
Please add why this change is made in the PR description.
| @@ -66,6 +66,8 @@ public class CassandraSourceConnectorConfig { | |||
| public static final String CACHE_MAX_DIGESTS_CONFIG = "cache.max.digest"; | |||
| public static final String CACHE_MAX_CAPACITY_CONFIG = "cache.max.capacity"; | |||
| public static final String CACHE_EXPIRE_AFTER_MS_CONFIG = "cache.expire.after.ms"; | |||
| public static final String CACHE_PERSISTENT_DIRECTORY_CONFIG = "cache.persistent.directory"; | |||
|
|
|||
There was a problem hiding this comment.
Is the extra line necessary for formatting checks? If not, please remove
|
|
||
| import java.util.List; | ||
|
|
||
| public interface MutationCache<K> { |
There was a problem hiding this comment.
Interface level description on what this cache will be used for. (de-duplication)
| * Keep MD5 digests to deduplicate Cassandra mutations | ||
| */ | ||
| public class MutationCache<K> { | ||
|
|
There was a problem hiding this comment.
Add @Override annotation to the methods getting implemented from the interface in both the classes.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Types of changes
Description
The Pull request implements Persistent Cache with RocksDB.
The implementation uses an in-memory cache mechanism using Caffeine and a Write-through persistent cache based on RocksDB.
Fixes
cache.max.capacity) being "processed" in a very short period of time, this can occur on high loads on the in-memory cache, causing eviction and duplication in the data topic.What's added/changed
PersistentCache.javawhich implements theMutationCache.javainterface, having the persistent cache implementation.cache.persistent.directoryto enable the persistent cache and set the cache directory at the same time.MutationCache.javatoInMemory.java, which has a non-persistent in-memory cache implementation.Checklist: