diff --git a/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java b/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java
index a6935b311e4a..43d5bc0ba5e4 100644
--- a/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java
+++ b/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java
@@ -29,6 +29,7 @@
 import com.github.luben.zstd.ZstdDictCompress;
 import com.github.luben.zstd.ZstdDictDecompress;
 import org.apache.cassandra.io.compress.ZstdCompressorBase;
+import org.apache.cassandra.io.compress.ZstdDictionaryCompressor;
 import org.apache.cassandra.utils.concurrent.Ref;
 import org.apache.cassandra.utils.concurrent.RefCounted;
 import org.apache.cassandra.utils.concurrent.SelfRefCounted;
@@ -113,7 +114,7 @@ public ZstdDictCompress dictionaryForCompression(int compressionLevel)
         if (closed.get())
             throw new IllegalStateException("Dictionary has been closed. " + dictId);
 
-        ZstdCompressorBase.validateCompressionLevel(compressionLevel);
+        ZstdCompressorBase.validateCompressionLevel(compressionLevel, ZstdDictionaryCompressor.BEST_COMPRESSION_LEVEL);
 
         return zstdDictCompressPerLevel.computeIfAbsent(compressionLevel, level -> {
             if (closed.get())
diff --git a/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java b/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java
index 9327de4c139c..f41f3c6f40cb 100644
--- a/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java
+++ b/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java
@@ -22,11 +22,14 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import com.github.luben.zstd.Zstd;
+
 /**
  * ZSTD Compressor
  */
 public class ZstdCompressor extends ZstdCompressorBase implements ICompressor
 {
+    public static final int BEST_COMPRESSION_LEVEL = Zstd.maxCompressionLevel();
     private static final ConcurrentHashMap<Integer, ZstdCompressor> instances = new ConcurrentHashMap<>();
 
     /**
@@ -39,7 +42,7 @@ public class ZstdCompressor extends ZstdCompressorBase implements ICompressor
     public static ZstdCompressor create(Map<String, String> options)
     {
         int level = getOrDefaultCompressionLevel(options);
-        validateCompressionLevel(level);
+        validateCompressionLevel(level, BEST_COMPRESSION_LEVEL);
         return getOrCreate(level);
     }
 
diff --git a/src/java/org/apache/cassandra/io/compress/ZstdCompressorBase.java b/src/java/org/apache/cassandra/io/compress/ZstdCompressorBase.java
index bbe278cbf43c..e6985668a9cb 100644
--- a/src/java/org/apache/cassandra/io/compress/ZstdCompressorBase.java
+++ b/src/java/org/apache/cassandra/io/compress/ZstdCompressorBase.java
@@ -34,7 +34,6 @@ public abstract class ZstdCompressorBase implements ICompressor
 {
     // These might change with the version of Zstd we're using
     public static final int FAST_COMPRESSION_LEVEL = Zstd.minCompressionLevel();
-    public static final int BEST_COMPRESSION_LEVEL = Zstd.maxCompressionLevel();
 
     // Compressor Defaults
     public static final int DEFAULT_COMPRESSION_LEVEL = 3;
@@ -168,9 +167,9 @@ public void compress(ByteBuffer input, ByteBuffer output) throws IOException
      *
      * @param level compression level
      */
-    public static void validateCompressionLevel(int level)
+    public static void validateCompressionLevel(int level, int bestCompressionLevel)
     {
-        if (level < FAST_COMPRESSION_LEVEL || level > BEST_COMPRESSION_LEVEL)
+        if (level < FAST_COMPRESSION_LEVEL || level > bestCompressionLevel)
         {
             throw new IllegalArgumentException(String.format("%s=%d is invalid", COMPRESSION_LEVEL_OPTION_NAME, level));
         }
diff --git a/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java b/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java
index 3ba5841aaa54..da8d582a406e 100644
--- a/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java
+++ b/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java
@@ -41,6 +41,9 @@
 
 public class ZstdDictionaryCompressor extends ZstdCompressorBase implements ICompressor, IDictionaryCompressor<ZstdCompressionDictionary>
 {
+    // see CASSANDRA-21021
+    public static final int BEST_COMPRESSION_LEVEL = 9;
+
     private static final ConcurrentHashMap<Integer, ZstdDictionaryCompressor> instancesPerLevel = new ConcurrentHashMap<>();
     private static final Cache<ZstdCompressionDictionary, ZstdDictionaryCompressor> instancePerDict =
     Caffeine.newBuilder()
@@ -74,7 +77,7 @@ public class ZstdDictionaryCompressor extends ZstdCompressorBase implements ICom
     public static ZstdDictionaryCompressor create(Map<String, String> options)
     {
         int level = getOrDefaultCompressionLevel(options);
-        validateCompressionLevel(level);
+        validateCompressionLevel(level, BEST_COMPRESSION_LEVEL);
         return getOrCreate(level, null);
     }
 
diff --git a/test/unit/org/apache/cassandra/io/compress/ZstdDictionaryCompressorTest.java b/test/unit/org/apache/cassandra/io/compress/ZstdDictionaryCompressorTest.java
index ef5e3d11d8d7..0afeda9db972 100644
--- a/test/unit/org/apache/cassandra/io/compress/ZstdDictionaryCompressorTest.java
+++ b/test/unit/org/apache/cassandra/io/compress/ZstdDictionaryCompressorTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.cassandra.io.compress;
 
-import com.github.luben.zstd.Zstd;
 import com.github.luben.zstd.ZstdDictTrainer;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.compression.ZstdCompressionDictionary;
@@ -104,7 +103,7 @@ public void testCreateWithDictionary()
     @Test
     public void testCreateWithInvalidCompressionLevel()
     {
-        String invalidLevel = String.valueOf(Zstd.maxCompressionLevel() + 1);
+        String invalidLevel = String.valueOf(ZstdDictionaryCompressor.BEST_COMPRESSION_LEVEL + 1);
         Map<String, String> options = Map.of(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME, invalidLevel);
 
         assertThatThrownBy(() -> ZstdDictionaryCompressor.create(options))
@@ -112,6 +111,14 @@ public void testCreateWithInvalidCompressionLevel()
         .hasMessage(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME + '=' + invalidLevel + " is invalid");
     }
 
+    @Test
+    public void testCreateWithMaxCompressionLevel()
+    {
+        String invalidLevel = String.valueOf(ZstdDictionaryCompressor.BEST_COMPRESSION_LEVEL);
+        Map<String, String> options = Map.of(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME, invalidLevel);
+        ZstdDictionaryCompressor.create(options);
+    }
+
     @Test
     public void testCompressDecompressWithDictionary() throws IOException
     {