Address Stefan's review comments

Get rid of java.util.concurrent.CompletableFuture usage

Use clock

Relocate dictionary configurations and use IntKibibytesBound

Change compression_dictionary_training_sampling_rate data type from int to float

Address other various comments

Author: yifan-c

conf/cassandra.yaml

@@ -617,54 +617,6 @@ counter_cache_save_period: 7200s
 # Disabled by default, meaning all keys are going to be saved
 # counter_cache_keys_to_save: 100
 
-# Dictionary compression settings for ZSTD dictionary-based compression
-# These settings control the automatic training and caching of compression dictionaries
-# for tables that use ZSTD dictionary compression.
-
-# How often to refresh compression dictionaries across the cluster.
-# During refresh, nodes will check for newer dictionary versions and update their caches.
-# Min unit: s
-compression_dictionary_refresh_interval: 3600s
-
-# Initial delay before starting the first dictionary refresh cycle after node startup.
-# This prevents all nodes from refreshing simultaneously when the cluster starts.
-# Min unit: s
-compression_dictionary_refresh_initial_delay: 10s
-
-# Maximum number of compression dictionaries to cache per table.
-# Each table using dictionary compression can have multiple dictionaries cached
-# (current version plus recently used versions for reading older SSTables).
-compression_dictionary_cache_size: 10
-
-# How long to keep compression dictionaries in the cache before they expire.
-# Expired dictionaries will be removed from memory but can be reloaded if needed.
-# Min unit: s
-compression_dictionary_cache_expire: 3600s
-
-# Dictionary training configuration (advanced settings)
-# These settings control how compression dictionaries are trained from sample data.
-
-# Maximum size of a trained compression dictionary in bytes.
-# Larger dictionaries may provide better compression but use more memory.
-# Min unit: B
-compression_dictionary_training_max_dictionary_size: 65536
-
-# Maximum total size of sample data to collect for dictionary training.
-# More sample data generally produces better dictionaries but takes longer to train.
-# The recommended sample size is 100x the dictionary size.
-# Min unit: B
-compression_dictionary_training_max_total_sample_size: 10485760
-
-# Enable automatic dictionary training based on sampling of write operations.
-# When enabled, the system will automatically collect samples and train new dictionaries.
-# Manual training via nodetool is always available regardless of this setting.
-compression_dictionary_training_auto_train_enabled: false
-
-# Sampling rate for automatic dictionary training (1-10000).
-# Value of 100 means 1% of writes are sampled. Lower values reduce overhead but may
-# result in less representative sample data for dictionary training.
-compression_dictionary_training_sampling_rate: 100
-
 # saved caches
 # If not set, the default directory is $CASSANDRA_HOME/data/saved_caches.
 # saved_caches_directory: /var/lib/cassandra/saved_caches
@@ -2897,3 +2849,49 @@ storage_compatibility_mode: NONE
 #         # especially in keyspaces with many tables. The splitter avoids batching tables together if they
 #         # exceed other configuration parameters like bytes_per_assignment or partitions_per_assignment.
 #         max_tables_per_assignment: 64
+
+# Dictionary compression settings for ZSTD dictionary-based compression
+# These settings control the automatic training and caching of compression dictionaries
+# for tables that use ZSTD dictionary compression.
+
+# How often to refresh compression dictionaries across the cluster.
+# During refresh, nodes will check for newer dictionary versions and update their caches.
+# Min unit: s
+compression_dictionary_refresh_interval: 3600s
+
+# Initial delay before starting the first dictionary refresh cycle after node startup.
+# This prevents all nodes from refreshing simultaneously when the cluster starts.
+# Min unit: s
+compression_dictionary_refresh_initial_delay: 10s
+
+# Maximum number of compression dictionaries to cache per table.
+# Each table using dictionary compression can have multiple dictionaries cached
+# (current version plus recently used versions for reading older SSTables).
+compression_dictionary_cache_size: 10
+
+# How long to keep compression dictionaries in the cache before they expire.
+# Expired dictionaries will be removed from memory but can be reloaded if needed.
+# Min unit: s
+compression_dictionary_cache_expire: 24h
+
+# Dictionary training configuration (advanced settings)
+# These settings control how compression dictionaries are trained from sample data.
+
+# Maximum size of a trained compression dictionary.
+# Larger dictionaries may provide better compression but use more memory.
+compression_dictionary_training_max_dictionary_size: 64KiB
+
+# Maximum total size of sample data to collect for dictionary training.
+# More sample data generally produces better dictionaries but takes longer to train.
+# The recommended sample size is 100x the dictionary size.
+compression_dictionary_training_max_total_sample_size: 10MiB
+
+# Enable automatic dictionary training based on sampling of write operations.
+# When enabled, the system will automatically collect samples and train new dictionaries.
+# Manual training via nodetool is always available regardless of this setting.
+compression_dictionary_training_auto_train_enabled: false
+
+# Sampling rate for automatic dictionary training (1-10000).
+# Value of 100 means 1% of writes are sampled. Lower values reduce overhead but may
+# result in less representative sample data for dictionary training.
+compression_dictionary_training_sampling_rate: 0.01

conf/cassandra_latest.yaml

@@ -2529,3 +2529,49 @@ storage_compatibility_mode: NONE
 #         # especially in keyspaces with many tables. The splitter avoids batching tables together if they
 #         # exceed other configuration parameters like bytes_per_assignment or partitions_per_assignment.
 #         max_tables_per_assignment: 64
+
+# Dictionary compression settings for ZSTD dictionary-based compression
+# These settings control the automatic training and caching of compression dictionaries
+# for tables that use ZSTD dictionary compression.
+
+# How often to refresh compression dictionaries across the cluster.
+# During refresh, nodes will check for newer dictionary versions and update their caches.
+# Min unit: s
+compression_dictionary_refresh_interval: 3600s
+
+# Initial delay before starting the first dictionary refresh cycle after node startup.
+# This prevents all nodes from refreshing simultaneously when the cluster starts.
+# Min unit: s
+compression_dictionary_refresh_initial_delay: 10s
+
+# Maximum number of compression dictionaries to cache per table.
+# Each table using dictionary compression can have multiple dictionaries cached
+# (current version plus recently used versions for reading older SSTables).
+compression_dictionary_cache_size: 10
+
+# How long to keep compression dictionaries in the cache before they expire.
+# Expired dictionaries will be removed from memory but can be reloaded if needed.
+# Min unit: s
+compression_dictionary_cache_expire: 24h
+
+# Dictionary training configuration (advanced settings)
+# These settings control how compression dictionaries are trained from sample data.
+
+# Maximum size of a trained compression dictionary.
+# Larger dictionaries may provide better compression but use more memory.
+compression_dictionary_training_max_dictionary_size: 64KiB
+
+# Maximum total size of sample data to collect for dictionary training.
+# More sample data generally produces better dictionaries but takes longer to train.
+# The recommended sample size is 100x the dictionary size.
+compression_dictionary_training_max_total_sample_size: 10MiB
+
+# Enable automatic dictionary training based on sampling of write operations.
+# When enabled, the system will automatically collect samples and train new dictionaries.
+# Manual training via nodetool is always available regardless of this setting.
+compression_dictionary_training_auto_train_enabled: false
+
+# Sampling rate for automatic dictionary training (1-10000).
+# Value of 100 means 1% of writes are sampled. Lower values reduce overhead but may
+# result in less representative sample data for dictionary training.
+compression_dictionary_training_sampling_rate: 0.01

src/java/org/apache/cassandra/config/Config.java

@@ -517,13 +517,13 @@ public static class SSTableConfig
     public volatile DurationSpec.IntSecondsBound compression_dictionary_refresh_interval = new DurationSpec.IntSecondsBound("3600s"); // 1 hour - TODO: re-assess whether daily (86400s) is more appropriate
     public volatile DurationSpec.IntSecondsBound compression_dictionary_refresh_initial_delay = new DurationSpec.IntSecondsBound("10s"); // 10 seconds default
     public volatile int compression_dictionary_cache_size = 10; // max dictionaries per table
-    public volatile DurationSpec.IntSecondsBound compression_dictionary_cache_expire = new DurationSpec.IntSecondsBound("3600s"); // 1 hour default
+    public volatile DurationSpec.IntSecondsBound compression_dictionary_cache_expire = new DurationSpec.IntSecondsBound("24h");
 
     // Dictionary training settings
-    public volatile int compression_dictionary_training_max_dictionary_size = 65536; // 64KB
-    public volatile int compression_dictionary_training_max_total_sample_size = 10485760; // 10MB total
+    public volatile DataStorageSpec.IntKibibytesBound compression_dictionary_training_max_dictionary_size = new DataStorageSpec.IntKibibytesBound("64KiB");
+    public volatile DataStorageSpec.IntKibibytesBound compression_dictionary_training_max_total_sample_size = new DataStorageSpec.IntKibibytesBound("10MiB");
     public volatile boolean compression_dictionary_training_auto_train_enabled = false;
-    public volatile int compression_dictionary_training_sampling_rate = 100; // samples 1%; using int since random.nextInt is generally faster than random.nextDouble
+    public volatile float compression_dictionary_training_sampling_rate = 0.01f; // samples 1%
 
     public DataStorageSpec.LongMebibytesBound paxos_cache_size = null;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -4383,12 +4383,12 @@ public static int getCompressionDictionaryCacheExpireSeconds()
 
     public static int getCompressionDictionaryTrainingMaxDictionarySize()
     {
-        return conf.compression_dictionary_training_max_dictionary_size;
+        return conf.compression_dictionary_training_max_dictionary_size.toBytes();
     }
 
     public static int getCompressionDictionaryTrainingMaxTotalSampleSize()
     {
-        return conf.compression_dictionary_training_max_total_sample_size;
+        return conf.compression_dictionary_training_max_total_sample_size.toBytes();
     }
 
     public static boolean getCompressionDictionaryTrainingAutoTrainEnabled()
@@ -4397,7 +4397,7 @@ public static boolean getCompressionDictionaryTrainingAutoTrainEnabled()
     }
 
 
-    public static int getCompressionDictionaryTrainingSamplingRate()
+    public static float getCompressionDictionaryTrainingSamplingRate()
     {
         return conf.compression_dictionary_training_sampling_rate;
     }

src/java/org/apache/cassandra/db/compression/CompressionDictionary.java

@@ -40,7 +40,7 @@ public interface CompressionDictionary extends AutoCloseable
      *
      * @return dictionary id
      */
-    DictId identifier();
+    DictId dictId();
 
     /**
      * Get the raw bytes of the compression dictionary
@@ -56,7 +56,7 @@ public interface CompressionDictionary extends AutoCloseable
      */
     default Kind kind()
     {
-        return identifier().kind;
+        return dictId().kind;
     }
 
     /**
@@ -67,7 +67,7 @@ default Kind kind()
      */
     default void serialize(DataOutput out) throws IOException
     {
-        DictId dictId = identifier();
+        DictId dictId = dictId();
         int ordinal = dictId.kind.ordinal();
         out.writeByte(ordinal);
         out.writeLong(dictId.id);
@@ -123,7 +123,8 @@ static CompressionDictionary deserialize(DataInput input, @Nullable CompressionD
         int checksum = input.readInt();
         int calculatedChecksum = calculateChecksum((byte) kindOrdinal, id, dict);
         if (checksum != calculatedChecksum)
-            throw new IOException("Compression dictionary checksum does not match");
+            throw new IOException("Compression dictionary checksum does not match. " +
+                                  "Expected: " + checksum + "; actual: " + calculatedChecksum);
 
         CompressionDictionary dictionary = kind.createDictionary(dictId, dict);
 

src/java/org/apache/cassandra/db/compression/CompressionDictionaryCache.java

@@ -47,10 +47,9 @@ public class CompressionDictionaryCache implements ICompressionDictionaryCache
 
     public CompressionDictionaryCache()
     {
-        Duration expiryTime = Duration.ofSeconds(DatabaseDescriptor.getCompressionDictionaryCacheExpireSeconds());
         this.cache = Caffeine.newBuilder()
                              .maximumSize(DatabaseDescriptor.getCompressionDictionaryCacheSize())
-                             .expireAfterAccess(expiryTime)
+                             .expireAfterAccess(Duration.ofSeconds(DatabaseDescriptor.getCompressionDictionaryCacheExpireSeconds()))
                              .removalListener((CompressionDictionary.DictId dictId,
                                                CompressionDictionary dictionary,
                                                RemovalCause cause) -> {
@@ -88,7 +87,7 @@ public CompressionDictionary get(CompressionDictionary.DictId dictId)
     @Override
     public void add(CompressionDictionary compressionDictionary)
     {
-        cache.put(compressionDictionary.identifier(), compressionDictionary);
+        cache.put(compressionDictionary.dictId(), compressionDictionary);
     }
 
     @Override
@@ -100,7 +99,7 @@ public void setCurrentIfNewer(@Nullable CompressionDictionary dictionary)
         add(dictionary);
         // Only update the current dictionary if we don't have one or the new one has a higher ID (newer)
         CompressionDictionary current = currentDictionary.get();
-        while ((current == null || dictionary.identifier().id > current.identifier().id)
+        while ((current == null || dictionary.dictId().id > current.dictId().id)
                && !currentDictionary.compareAndSet(current, dictionary))
         {
             current = currentDictionary.get();

src/java/org/apache/cassandra/db/compression/CompressionDictionaryEventHandler.java

@@ -31,7 +31,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
 
 /**
  * Handles compression dictionary events including training completion and cluster notifications.
@@ -79,7 +78,7 @@ public void onNewDictionaryTrained(CompressionDictionary.DictId dictionaryId)
     public void onNewDictionaryAvailable(CompressionDictionary.DictId dictionaryId)
     {
         // Best effort to retrieve the dictionary; otherwise, the periodic task should retrieve the dictionary later
-        CompletableFuture.runAsync(() -> {
+        ScheduledExecutors.nonPeriodicTasks.submit(() -> {
             try
             {
                 if (!cfs.metadata().params.compression.isDictionaryCompressionEnabled())
@@ -95,7 +94,7 @@ public void onNewDictionaryAvailable(CompressionDictionary.DictId dictionaryId)
                 logger.warn("Failed to retrieve compression dictionary for {}.{}. {}",
                             keyspaceName, tableName, dictionaryId, e);
             }
-        }, ScheduledExecutors.nonPeriodicTasks);
+        });
     }
 
     // Best effort to notify the peer regarding the new dictionary being available to pull.

src/java/org/apache/cassandra/db/compression/CompressionDictionaryManager.java

@@ -83,8 +83,7 @@ public CompressionDictionaryManager(ColumnFamilyStore columnFamilyStore, boolean
 
     static String mbeanName(String keyspaceName, String tableName)
     {
-        return "org.apache.cassandra.db.compression:type=CompressionDictionaryManager" +
-               ",keyspace=" + keyspaceName + ",table=" + tableName;
+        return MBEAN_NAME + ",keyspace=" + keyspaceName + ",table=" + tableName;
     }
 
     public boolean isEnabled()
@@ -209,7 +208,7 @@ public synchronized void train(Map<String, String> options)
         // Validate table supports dictionary compression
         if (!isEnabled)
         {
-            throw new IllegalArgumentException("Table " + keyspaceName + '.' + tableName + " does not support dictionary compression");
+            throw new UnsupportedOperationException("Table " + keyspaceName + '.' + tableName + " does not support dictionary compression");
         }
 
         if (trainer == null)
@@ -266,7 +265,7 @@ private void handleNewDictionary(CompressionDictionary dictionary)
     {
         // sequence meatters; persist the new dictionary before broadcasting to others.
         storeDictionary(dictionary);
-        onNewDictionaryTrained(dictionary.identifier());
+        onNewDictionaryTrained(dictionary.dictId());
     }
 
     private CompressionDictionaryTrainingConfig createTrainingConfig()

src/java/org/apache/cassandra/db/compression/CompressionDictionaryManagerMBean.java

@@ -22,11 +22,13 @@
 
 public interface CompressionDictionaryManagerMBean
 {
+    String MBEAN_NAME = "org.apache.cassandra.db.compression:type=CompressionDictionaryManager";
+
     /**
      * Starts sampling and training for this table.
      * 
      * @param options options for the training process (currently unused, reserved for future extensions)
-     * @throws IllegalArgumentException if table doesn't support dictionary compression
+     * @throws UnsupportedOperationException if table doesn't support dictionary compression
      * @throws IllegalStateException if training is already in progress for this table
      */
     void train(Map<String, String> options);

src/java/org/apache/cassandra/db/compression/CompressionDictionaryScheduler.java

@@ -29,6 +29,7 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.compression.ICompressionDictionaryTrainer.TrainingStatus;
 import org.apache.cassandra.schema.SystemDistributedKeyspace;
+import org.apache.cassandra.utils.Clock;
 
 /**
  * Manages scheduled tasks for compression dictionary operations.
@@ -90,7 +91,7 @@ public void scheduleManualTraining(ManualTrainingOptions options, ICompressionDi
         logger.info("Starting manual dictionary training for {}.{} with max sampling duration: {} seconds",
                     keyspaceName, tableName, maxSamplingDurationSeconds);
 
-        long deadlineMillis = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(maxSamplingDurationSeconds);
+        long deadlineMillis = Clock.Global.currentTimeMillis() + TimeUnit.SECONDS.toMillis(maxSamplingDurationSeconds);
 
         ManualTrainingTask task = new ManualTrainingTask(deadlineMillis, trainer);
 
@@ -182,15 +183,15 @@ public void run()
                 return;
             }
 
-            long now = System.currentTimeMillis();
+            long now = Clock.Global.currentTimeMillis();
             // Force training if there are not enough samples, but we have hit the max sampling duration
             boolean reachedDeadline = now >= deadlineMillis;
             if (!isTraining && (trainer.isReady() || reachedDeadline))
             {
                 // Set isTraining to only enter the branch once
                 isTraining = true;
                 trainer.trainDictionaryAsync(reachedDeadline)
-                       .whenComplete((dictionary, throwable) -> {
+                       .addCallback((dictionary, throwable) -> {
                            cancelManualTraining();
                            if (throwable != null)
                            {