Implementation of CEP-49: Hardware-accelerated compression

The implementation of this CEP adds the ability to plug-in custom compressors providers
which might delegate the de/compression to a specialized hardware. An operator does not
need to change any schema definition, the mere implementation of a compressor provider
put on a class path and its related configuration in cassandra.yaml will transparently
start to use appropriate compressor, e.g. backed by a specialized hardware.

patch by Shylaja Kokoori; reviewed by Joey Lynch, Stefan Miklosovic for CASSANDRA-20975

Co-authored-by: Stefan Miklosovic <smiklosovic@apache.org>

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975)
  * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394)
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)

NEWS.txt

@@ -80,6 +80,10 @@ using the provided 'sstableupgrade' tool.
 New features
 ------------
 
+    - CEP-49 - it is possible to implement custom compression providers, e.g.
+      integrating with a specialized hardware which makes de/compression faster.
+      See CASSANDRA-20975 for more information.
+
 Upgrading
 ---------
 

conf/cassandra.yaml

@@ -2999,3 +2999,21 @@ compression_dictionary_cache_size: 10
 # Expired dictionaries will be removed from memory but can be reloaded if needed.
 # Min unit: s
 compression_dictionary_cache_expire: 24h
+
+# Configuring pluggable compressor providers.
+# Keys of compressor_providers map to implementations of ICompressor. FQCN are allowed to be keys as well.
+# Cassandra will attempt to load the provider with the specified class_name for each compressor key.
+# You can only override built-in compressors.
+# Parameters:
+#   - fail_on_missing_provider: If "false", falls back to DefaultCompressionProvider
+#     when the plugin fails to load. If "true", throws an exception on failure.
+#     The default value is "false".
+# If this section is commented out, Cassandra uses DefaultCompressionProvider
+# with built-in compressor implementation.
+#compressor_providers:
+#  LZ4Compressor:
+#    class_name: org.apache.cassandra.io.compress.DefaultCompressionProvider
+#    parameters:
+#      fail_on_missing_provider: "false"
+#  org.apache.cassandra.io.compress.ZstdCompressor:
+#    class_name: org.apache.cassandra.io.compress.DefaultCompressionProvider

conf/cassandra_latest.yaml

@@ -2742,3 +2742,21 @@ compression_dictionary_cache_size: 10
 # Expired dictionaries will be removed from memory but can be reloaded if needed.
 # Min unit: s
 compression_dictionary_cache_expire: 24h
+
+# Configuring pluggable compressor providers.
+# Keys of compressor_providers map to implementations of ICompressor. FQCN are allowed to be keys as well.
+# Cassandra will attempt to load the provider with the specified class_name for each compressor key.
+# You can only override built-in compressors.
+# Parameters:
+#   - fail_on_missing_provider: If "false", falls back to DefaultCompressionProvider
+#     when the plugin fails to load. If "true", throws an exception on failure.
+#     The default value is "false".
+# If this section is commented out, Cassandra uses DefaultCompressionProvider
+# with built-in compressor implementation.
+#compressor_providers:
+#  LZ4Compressor:
+#    class_name: org.apache.cassandra.io.compress.DefaultCompressionProvider
+#    parameters:
+#      fail_on_missing_provider: "false"
+#  org.apache.cassandra.io.compress.ZstdCompressor:
+#    class_name: org.apache.cassandra.io.compress.DefaultCompressionProvider

doc/modules/cassandra/pages/managing/operating/compression.adoc

@@ -416,6 +416,53 @@ Importing a dictionary to a table from a file should happen only against one nod
 dictionary will be eventually stored in `system_distributed.compression_dictionaries` table and reused
 cluster-wide. When imports happen from multiple nodes, the highest-version dictionary will be used.
 
+== Plugin compression providers
+
+Pluggable compression providers allow replacing the built-in compressor implementation
+with a custom one — for example, to delegate compression to a hardware-accelerated library.
+
+Providers are configured in `cassandra.yaml` under the `compressor_providers` map.
+Each key identifies a built-in compressor by its simple class name (e.g. `LZ4Compressor`) its FQCN, or
+its registered abbreviation (e.g. `lz4`). Only built-in compressors can be overridden; new compressor types cannot be introduced this way.
+
+Example:
+[source,yaml]
+----
+compressor_providers:
+  LZ4Compressor:
+    class_name: com.example.MyLZ4Provider
+    parameters:
+      fail_on_missing_provider: "true"
+  org.apache.cassandra.io.compress.ZstdCompressor:
+    class_name: com.example.MyZstdProvider
+----
+
+Each provider entry requires:
+
+* `class_name`: Fully qualified class name of an `AbstractCompressionProvider` subclass.
+* `parameters` (optional): A string map passed to the provider's `init()` method.
+  The reserved key `fail_on_missing_provider` (default `"false"`) controls fallback behaviour:
+     - If `"false"`: when initialization, health checks, or compressor creation fails,
+       falls back to `DefaultCompressionProvider`
+     - If `"true"`: throws an exception instead of falling back
+
+If `compressor_providers` is omitted entirely, `DefaultCompressionProvider` is used
+for all compressor types, which is the same behaviour as previous Cassandra releases.
+
+==== Implementing a custom provider
+
+A custom provider must extend `AbstractCompressionProvider`, implement the following methods and add the jar
+to the classpath, before node startup:
+
+* `boolean isHealthy()` — returns `true` if the provider is ready.
+* `ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options)` — factory
+  that returns a compressor instance configured with the given options.
+
+The `ICompressor` returned by `createCompressor()` must override `serializedAs()` to return
+the exact built-in compressor class it is substituting (e.g. `LZ4Compressor.class`). This
+ensures the SSTable metadata records the canonical name, which is required for
+compatibility across nodes and restarts.
+
 == Advanced Use
 
 Advanced users can provide their own compression class by implementing

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

@@ -89,6 +89,7 @@ public class Config
     public ParameterizedClass crypto_provider;
     public ParameterizedClass network_authorizer;
     public ParameterizedClass cidr_authorizer;
+    public Map<String, ParameterizedClass> compressor_providers = new HashMap<>();
 
     @Replaces(oldName = "permissions_validity_in_ms", converter = Converters.MILLIS_DURATION_INT, deprecated = true)
     public volatile DurationSpec.IntMillisecondsBound permissions_validity = new DurationSpec.IntMillisecondsBound("2s");

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

@@ -97,6 +97,7 @@ import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
 import org.apache.cassandra.gms.IFailureDetector;
 import org.apache.cassandra.gms.VersionedValue;
 import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.compress.CompressorRegistry;
 import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.big.BigFormat;
 import org.apache.cassandra.io.util.DiskOptimizationStrategy;
@@ -375,6 +376,8 @@ public class DatabaseDescriptor
 
         applySSTableFormats();
 
+        applyCompressorProviders();
+
         applySimpleConfig();
 
         applyPartitioner();
@@ -439,6 +442,7 @@ public class DatabaseDescriptor
         applyCompatibilityMode();
         diskOptimizationStrategy = new SpinningDiskOptimizationStrategy();
         applySSTableFormats();
+        applyCompressorProviders();
     }
 
     private static void assertNotDaemonInitialized()
@@ -552,6 +556,8 @@ public class DatabaseDescriptor
 
         applySSTableFormats();
 
+        applyCompressorProviders();
+
         applyCryptoProvider();
 
         applySimpleConfig();
@@ -581,6 +587,11 @@ public class DatabaseDescriptor
         applyStartupChecks();
     }
 
+    public static void applyCompressorProviders()
+    {
+        CompressorRegistry.instance.registerProviders(conf.compressor_providers);
+    }
+
     private static void applySimpleConfig()
     {
         //Doing this first before all other things in case other pieces of config want to construct
@@ -2031,6 +2042,11 @@ public class DatabaseDescriptor
         DatabaseDescriptor.cryptoProvider = cryptoProvider;
     }
 
+    public static Map<String, ParameterizedClass> getCompressionProviders()
+    {
+        return conf.compressor_providers;
+    }
+
     /**
      * Returns the authenticator configured for this node.
      */

src/java/org/apache/cassandra/io/compress/AbstractCompressionProvider.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.compress;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * Abstract base class for compression providers.
+ * Provides common functionality for loading and managing compressors
+ * with configurable fallback behavior for plugin compression libraries.
+ */
+public abstract class AbstractCompressionProvider
+{
+    /**
+     * Configuration key for enabling fallback to the default provider.
+     */
+    public static final String FAIL_ON_MISSING_PROVIDER = "fail_on_missing_provider";
+
+    private Map<String, String> parameters = Collections.emptyMap();
+
+    /**
+     * Checks if this compression provider is in a healthy state and ready to use.
+     *
+     * <p>This method should perform any necessary health checks, such as verifying
+     * that required native libraries are loaded, able to do compress/decompress successfully, etc.</p>
+     *
+     * @return true if the provider is healthy and ready to use, false otherwise
+     * @throws Exception if an error occurs during the health check
+     */
+    public abstract boolean isHealthy() throws Exception;
+
+    /**
+     * Creates a new compressor instance with the given compression parameters.
+     *
+     * <p>This method is called when a new compressor is needed. The implementation
+     * should create and configure a compressor based on the provided options.</p>
+     *
+     * @param options Configuration options for the compressor. May be null or empty.
+     * @return A new ICompressor instance
+     * @throws IllegalStateException if the compressor cannot be created
+     */
+    public abstract ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options) throws IllegalStateException;
+
+    /**
+     * Initialises this provider with the {@code parameters} block from its
+     * {@code compressor_providers} entry in cassandra.yaml. Called by {@link CompressorRegistry}
+     * exactly once, after no-arg construction and before {@link #isHealthy()}. May be empty but is
+     * never null.
+     * <p>
+     * The map still contains the registry-reserved key {@link #FAIL_ON_MISSING_PROVIDER} when set:
+     * the registry leaves it in place so that the default {@link #isFailOnMissingProvider()} (which
+     * reads it back out of the stored parameters) reflects the configured value at compressor-creation
+     * time. Subclasses should ignore that key when interpreting their own options.
+     * <p>
+     * The default implementation stores the map so subclasses can retrieve it via
+     * {@link #getParameters()}. Subclasses that override this method must call
+     * {@code super.init(parameters)} if they want {@code getParameters()} to work.
+     */
+    public void init(Map<String, String> parameters)
+    {
+        if (parameters != null)
+            this.parameters = parameters;
+    }
+
+    /**
+     * Returns the {@code parameters} this provider was initialised with, or an empty map if
+     * {@link #init} has not been called.
+     */
+    public Map<String, String> getParameters()
+    {
+        return parameters;
+    }
+
+    public boolean isFailOnMissingProvider()
+    {
+        return Boolean.parseBoolean(getParameters().getOrDefault(FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString()));
+    }
+}

src/java/org/apache/cassandra/io/compress/CompressionMetadata.java

@@ -497,7 +497,7 @@ public class CompressionMetadata extends WrappedSharedCloseable
         {
             try
             {
-                out.writeUTF(parameters.getSstableCompressor().getClass().getSimpleName());
+                out.writeUTF(parameters.getSstableCompressor().serializedAs().getSimpleName());
                 out.writeInt(parameters.getOtherOptions().size());
                 for (Map.Entry<String, String> entry : parameters.getOtherOptions().entrySet())
                 {

src/java/org/apache/cassandra/io/compress/CompressorRegistry.java

@@ -0,0 +1,305 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.compress;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static org.apache.cassandra.io.compress.AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER;
+
+/**
+ * CompressorRegistry manages the registration and retrieval of compression providers.
+ * Compression providers supply compressor implementations used for compressing data.
+ */
+public final class CompressorRegistry
+{
+    private static final Logger logger = LoggerFactory.getLogger(CompressorRegistry.class);
+
+    @VisibleForTesting
+    public static final AbstractCompressionProvider DEFAULT_COMPRESSION_PROVIDER = new DefaultCompressionProvider();
+
+    /**
+     * Map of compressor class names to their registered providers. Keyed by FQN (not {@code Class<?>})
+     * so that registration of built-in compressors does not force their classloading — that would
+     * trigger native library loads (snappy, zstd) at tool/client init time. See CASSANDRA-20975.
+     */
+    private static final Map<String, AbstractCompressionProvider> compressionProviders = new HashMap<>();
+
+    /**
+     * Singleton instance of the registry.
+     */
+    public static final CompressorRegistry instance = new CompressorRegistry();
+
+    private CompressorRegistry()
+    {
+    }
+
+    @VisibleForTesting
+    public void reset()
+    {
+        compressionProviders.clear();
+    }
+
+    /**
+     * Enum representing in-built supported compressor types and their abbreviations.
+     * </p>
+     * The compressor class is stored as a FQN string rather than a {@code Class<?>} literal so that
+     * iterating the enum (e.g. during {@link #registerProviders}) does not force classloading of
+     * every built-in compressor — important because {@link SnappyCompressor} and the {@code Zstd*}
+     * compressors trigger native-library loads in their static initializers, which we want to defer
+     * until the compressor is actually used.
+     */
+    public enum CompressorType
+    {
+        DEFLATE("org.apache.cassandra.io.compress.DeflateCompressor", "deflate"),
+        LZ4("org.apache.cassandra.io.compress.LZ4Compressor", "lz4"),
+        NOOP("org.apache.cassandra.io.compress.NoopCompressor", "noop"),
+        SNAPPY("org.apache.cassandra.io.compress.SnappyCompressor", "snappy"),
+        ZSTD("org.apache.cassandra.io.compress.ZstdCompressor", "zstd"),
+        ZSTD_DICTIONARY("org.apache.cassandra.io.compress.ZstdDictionaryCompressor", "zstd_dictionary");
+
+        public final String compressorClassName;
+        public final String simpleName;
+        public final String abbreviation;
+
+        CompressorType(String compressorClassName, String abbreviation)
+        {
+            this.compressorClassName = compressorClassName;
+            this.simpleName = compressorClassName.substring(compressorClassName.lastIndexOf('.') + 1);
+            this.abbreviation = abbreviation;
+        }
+
+        /**
+         * Resolves the compressor {@link Class} on demand. First call forces classloading of the
+         * underlying compressor implementation (and any side effects of its static initializer).
+         */
+        public Class<?> compressorClass()
+        {
+            return FBUtilities.classForName(compressorClassName, "compressor");
+        }
+    }
+
+    /**
+     * Returns the compression provider for the given compressor class. If the class is not one of
+     * the built-in {@link CompressorType} entries (e.g. a 3rd-party / user-supplied compressor not
+     * configured in {@code compressor_providers}), returns {@link #DEFAULT_COMPRESSION_PROVIDER},
+     * which reflectively invokes the class's static {@code create(Map)} factory.
+     *
+     * @param compressorClass class of the compressor
+     * @return the compression provider instance, never null
+     */
+    public AbstractCompressionProvider getProvider(Class<?> compressorClass)
+    {
+        return compressionProviders.getOrDefault(compressorClass.getName(), DEFAULT_COMPRESSION_PROVIDER);
+    }
+
+    /**
+     * Returns a filtered view of the registered custom compression providers, with key as compressor class.
+     * <p>
+     * Note that this method triggers classloading for each compressor entry that passes the filter,
+     * as the internal FQN string keys are resolved to {@code Class<?>} via {@link FBUtilities#classForName}.
+     * Call this method only after the server has fully initialised (e.g. from startup checks),
+     * not at tool or client init time.
+
+     * @return map of compressor {@link Class} to their registered providers for entries matching
+     * the filter; never {@code null}
+     */
+    public Map<Class<?>, AbstractCompressionProvider> getCustomProviders()
+    {
+        return compressionProviders.entrySet()
+                                   .stream()
+                                   .filter(e -> e.getValue() != CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER)
+                                   .collect(Collectors.toMap(e -> FBUtilities.classForName(e.getKey(), "compressor"),
+                                                             Map.Entry::getValue));
+    }
+
+    /**
+     * Creates a compressor. Firstly, we get a provider for this compressor class, or we use default compressor provider
+     * if there is not such a mapping. Then we try to use that provider to instantiate a compressor with given parameters.
+     * When the construction of a compressor fails, if a provider is configured is not configured to fallback,
+     * an exception is thrown, otherwise a compressor by a default compressor provider is created.
+     *
+     * @param compressorClass    compressor class to create a compressor of
+     * @param compressionOptions compressor options
+     * @return an instance of a given compressor class
+     */
+    public ICompressor getCompressor(Class<?> compressorClass, Map<String, String> compressionOptions)
+    {
+        AbstractCompressionProvider provider = getProvider(compressorClass);
+        ICompressor compressor;
+        try
+        {
+            compressor = provider.createCompressor(compressorClass, compressionOptions);
+        }
+        catch (Throwable t)
+        {
+            if (provider == DEFAULT_COMPRESSION_PROVIDER)
+                throw t;
+
+            logger.warn("Failed to create compressor {}. Will attempt fallback to default if enabled. Message: {}",
+                        compressorClass.getName(),
+                        t.getMessage());
+
+            if (provider.isFailOnMissingProvider())
+                throw t;
+
+            compressor = DEFAULT_COMPRESSION_PROVIDER.createCompressor(compressorClass, compressionOptions);
+        }
+
+        // Validate the masquerade target for both the provider and the fallback path: whatever is
+        // returned must serialize as exactly the compressor class that was requested, otherwise the
+        // schema / on-disk format would record a name that cannot be resolved on peers and restarts.
+        Class<? extends ICompressor> serializedAs = compressor.serializedAs();
+        if (serializedAs == null || serializedAs.getSimpleName().isEmpty())
+            throw new ConfigurationException(String.format("ICompressor.serializedAs() of a compressor created by provider %s returned %s. " +
+                                                           "It must return a non-anonymous built-in compressor class in package " +
+                                                           "org.apache.cassandra.io.compress.",
+                                                           provider.getClass(),
+                                                           serializedAs));
+        if (serializedAs != compressorClass)
+            throw new ConfigurationException(String.format("The result of ICompressor.serializedAs(), %s, of a compressor object created " +
+                                                           "by a compressor provider %s does not match the compressor class to get a compressor for. " +
+                                                           "You need to override serializedAs() method of your custom compressor and return " +
+                                                           "base compressor class it is the substitute for.",
+                                                           serializedAs,
+                                                           provider.getClass()));
+        return compressor;
+    }
+
+    /**
+     * Populates the registry with compression providers specified in the configuration.
+     * Should be called once during initialization to ensure providers are registered and available
+     * for use. If a provider fails to initialize and fallback is enabled, the default provider is used.
+     * <p>
+     * Each invocation constructs fresh provider instances and re-runs {@link AbstractCompressionProvider#init}
+     * without tearing down any previously registered instance, so a custom provider that acquires
+     * resources (native memory, threads, handles) in {@code init()} must tolerate being discarded if
+     * this is called more than once (as tests do via {@link #reset()}).
+     *
+     * @param providerOptions map of compressor names to their configuration
+     * @throws ConfigurationException if a provider fails to initialize and fallback is not enabled
+     */
+    public void registerProviders(Map<String, ParameterizedClass> providerOptions)
+    {
+        if (providerOptions == null)
+            return;
+
+        Set<String> validKeys = new HashSet<>();
+        for (CompressorType type : CompressorType.values())
+        {
+            validKeys.add(type.compressorClassName);
+            validKeys.add(type.simpleName);
+            validKeys.add(type.abbreviation);
+        }
+
+        for (String key : providerOptions.keySet())
+        {
+            if (!validKeys.contains(key))
+                throw new ConfigurationException("Unknown compressor key '" + key + "' in compressor_providers. " +
+                                                 "Expected a built-in compressor's fully qualified class name, simple class name, " +
+                                                 "or abbreviation: " + validKeys);
+        }
+
+        for (CompressorType type : CompressorType.values())
+        {
+            ParameterizedClass providerConfig = findProviderConfig(providerOptions, type);
+            if (providerConfig == null)
+            {
+                compressionProviders.put(type.compressorClassName, DEFAULT_COMPRESSION_PROVIDER);
+            }
+            else
+            {
+                AbstractCompressionProvider provider = resolveProvider(providerConfig);
+                compressionProviders.put(type.compressorClassName, provider);
+                logger.info("Adding '{}' provider for '{}'", provider.getClass().getName(), type.compressorClassName);
+            }
+        }
+    }
+
+    private ParameterizedClass findProviderConfig(Map<String, ParameterizedClass> providerOptions, CompressorType type)
+    {
+        ParameterizedClass parameterizedClass = providerOptions.get(type.compressorClassName);
+
+        if (parameterizedClass == null)
+            parameterizedClass = providerOptions.get(type.simpleName);
+
+        if (parameterizedClass == null)
+            parameterizedClass = providerOptions.get(type.abbreviation);
+
+        return parameterizedClass;
+    }
+
+    /**
+     * Returns a compression provider with configuration specified in the config file.
+     * If the provider fails to initialize or is not healthy, will attempt to fall back to the default provider
+     * if enabled in the configuration.
+     *
+     * @param providerConfig the configuration for the provider
+     * @return the compression provider instance
+     * @throws ConfigurationException if both the specified and fallback providers fail to initialize
+     */
+    AbstractCompressionProvider resolveProvider(ParameterizedClass providerConfig)
+    {
+        if (providerConfig.class_name == null)
+            throw new ConfigurationException("compressor_providers entry is missing required 'class_name': " + providerConfig);
+
+        Map<String, String> p = providerConfig.parameters == null ? Collections.emptyMap() : providerConfig.parameters;
+
+        try
+        {
+            AbstractCompressionProvider compressionProvider = FBUtilities.newCompressionProvider(providerConfig.class_name);
+            compressionProvider.init(new HashMap<>(p));
+
+            if (compressionProvider.isHealthy())
+            {
+                return compressionProvider;
+            }
+            else
+            {
+                logger.warn("Compression provider {} is not healthy, attempting fallback.", providerConfig.class_name);
+            }
+        }
+        catch (Throwable e)
+        {
+            logger.warn("Failed to initialize specified compression provider {}. Will attempt fallback to default if enabled.",
+                        providerConfig.class_name,
+                        e);
+        }
+
+        boolean failOnMissingProvider = Boolean.parseBoolean(p.getOrDefault(FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString()));
+
+        if (!failOnMissingProvider)
+            return DEFAULT_COMPRESSION_PROVIDER;
+
+        throw new ConfigurationException("Failed to initialize compression provider " + providerConfig);
+    }
+}

src/java/org/apache/cassandra/io/compress/DefaultCompressionProvider.java

@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.compress;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Map;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+
+import static java.lang.String.format;
+
+/**
+ * Default implementation of {@link AbstractCompressionProvider} 
+ * 
+ * This provider acts as default implementation which creates compressor for built-in algorithms (LZ4, Snappy, Deflate, etc.)
+ * It relies on Cassandra's built-in compression implementations and does not require any external dependencies.
+ * It is used when:
+ *   No custom compression provider is configured
+ *   A custom compression provider fails and fallback is enabled
+ */
+public class DefaultCompressionProvider extends AbstractCompressionProvider
+{
+    /**
+     * Checks if this compression provider is healthy and ready to use.
+     * The default provider is always considered healthy as it has no external
+     * dependencies and relies on Cassandra's built-in compression mechanisms.
+     *
+     * @return Always returns {@code true}
+     */
+    @Override
+    public boolean isHealthy()
+    {
+        return true;
+    }
+
+    /**
+     * Creates a new compressor instance with the given compression parameters.
+     *
+     * @param compressionOptions Configuration options for the compressor
+     * @return A new ICompressor instance
+     * @throws ConfigurationException if the compressor class does not have a valid create method,
+     * if there are unknown compression options, or if the compressor creation fails
+     */
+    @Override
+    public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> compressionOptions) throws IllegalStateException
+    {
+        try
+        {
+            Method method = compressorClass.getMethod("create", Map.class);
+            ICompressor compressor = (ICompressor)method.invoke(null, compressionOptions);
+            // Check for unknown options
+            for (String provided : compressionOptions.keySet())
+                if (!compressor.supportedOptions().contains(provided))
+                    throw new ConfigurationException("Unknown compression options " + provided);
+            return compressor;
+        }
+        catch (NoSuchMethodException e)
+        {
+            throw new ConfigurationException("create method not found", e);
+        }
+        catch (SecurityException e)
+        {
+            throw new ConfigurationException("Access forbidden", e);
+        }
+        catch (IllegalAccessException e)
+        {
+            throw new ConfigurationException("Cannot access method create in " + compressorClass.getName(), e);
+        }
+        catch (InvocationTargetException e)
+        {
+            if (e.getTargetException() instanceof ConfigurationException)
+                throw (ConfigurationException) e.getTargetException();
+
+            Throwable cause = e.getCause() == null
+                            ? e
+                            : e.getCause();
+
+            throw new ConfigurationException(format("%s.create() threw an error: %s %s",
+                                                    compressorClass.getSimpleName(),
+                                                    cause.getClass().getName(),
+                                                    cause.getMessage()),
+                                             e);
+        }
+        catch (ExceptionInInitializerError e)
+        {
+            throw new ConfigurationException("Cannot initialize class " + compressorClass.getName());
+        }
+    }
+}

src/java/org/apache/cassandra/io/compress/ICompressor.java

@@ -88,4 +88,19 @@ public interface ICompressor
     {
         return ImmutableSet.copyOf(EnumSet.allOf(Uses.class));
     }
+
+    /**
+     * Compressor class recorded in the table schema and SSTable compression metadata for
+     * compressors built by this instance. Plugin wrappers should override this to return the
+     * built-in they substitute for, so schema and on-disk format stay portable to peers and
+     * restarts without the plugin.
+     * <p>
+     * Contract: the returned class must (a) live in {@code org.apache.cassandra.io.compress}.
+     * Serializers write only its simple name; deserialization prepends that package.
+     * (b) have a non-empty simple name (no anonymous/synthetic classes).
+     */
+    default Class<? extends ICompressor> serializedAs()
+    {
+        return getClass();
+    }
 }

src/java/org/apache/cassandra/schema/CompressionParams.java

@@ -18,8 +18,6 @@
 package org.apache.cassandra.schema;
 
 import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -36,6 +34,7 @@ import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.db.compression.CompressionDictionary;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.compress.CompressorRegistry;
 import org.apache.cassandra.io.compress.DeflateCompressor;
 import org.apache.cassandra.io.compress.ICompressor;
 import org.apache.cassandra.io.compress.IDictionaryCompressor;
@@ -63,6 +62,7 @@ public final class CompressionParams
     public static final String ENABLED = "enabled";
     public static final String MIN_COMPRESS_RATIO = "min_compress_ratio";
 
+    private static final CompressorRegistry registry = CompressorRegistry.instance;
     public static final CompressionParams DEFAULT = !CassandraRelevantProperties.DETERMINISM_SSTABLE_COMPRESSION_DEFAULT.getBoolean()
                                                     ? noCompression()
                                                     : new CompressionParams(LZ4Compressor.create(Collections.emptyMap()),
@@ -321,47 +321,7 @@ public final class CompressionParams
             return null;
         }
 
-        try
+        return registry.getCompressor(compressorClass, compressionOptions);
-        {
-            Method method = compressorClass.getMethod("create", Map.class);
-            ICompressor compressor = (ICompressor)method.invoke(null, compressionOptions);
-            // Check for unknown options
-            for (String provided : compressionOptions.keySet())
-                if (!compressor.supportedOptions().contains(provided))
-                    throw new ConfigurationException("Unknown compression options " + provided);
-            return compressor;
-        }
-        catch (NoSuchMethodException e)
-        {
-            throw new ConfigurationException("create method not found", e);
-        }
-        catch (SecurityException e)
-        {
-            throw new ConfigurationException("Access forbiden", e);
-        }
-        catch (IllegalAccessException e)
-        {
-            throw new ConfigurationException("Cannot access method create in " + compressorClass.getName(), e);
-        }
-        catch (InvocationTargetException e)
-        {
-            if (e.getTargetException() instanceof ConfigurationException)
-                throw (ConfigurationException) e.getTargetException();
-
-            Throwable cause = e.getCause() == null
-                            ? e
-                            : e.getCause();
-
-            throw new ConfigurationException(format("%s.create() threw an error: %s %s",
-                                                    compressorClass.getSimpleName(),
-                                                    cause.getClass().getName(),
-                                                    cause.getMessage()),
-                                             e);
-        }
-        catch (ExceptionInInitializerError e)
-        {
-            throw new ConfigurationException("Cannot initialize class " + compressorClass.getName());
-        }
     }
 
     public static ICompressor createCompressor(ParameterizedClass compression) throws ConfigurationException
@@ -509,7 +469,8 @@ public final class CompressionParams
             return Collections.singletonMap(ENABLED, "false");
 
         Map<String, String> options = new HashMap<>(otherOptions);
-        options.put(CLASS, sstableCompressor.getClass().getName());
+        // Use the one saved in the registry, we don't want to save the name of the service provider compressor here!
+        options.put(CLASS, sstableCompressor.serializedAs().getName());
         options.put(CHUNK_LENGTH_IN_KB, chunkLengthInKB());
         if (minCompressRatio != DEFAULT_MIN_COMPRESS_RATIO)
             options.put(MIN_COMPRESS_RATIO, String.valueOf(minCompressRatio));
@@ -555,7 +516,7 @@ public final class CompressionParams
         public void serialize(CompressionParams parameters, DataOutputPlus out, int version) throws IOException
         {
             assert version >= MessagingService.VERSION_40;
-            out.writeUTF(parameters.sstableCompressor.getClass().getSimpleName());
+            out.writeUTF(parameters.sstableCompressor.serializedAs().getSimpleName());
             out.writeInt(parameters.otherOptions.size());
             for (Map.Entry<String, String> entry : parameters.otherOptions.entrySet())
             {
@@ -596,7 +557,7 @@ public final class CompressionParams
         public long serializedSize(CompressionParams parameters, int version)
         {
             assert version >= MessagingService.VERSION_40;
-            long size = TypeSizes.sizeof(parameters.sstableCompressor.getClass().getSimpleName());
+            long size = TypeSizes.sizeof(parameters.sstableCompressor.serializedAs().getSimpleName());
             size += TypeSizes.sizeof(parameters.otherOptions.size());
             for (Map.Entry<String, String> entry : parameters.otherOptions.entrySet())
             {

src/java/org/apache/cassandra/service/StartupChecks.java

@@ -21,6 +21,7 @@ import java.io.BufferedReader;
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.lang.management.RuntimeMXBean;
+import java.nio.ByteBuffer;
 import java.nio.file.FileStore;
 import java.nio.file.FileVisitResult;
 import java.nio.file.FileVisitor;
@@ -30,12 +31,14 @@ import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Random;
 import java.util.ServiceConfigurationError;
 import java.util.ServiceLoader;
 import java.util.Set;
@@ -67,6 +70,9 @@ import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.StartupException;
+import org.apache.cassandra.io.compress.AbstractCompressionProvider;
+import org.apache.cassandra.io.compress.CompressorRegistry;
+import org.apache.cassandra.io.compress.ICompressor;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.UUIDBasedSSTableId;
 import org.apache.cassandra.io.util.File;
@@ -131,6 +137,7 @@ public class StartupChecks
                                                                       checkSystemKeyspaceState,
                                                                       checkLegacyAuthTables,
                                                                       checkKernelParamsForAsyncProfiler,
+                                                                      checkCustomCompressionProviders,
                                                                       new DataResurrectionCheck());
 
     public List<StartupCheck> getChecks()
@@ -346,6 +353,157 @@ public class StartupChecks
         }
     };
 
+    public static final StartupCheck checkCustomCompressionProviders = new StartupCheck()
+    {
+        @Override
+        public String name()
+        {
+            return "custom_compression_providers";
+        }
+
+        @Override
+        public void execute(StartupChecksConfiguration configuration) throws StartupException
+        {
+            if (configuration.isDisabled(name()))
+                return;
+
+            // Resolving the custom providers forces classloading (and native-lib init) of each
+            // configured compressor; a missing class or failed native init is a configuration error.
+            Map<Class<?>, AbstractCompressionProvider> providers = getCustomProviders();
+
+            if (providers.isEmpty())
+                return;
+
+            long seed = (new Random()).nextLong();
+            Random random = new Random(seed);
+            byte[] payload = smokeTestPayload(random);
+
+            logger.info("Running compression smoke test for {} custom provider(s) with seed {}. " +
+                        "To reproduce a failure, regenerate the 4 KiB payload with new java.util.Random({}).",
+                        providers.size(), seed, seed);
+
+            List<String> failedProviders = new ArrayList<>();
+            for (Map.Entry<Class<?>, AbstractCompressionProvider> entry : providers.entrySet())
+            {
+                Class<?> compressorClass = entry.getKey();
+                AbstractCompressionProvider provider = entry.getValue();
+
+                ICompressor custom;
+
+                try
+                {
+                    custom = provider.createCompressor(compressorClass, Collections.emptyMap());
+                }
+                catch (Throwable t)
+                {
+                    throw new StartupException(StartupException.ERR_WRONG_CONFIG,
+                                               String.format("Unable to instantiate a compressor for class %s " +
+                                                             "for the purposes of a startup check from provider %s.",
+                                                             compressorClass.getName(),
+                                                             provider.getClass().getName()));
+                }
+
+                if (custom.serializedAs() != compressorClass)
+                {
+                    throw new StartupException(StartupException.ERR_WRONG_CONFIG,
+                                               String.format("Provider %s returned a compressor whose serializedAs() is %s, " +
+                                                             "but it must be %s (the built-in it substitutes for).",
+                                                             provider.getClass().getName(),
+                                                             custom.serializedAs(),
+                                                             compressorClass.getName()));
+                }
+
+                ICompressor builtin = CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER.createCompressor(compressorClass, Collections.emptyMap());
+
+                try
+                {
+                    // Round trip both ways so the custom and built-in compressors are proven to share
+                    // an on-disk-compatible format (peers/restarts without the plugin read the data).
+                    assertCompatibleRoundTrip(custom, builtin, payload);
+                    assertCompatibleRoundTrip(builtin, custom, payload);
+
+                    logger.info("Compression smoke test passed for custom provider {} ({}).",
+                                provider.getClass().getName(), compressorClass.getSimpleName());
+                }
+                catch (Throwable t)
+                {
+                    logger.error("Compression smoke test failed for custom provider {} ({}); reproduce with seed {}.",
+                                 provider.getClass().getName(), compressorClass.getSimpleName(), seed, t);
+                    failedProviders.add(provider.getClass().getName() + " -> " + compressorClass.getSimpleName());
+                }
+            }
+
+            if (!failedProviders.isEmpty())
+            {
+                throw new StartupException(StartupException.ERR_WRONG_MACHINE_STATE,
+                                           String.format("The following custom compression providers failed smoke test: %s. " +
+                                                         "Providers substitute for a built-in compressor, so non byte-compatible output " +
+                                                         "is silent data corruption when read without the provider. Reproduce with the seed %s.",
+                                                         Joiner.on(", ").join(failedProviders),
+                                                         seed));
+            }
+        }
+
+        private Map<Class<?>, AbstractCompressionProvider> getCustomProviders() throws StartupException
+        {
+            Map<Class<?>, AbstractCompressionProvider> providers;
+            try
+            {
+                providers = CompressorRegistry.instance.getCustomProviders();
+            }
+            catch (Throwable t)
+            {
+                throw new StartupException(StartupException.ERR_WRONG_CONFIG,
+                                           "Failed to load configured custom compression providers; " +
+                                           "check compressor_providers in cassandra.yaml", t);
+            }
+            return providers;
+        }
+        // Compresses payload with `compressor`, decompresses with `decompressor`, and verifies the
+        // result matches - proving the two share an on-disk-compatible format. Throws on mismatch.
+        private void assertCompatibleRoundTrip(ICompressor compressor, ICompressor decompressor, byte[] payload) throws IOException
+        {
+            ByteBuffer input = compressor.preferredBufferType().allocate(payload.length);
+            int compressedLength = compressor.initialCompressedBufferLength(payload.length);
+            ByteBuffer compressed = compressor.preferredBufferType().allocate(compressedLength);
+            input.put(payload);
+            input.flip();
+            compressor.compress(input, compressed);
+            compressed.flip();
+
+            // Within a compress/uncompress call the in/out buffers must share a type, but the compressor
+            // and decompressor may prefer different ones; only re-stage the compressed bytes in that case.
+            if (compressor.preferredBufferType() != decompressor.preferredBufferType())
+            {
+                ByteBuffer staged = decompressor.preferredBufferType().allocate(compressed.remaining());
+                staged.put(compressed);
+                staged.flip();
+                compressed = staged;
+            }
+
+            ByteBuffer output = decompressor.preferredBufferType().allocate(payload.length);
+            decompressor.uncompress(compressed, output);
+            output.flip();
+
+            if (!output.equals(ByteBuffer.wrap(payload)))
+            {
+                throw new IOException(String.format("Round-trip mismatch: compressed with %s, decompressed with %s",
+                                                    compressor.getClass().getName(), decompressor.getClass().getName()));
+            }
+        }
+        private byte[] smokeTestPayload(Random random)
+        {
+            // 4 KiB payload: the first half zeros (highly compressible, exercises the real compression
+            // path), the second half high-entropy bytes (incompressible, exercises the compressor's
+            // worst-case output sizing via initialCompressedBufferLength).
+            byte[] testPayload = new byte[4 * 1024];
+            byte[] randomHalf = new byte[testPayload.length / 2];
+            random.nextBytes(randomHalf);
+            System.arraycopy(randomHalf, 0, testPayload, testPayload.length / 2, randomHalf.length);
+            return testPayload;
+        }
+    };
+
     public static final StartupCheck checkValidLaunchDate = new StartupCheck()
     {
         @Override

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -86,6 +86,7 @@ import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.compress.AbstractCompressionProvider;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.format.StatsComponent;
 import org.apache.cassandra.io.sstable.metadata.MetadataType;
@@ -733,6 +734,26 @@ public class FBUtilities
         }
     }
 
+    public static AbstractCompressionProvider newCompressionProvider(String className) throws ConfigurationException
+    {
+        try
+        {
+            if (!className.contains("."))
+                className = "org.apache.cassandra.io.compress." + className;
+
+            Class<?> compressionProviderClass = FBUtilities.classForName(className, "compression service provider");
+            return (AbstractCompressionProvider) compressionProviderClass.getConstructor().newInstance();
+        }
+        catch (ConfigurationException e)
+        {
+            throw e;
+        }
+        catch (Exception e)
+        {
+            throw new ConfigurationException(String.format("Unable to create an instance of the compression service provider for %s", className), e);
+        }
+    }
+
     /**
      * @return The Class for the given name.
      * @param classname Fully qualified classname.

test/distributed/org/apache/cassandra/distributed/test/CompressorProviderTest.java

@@ -0,0 +1,552 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableCallable;
+import org.apache.cassandra.io.compress.AbstractCompressionProvider;
+import org.apache.cassandra.io.compress.CompressionMetadata;
+import org.apache.cassandra.io.compress.CompressorRegistry;
+import org.apache.cassandra.io.compress.DefaultCompressionProvider;
+import org.apache.cassandra.io.compress.ICompressor;
+import org.apache.cassandra.io.compress.LZ4Compressor;
+import org.apache.cassandra.io.compress.SnappyCompressor;
+import org.apache.cassandra.io.sstable.format.CompressionInfoComponent;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Distributed tests for pluggable compression providers configured via
+ * the {@code compressor_providers} option in cassandra.yaml.
+ * <p>
+ * Covers (CASSANDRA-20975):
+ * - default providers are registered when no plugin is configured;
+ * - a custom healthy provider is wired into the registry and used by table compression;
+ * - an unhealthy provider falls back to the default when fallback is enabled;
+ * - an unhealthy provider fails node startup when fallback is disabled.
+ */
+public class CompressorProviderTest extends TestBaseImpl
+{
+    @Test
+    public void testDefaultProvidersRegisteredWhenNoConfig() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.build(1).start()))
+        {
+            cluster.get(1).runOnInstance(() -> {
+                for (CompressorRegistry.CompressorType type : CompressorRegistry.CompressorType.values())
+                {
+                    AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(type.compressorClass());
+                    assertNotNull("Provider missing for " + type, provider);
+                    assertSame("Built-in compressor " + type + " must use the default provider when none is configured",
+                               DefaultCompressionProvider.class, provider.getClass());
+                }
+            });
+        }
+    }
+
+    @Test
+    public void testCustomProviderRegisteredAndUsedForTableCompression() throws Throwable
+    {
+        Map<String, Object> snappyProviderOptions = new HashMap<>();
+        snappyProviderOptions.put("class_name", HealthyTestProvider.class.getName());
+        snappyProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> providers = new HashMap<>();
+        providers.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptions);
+
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .withConfig(config -> config.set("compressor_providers", providers))
+                                           .start()))
+        {
+            // The registry must hold our custom provider for SnappyCompressor and the default
+            // for every other built-in type.
+            cluster.get(1).runOnInstance(() -> {
+                AbstractCompressionProvider snappy = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+                assertNotNull(snappy);
+                assertEquals(HealthyTestProvider.class.getName(), snappy.getClass().getName());
+
+                for (CompressorRegistry.CompressorType type : CompressorRegistry.CompressorType.values())
+                {
+                    if (type.compressorClass() == SnappyCompressor.class)
+                        continue;
+                    AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(type.compressorClass());
+                    assertSame(DefaultCompressionProvider.class, provider.getClass());
+                }
+            });
+
+            // Reset the counter so the assertion below only sees calls produced by this test's DDL/writes.
+            cluster.get(1).runOnInstance(() -> HealthyTestProvider.CREATE_CALLS.set(0));
+
+            // End-to-end: create a table that uses Snappy compression. The compressor must be sourced
+            // through HealthyTestProvider; writes and reads must succeed because the provider
+            // delegates compressor creation to the default provider.
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".plugin_snappy (k int PRIMARY KEY, v text) " +
+                                 "WITH compression = {'class': 'SnappyCompressor', 'chunk_length_in_kb': '4'}");
+
+            for (int i = 0; i < 100; i++)
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".plugin_snappy (k, v) VALUES (?, ?)",
+                                               ConsistencyLevel.ONE, i, "value-" + i);
+
+            cluster.get(1).flush(KEYSPACE);
+
+            Object[][] rows = cluster.coordinator(1).execute("SELECT count(*) FROM " + KEYSPACE + ".plugin_snappy",
+                                                             ConsistencyLevel.ONE);
+            assertEquals(100L, rows[0][0]);
+
+            int creates = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + creates,
+                       creates > 0);
+
+            int createsSnappy = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_SNAPPY_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + createsSnappy,
+                       createsSnappy > 0);
+
+            // The schema must record the *concrete* compressor class, not the provider — this is what
+            // peers and future restarts use to rebuild the compression chain.
+            Object[][] schemaRows = cluster.coordinator(1).execute(
+            "SELECT compression FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+            ConsistencyLevel.ONE, KEYSPACE, "plugin_snappy");
+            @SuppressWarnings("unchecked")
+            Map<String, String> compressionOptions = (Map<String, String>) schemaRows[0][0];
+            assertEquals("system_schema.tables.compression must record the concrete compressor class, not the provider",
+                         SnappyCompressor.class.getName(), compressionOptions.get("class"));
+
+            // Reconstruct CompressionMetadata from the flushed SSTable on disk and check the
+            // rebuilt compressor's serializedAs() — this is the class name that was actually
+            // written into *-CompressionInfo.db, and the one a peer without the plugin would
+            // read. getClass() on the rebuilt instance would just yield MySnappyCompressor again
+            // because the plugin replays its substitution during CompressionMetadata.open.
+            String onDiskSerializedAs = cluster.get(1).callOnInstance((SerializableCallable<String>) () -> {
+                ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("plugin_snappy");
+                SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+                try (CompressionMetadata metadata = CompressionInfoComponent.loadIfExists(sstable.descriptor, null))
+                {
+                    Assert.assertNotNull(metadata);
+                    return metadata.parameters.getSstableCompressor().serializedAs().getName();
+                }
+            });
+            assertEquals("Rebuilt CompressionMetadata.serializedAs() must yield the underlying compressor - " +
+                         "what the plugin advertises for portability - not the wrapper",
+                         SnappyCompressor.class.getName(), onDiskSerializedAs);
+        }
+    }
+
+    @Test
+    public void testCompressWithCustomProviderAndDecompressWithDefault() throws Throwable
+    {
+        //This test will simulate a scenario where data is compressed with a custom provider and then read on a node which does not have the provider configured,
+        // In the initial phase the test does write/read checks like testCustomProviderRegisteredAndUsedForTableCompression then it brings down the cluster,
+        // restarts with DefaultCompressionProvider for SnappyCompressor, and checks that the data can still be read.
+        Map<String, Object> snappyProviderOptions = new HashMap<>();
+        snappyProviderOptions.put("class_name", HealthyTestProvider.class.getName());
+        snappyProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> providers = new HashMap<>();
+        providers.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptions);
+
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .withConfig(config -> config.set("compressor_providers", providers))
+                                           .start()))
+        {
+            // Mae sure registry holds the custom provider for SnappyCompressor
+            cluster.get(1).runOnInstance(() -> {
+                AbstractCompressionProvider snappy = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+                assertNotNull(snappy);
+                assertEquals(HealthyTestProvider.class.getName(), snappy.getClass().getName());
+            });
+
+            // Reset the counter so the assertion below only sees calls produced by this test's DDL/writes.
+            cluster.get(1).runOnInstance(() -> HealthyTestProvider.CREATE_CALLS.set(0));
+
+            // End-to-end: create a table that uses Snappy compression
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".plugin_snappy (k int PRIMARY KEY, v text) " +
+                                 "WITH compression = {'class': 'SnappyCompressor', 'chunk_length_in_kb': '4'}");
+
+            for (int i = 0; i < 100; i++)
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".plugin_snappy (k, v) VALUES (?, ?)",
+                                               ConsistencyLevel.ONE, i, "value-" + i);
+
+            cluster.get(1).flush(KEYSPACE);
+
+            Object[][] rows = cluster.coordinator(1).execute("SELECT count(*) FROM " + KEYSPACE + ".plugin_snappy",
+                                                             ConsistencyLevel.ONE);
+            assertEquals(100L, rows[0][0]);
+
+            int creates = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + creates,
+                       creates > 0);
+
+            int createsSnappy = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_SNAPPY_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + createsSnappy,
+                       createsSnappy > 0);
+
+            // Ensure schema records the *concrete* compressor class, not the provider
+            Object[][] schemaRows = cluster.coordinator(1).execute(
+            "SELECT compression FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+            ConsistencyLevel.ONE, KEYSPACE, "plugin_snappy");
+            @SuppressWarnings("unchecked")
+            Map<String, String> compressionOptions = (Map<String, String>) schemaRows[0][0];
+            assertEquals("system_schema.tables.compression must record the concrete compressor class, not the provider",
+                         SnappyCompressor.class.getName(), compressionOptions.get("class"));
+
+            // Reconstruct CompressionMetadata from the flushed SSTable on disk and check the
+            // rebuilt compressor's serializedAs()
+            String onDiskSerializedAs = cluster.get(1).callOnInstance((SerializableCallable<String>) () -> {
+                ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore("plugin_snappy");
+                SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+                try (CompressionMetadata metadata = CompressionInfoComponent.loadIfExists(sstable.descriptor, null))
+                {
+                    Assert.assertNotNull(metadata);
+                    return metadata.parameters.getSstableCompressor().serializedAs().getName();
+                }
+            });
+            assertEquals("Rebuilt CompressionMetadata.serializedAs() must yield the underlying compressor - " +
+                         "what the plugin advertises for portability - not the wrapper",
+                         SnappyCompressor.class.getName(), onDiskSerializedAs);
+
+            // Simulate a node which does not have plugin configured
+            // Shut down the cluster, change provider class name to default, and restart. It should be able to use the Default provider for SnappyCompressor,
+            // and be able to read the existing data.
+            cluster.get(1).shutdown().get();
+            snappyProviderOptions = new HashMap<>();
+            snappyProviderOptions.put("class_name", DefaultCompressionProvider.class.getName());
+            snappyProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+            providers.clear();
+            providers.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptions);
+            cluster.get(1).config().set("compressor_providers", providers);
+            cluster.get(1).startup();
+            cluster.get(1).runOnInstance(() -> {
+                AbstractCompressionProvider snappy = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+                assertNotNull(snappy);
+                assertEquals(DefaultCompressionProvider.class.getName(), snappy.getClass().getName());
+            });
+
+            //The node should be able to decompress the SSTables
+            // that were written by the plugin node, because serializedAs() ensured the
+            // on-disk CompressionInfo.db recorded SnappyCompressor (not MySnappyCompressor).
+            Object[][] rowsOnNewNode = cluster.coordinator(1).execute(
+            "SELECT count(*) FROM " + KEYSPACE + ".plugin_snappy", ConsistencyLevel.ONE);
+            assertEquals("The new node which does not have the plugin must read all rows written by the plugin node",
+                         100L, rowsOnNewNode[0][0]);
+
+        }
+    }
+
+    @Test
+    public void testCompressOn2NodesWithAndWithoutCustomProvider() throws Throwable
+    {
+        // This test will simulate a scenario with a cluster containing 2 nodes, one with custom provider configured and another with default provider.
+        // A table is created with Snappy compression on the first node with a custom provider, writes data and flushes to disk. Data is then read from both nodes
+        // to verify that both can read the data regardless of the provider they have configured.
+        Map<String, Object> snappyProviderOptionsCustom = new HashMap<>();
+        snappyProviderOptionsCustom.put("class_name", HealthyTestProvider.class.getName());
+        snappyProviderOptionsCustom.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> snappyProviderOptionsDefault = new HashMap<>();
+        snappyProviderOptionsDefault.put("class_name", DefaultCompressionProvider.class.getName());
+        snappyProviderOptionsDefault.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> firstNodeProvider = new HashMap<>();
+        firstNodeProvider.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptionsCustom);
+        Map<String, Object> secondNodeProvider = new HashMap<>();
+        secondNodeProvider.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptionsDefault);
+
+        try (Cluster cluster = init(Cluster.build(2)
+                                           .withConfig(config -> {
+                                               if (config.num() == 1)
+                                                   config.set("compressor_providers", firstNodeProvider);   // custom HealthyTestProvider
+                                               else
+                                                   config.set("compressor_providers", secondNodeProvider);  // DefaultCompressionProvider
+                                               config.set("accord.enabled", false); // disable accord to avoid schema disagreement during startup due to different compressor providers
+                                           })
+                                           .start()))
+        {
+            // First node must use HealthyTestProvider for SnappyCompressor,
+            // Second node must use DefaultCompressionProvider for SnappyCompressor
+            cluster.get(1).runOnInstance(() -> {
+                AbstractCompressionProvider snappy = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+                assertNotNull(snappy);
+                assertEquals(HealthyTestProvider.class.getName(), snappy.getClass().getName());
+            });
+            cluster.get(2).runOnInstance(() -> {
+                AbstractCompressionProvider snappy = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+                assertNotNull(snappy);
+                assertEquals(DefaultCompressionProvider.class.getName(), snappy.getClass().getName());
+            });
+
+
+            // Reset the counter so the assertion below only sees calls produced by this test's DDL/writes.
+            cluster.get(1).runOnInstance(() -> HealthyTestProvider.CREATE_CALLS.set(0));
+
+            // End-to-end: create a table that uses Snappy compression.
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".plugin_snappy (k int PRIMARY KEY, v text) " +
+                                 "WITH compression = {'class': 'SnappyCompressor', 'chunk_length_in_kb': '4'}");
+
+            for (int i = 0; i < 100; i++)
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".plugin_snappy (k, v) VALUES (?, ?)",
+                                               ConsistencyLevel.ALL, i, "value-" + i);
+
+            cluster.get(1).flush(KEYSPACE);
+            cluster.get(2).flush(KEYSPACE);
+
+            // Read from node 1
+            Object[][] rowsFromNode1 = cluster.coordinator(1).execute("SELECT count(*) FROM " + KEYSPACE + ".plugin_snappy",
+                                                                      ConsistencyLevel.ONE);
+            assertEquals(100L, rowsFromNode1[0][0]);
+
+            // Read from node 2
+            Object[][] rowsFromNode2 = cluster.coordinator(2).execute("SELECT count(*) FROM " + KEYSPACE + ".plugin_snappy",
+                                                                      ConsistencyLevel.ONE);
+            assertEquals(100L, rowsFromNode2[0][0]);
+
+            int creates = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + creates,
+                       creates > 0);
+
+            int createsSnappy = cluster.get(1).callOnInstance((SerializableCallable<Integer>) () -> HealthyTestProvider.CREATE_SNAPPY_CALLS.get());
+            assertTrue("HealthyTestProvider.createCompressor should have been invoked for table compression, was " + createsSnappy,
+                       createsSnappy > 0);
+
+            // The schema must record the *concrete* compressor class, not the provider — this is what
+            // peers and future restarts use to rebuild the compression chain.
+            Object[][] schemaRows = cluster.coordinator(1).execute(
+            "SELECT compression FROM system_schema.tables WHERE keyspace_name = ? AND table_name = ?",
+            ConsistencyLevel.ALL, KEYSPACE, "plugin_snappy");
+            @SuppressWarnings("unchecked")
+            Map<String, String> compressionOptions = (Map<String, String>) schemaRows[0][0];
+            assertEquals("system_schema.tables.compression must record the concrete compressor class, not the provider",
+                         SnappyCompressor.class.getName(), compressionOptions.get("class"));
+
+            // Reconstruct CompressionMetadata from the flushed SSTable on disk and check the
+            // rebuilt compressor's serializedAs()
+            String onDiskSerializedAsNode1 = cluster.get(1).callOnInstance((SerializableCallable<String>) () -> {
+                ColumnFamilyStore cfs = Keyspace.open(KEYSPACE)
+                                                .getColumnFamilyStore("plugin_snappy");
+                SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+                try (CompressionMetadata metadata = CompressionInfoComponent.loadIfExists(sstable.descriptor, null))
+                {
+                    Assert.assertNotNull(metadata);
+                    return metadata.parameters.getSstableCompressor().serializedAs().getName();
+                }
+            });
+            assertEquals("Rebuilt CompressionMetadata.serializedAs() must yield the underlying compressor - " +
+                         "what the plugin advertises for portability - not the wrapper",
+                         SnappyCompressor.class.getName(), onDiskSerializedAsNode1);
+
+            // Node 2 (DefaultCompressionProvider): also verify the on-disk SSTable is readable and
+            // records SnappyCompressor — confirming it can decompress without the plugin.
+            String onDiskSerializedAsNode2 = cluster.get(2).callOnInstance((SerializableCallable<String>) () -> {
+                ColumnFamilyStore cfs = Keyspace.open(KEYSPACE)
+                                                .getColumnFamilyStore("plugin_snappy");
+                SSTableReader sstable = cfs.getLiveSSTables().iterator().next();
+                try (CompressionMetadata metadata = CompressionInfoComponent.loadIfExists(sstable.descriptor, null))
+                {
+                    Assert.assertNotNull(metadata);
+                    return metadata.parameters.getSstableCompressor().serializedAs().getName();
+                }
+            });
+            assertEquals("Node 2 (DefaultCompressionProvider): on-disk serializedAs() must also yield the concrete compressor class",
+                         SnappyCompressor.class.getName(), onDiskSerializedAsNode2);
+        }
+
+    }
+
+    @Test
+    public void testUnhealthyProviderWithFallbackBootsWithDefault() throws Throwable
+    {
+        Map<String, Object> lz4ProviderOptions = new HashMap<>();
+        lz4ProviderOptions.put("class_name", UnhealthyTestProvider.class.getName());
+        lz4ProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString()));
+
+        Map<String, Object> providers = new HashMap<>();
+        providers.put(LZ4Compressor.class.getSimpleName(), lz4ProviderOptions);
+
+        try (Cluster cluster = init(Cluster.build(1)
+                                           .withConfig(config -> config.set("compressor_providers", providers))
+                                           .start()))
+        {
+            cluster.get(1).runOnInstance(() -> {
+                AbstractCompressionProvider lz4Provider = CompressorRegistry.instance.getProvider(LZ4Compressor.class);
+                assertNotNull(lz4Provider);
+                assertSame("Unhealthy provider with fallback=true must resolve to the default",
+                           DefaultCompressionProvider.class,
+                           lz4Provider.getClass());
+            });
+
+            // Sanity check: table-level LZ4 compression still works because the fallback default is registered.
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".fallback_lz4 (k int PRIMARY KEY, v text) " +
+                                 "WITH compression = {'class': 'LZ4Compressor'}");
+
+            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".fallback_lz4 (k, v) VALUES (1, 'ok')",
+                                           ConsistencyLevel.ONE);
+            cluster.get(1).flush(KEYSPACE);
+
+            Object[][] rows = cluster.coordinator(1).execute("SELECT v FROM " + KEYSPACE + ".fallback_lz4 WHERE k = 1",
+                                                             ConsistencyLevel.ONE);
+            assertEquals("ok", rows[0][0]);
+        }
+    }
+
+    @Test
+    public void testUnhealthyProviderWithoutFallbackFailsStartup() throws Throwable
+    {
+        Map<String, Object> lz4ProviderOptions = new HashMap<>();
+        lz4ProviderOptions.put("class_name", UnhealthyTestProvider.class.getName());
+        lz4ProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> providers = new HashMap<>();
+        providers.put(LZ4Compressor.class.getSimpleName(), lz4ProviderOptions);
+
+        try (Cluster cluster = Cluster.build(1)
+                                      .withConfig(config -> config.set("compressor_providers", providers))
+                                      .createWithoutStarting())
+        {
+            try
+            {
+                cluster.get(1).startup();
+                fail("Startup should have failed because the configured compressor provider is unhealthy and fallback is disabled");
+            }
+            catch (Throwable t)
+            {
+                Throwable cause = t;
+                String expected = "Failed to initialize compression provider";
+                while (cause != null && (cause.getMessage() == null || !cause.getMessage().contains(expected)))
+                    cause = cause.getCause();
+                assertNotNull("Expected ConfigurationException containing '" + expected + "' in the cause chain, got: " + t, cause);
+            }
+        }
+    }
+
+    @Test
+    public void testUnknownCompressorKeyFailsStartup() throws Throwable
+    {
+        // 'no_such_compressor' matches no built-in compressor (neither FQCN nor simple name nor
+        // abbreviation), so registerProviders must reject it at apply-time and the node must
+        // refuse to start with a ConfigurationException.
+        Map<String, Object> bogusProviderOptions = new HashMap<>();
+        bogusProviderOptions.put("class_name", HealthyTestProvider.class.getName());
+        bogusProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()));
+
+        Map<String, Object> providers = new HashMap<>();
+        providers.put("no_such_compressor", bogusProviderOptions);
+
+        try (Cluster cluster = Cluster.build(1)
+                                      .withConfig(config -> config.set("compressor_providers", providers))
+                                      .createWithoutStarting())
+        {
+            try
+            {
+                cluster.get(1).startup();
+                fail("Startup should have failed because compressor_providers contains a key that maps to no known compressor");
+            }
+            catch (Throwable t)
+            {
+                Throwable configException = t;
+                String expectedClass = "org.apache.cassandra.exceptions.ConfigurationException";
+                String expectedMessage = "Unknown compressor key 'no_such_compressor'";
+                while (configException != null
+                       && (!expectedClass.equals(configException.getClass().getName())
+                           || configException.getMessage() == null
+                           || !configException.getMessage().contains(expectedMessage)))
+                    configException = configException.getCause();
+                assertNotNull("Expected ConfigurationException with message containing '" + expectedMessage
+                              + "' in the cause chain, got: " + t, configException);
+            }
+        }
+    }
+
+    /**
+     * A custom provider that reports healthy and delegates compressor creation to the
+     * built-in default provider. Used to prove the registry actually invokes the plugin
+     * during table compression rather than silently using the default.
+     * <p>
+     * Must be public with a no-arg constructor so {@code FBUtilities.newCompressionProvider}
+     * can instantiate it via reflection.
+     */
+    public static class HealthyTestProvider extends AbstractCompressionProvider
+    {
+        // Statics live in the InstanceClassLoader where the provider is loaded. Always read and
+        // mutate these via runOnInstance / callOnInstance — touching them directly from the test
+        // thread reads a separate copy in the test classloader and silently returns zero.
+        public static final AtomicInteger CREATE_CALLS = new AtomicInteger();
+        public static final AtomicInteger CREATE_SNAPPY_CALLS = new AtomicInteger();
+
+        @Override
+        public boolean isHealthy()
+        {
+            return true;
+        }
+
+        @Override
+        public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options)
+        {
+            CREATE_CALLS.incrementAndGet();
+
+            // inject here our custom compressor, mimmicking pluggable providers
+            if (compressorClass == SnappyCompressor.class)
+            {
+                CREATE_SNAPPY_CALLS.incrementAndGet();
+                return new MySnappyCompressor();
+            }
+
+            return CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER.createCompressor(compressorClass, options);
+        }
+    }
+
+    public static class MySnappyCompressor extends SnappyCompressor
+    {
+        @Override
+        public Class<? extends ICompressor> serializedAs()
+        {
+            // Tell Cassandra to record SnappyCompressor in schema/SSTable metadata so that the
+            // schema is portable to peers and restarts without this plugin.
+            return SnappyCompressor.class;
+        }
+    }
+
+    public static class UnhealthyTestProvider extends AbstractCompressionProvider
+    {
+        @Override
+        public boolean isHealthy()
+        {
+            return false;
+        }
+
+        @Override
+        public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options)
+        {
+            throw new IllegalStateException("unhealthy provider must not be asked to create compressors");
+        }
+    }
+}

test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java

@@ -244,6 +244,10 @@ public class DatabaseDescriptorRefTest
     "org.apache.cassandra.io.FSError",
     "org.apache.cassandra.io.FSWriteError",
     "org.apache.cassandra.io.MessageVersionProvider",
+    "org.apache.cassandra.io.compress.AbstractCompressionProvider",
+    "org.apache.cassandra.io.compress.CompressorRegistry",
+    "org.apache.cassandra.io.compress.CompressorRegistry$CompressorType",
+    "org.apache.cassandra.io.compress.DefaultCompressionProvider",
     "org.apache.cassandra.io.compress.ICompressor",
     "org.apache.cassandra.io.compress.ICompressor$Uses",
     "org.apache.cassandra.io.compress.LZ4Compressor",

test/unit/org/apache/cassandra/io/compress/CompressorRegistryTest.java

@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.io.compress;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.CompressionProviderHelper.SerializedTestCompressor;
+import org.apache.cassandra.utils.CompressionProviderHelper.TestCompressionProvider;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType;
+
+public class CompressorRegistryTest
+{
+    private Config config;
+
+    @Before
+    public void setUp()
+    {
+        config = DatabaseDescriptor.loadConfig();
+    }
+
+    @After
+    public void tearDown()
+    {
+        CompressorRegistry.instance.reset();
+        config.compressor_providers.clear();
+        DatabaseDescriptor.setConfig(config);
+    }
+
+    @Test
+    public void testRegisterAndGetInbuiltCompressor()
+    {
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+
+        for (CompressorRegistry.CompressorType type : CompressorRegistry.CompressorType.values())
+        {
+            AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(type.compressorClass());
+            assertThat(provider).isNotNull();
+            assertThat(provider).isEqualTo(CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER);
+        }
+    }
+
+    @Test
+    public void testRegisterServiceProvider() throws Exception
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString());
+        config.compressor_providers.put(NoopCompressor.class.getSimpleName(), new ParameterizedClass(TestCompressionProvider.class.getName(), params));
+
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(NoopCompressor.class);
+        assertThat(provider).isNotNull();
+        assertThat(provider.isHealthy()).isTrue();
+        assertThat(provider.getClass()).isEqualTo(TestCompressionProvider.class);
+    }
+
+    @Test
+    public void testProviderConfigKeyVariants()
+    {
+        // The registry resolves a compressor type to its plugin via three accepted key forms in
+        // compressor_providers: fully qualified class name, simple class name, and the
+        // CompressorType abbreviation. Verify each form independently routes to the plugin.
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString());
+        String[] keys = {
+        NoopCompressor.class.getName(),                                  // FQCN
+        NoopCompressor.class.getSimpleName(),                            // simple name
+        CompressorRegistry.CompressorType.NOOP.abbreviation              // abbreviation
+        };
+
+        for (String key : keys)
+        {
+            CompressorRegistry.instance.reset();
+            config.compressor_providers.clear();
+            config.compressor_providers.put(key, new ParameterizedClass(TestCompressionProvider.class.getName(), params));
+
+            DatabaseDescriptor.setConfig(config);
+            DatabaseDescriptor.applyCompressorProviders();
+
+            AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(NoopCompressor.class);
+            assertThat(provider).as("provider resolved via key '%s'", key).isNotNull();
+            assertThat(provider.getClass()).as("provider class resolved via key '%s'", key)
+                                           .isEqualTo(TestCompressionProvider.class);
+        }
+    }
+
+    @Test
+    public void testUnhealthyProviderWithFallback()
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString(),
+                                            TestCompressionProvider.FAIL_HEALTH, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.resolveProvider(parameterizedClass);
+        assertThat(provider).isNotNull();
+        assertThat(provider.getClass()).isEqualTo(DefaultCompressionProvider.class);
+    }
+
+    @Test
+    public void testUnhealthyProviderWithoutFallback()
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString(),
+                                            TestCompressionProvider.FAIL_HEALTH, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.resolveProvider(parameterizedClass))
+        .withMessageContaining("Failed to initialize compression provider " + parameterizedClass);
+    }
+
+    @Test
+    public void testProviderHealthCheckExceptionWithFallback()
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString(),
+                                            TestCompressionProvider.FAIL_HEALTH_EXCEPTION, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.resolveProvider(parameterizedClass);
+        assertThat(provider).isNotNull();
+        assertThat(provider.getClass()).isEqualTo(DefaultCompressionProvider.class);
+    }
+
+    @Test
+    public void testProviderHealthCheckExceptionWithoutFallbackThrows()
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString(),
+                                            TestCompressionProvider.FAIL_HEALTH_EXCEPTION, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.resolveProvider(parameterizedClass))
+        .withMessageContaining("Failed to initialize compression provider " + parameterizedClass);
+    }
+
+    @Test
+    public void testProviderInstantiationFailureFallsBackToDefault()
+    {
+        // FBUtilities.newCompressionProvider throws ConfigurationException for an unknown class —
+        // this exercises the catch block in resolveProvider, distinct from the isHealthy()==false
+        // path. With fallback enabled, the default provider must be returned and the original
+        // exception's cause chain should be available in the warn log (not just its message).
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString());
+        ParameterizedClass parameterizedClass = new ParameterizedClass("org.apache.cassandra.does.not.Exist", params);
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.resolveProvider(parameterizedClass);
+        assertThat(provider).isEqualTo(CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER);
+    }
+
+    @Test
+    public void testNullParametersDoesNotNPE()
+    {
+        // ParameterizedClass.parameters is nullable: SnakeYAML uses the no-arg constructor and
+        // leaves the field null when the yaml entry has only 'class_name:' (no 'parameters:'
+        // block). resolveProvider must tolerate this rather than NPE on getOrDefault.
+        ParameterizedClass pc = new ParameterizedClass();
+        pc.class_name = "org.apache.cassandra.does.not.Exist";
+        pc.parameters = new HashMap<>();
+        pc.parameters.put(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString());
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.resolveProvider(pc);
+        assertThat(provider).isEqualTo(CompressorRegistry.DEFAULT_COMPRESSION_PROVIDER);
+    }
+
+    @Test
+    public void testMissingClassNameThrows()
+    {
+        // A compressor_providers entry with no 'class_name' is a config mistake and must fail
+        // loudly — silently substituting the default would mask typos like 'clas_name:'.
+        ParameterizedClass parameterizedClass = new ParameterizedClass(null,
+                                                                       Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString()));
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.resolveProvider(parameterizedClass))
+        .withMessageContaining("missing required 'class_name'");
+    }
+
+    @Test
+    public void testProviderInstantiationFailureWithoutFallbackThrows()
+    {
+        // Same catch block as above, but with fallback disabled — the resolver must surface a
+        // ConfigurationException rather than silently use the default.
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString());
+        ParameterizedClass parameterizedClass = new ParameterizedClass("org.apache.cassandra.does.not.Exist", params);
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.resolveProvider(parameterizedClass))
+        .withMessageContaining("Failed to initialize compression provider " + parameterizedClass);
+    }
+
+    @Test
+    public void testInitFailureWithFallback()
+    {
+        // An exception during provider.init() will be treated the same as an unhealthy provider:
+        // if fallback is enabled, the default provider will be used
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString(),
+                                            TestCompressionProvider.FAIL_INIT, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.resolveProvider(parameterizedClass);
+        assertThat(provider).isNotNull();
+        assertThat(provider.getClass()).isEqualTo(DefaultCompressionProvider.class);
+    }
+
+    @Test
+    public void testInitFailureWithoutFallbackThrows()
+    {
+        // If provider.init() fails and if fallback is disabled, an exception will be thrown
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString(),
+                                            TestCompressionProvider.FAIL_INIT, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.resolveProvider(parameterizedClass))
+        .withMessageContaining("Failed to initialize compression provider " + parameterizedClass);
+    }
+
+    @Test
+    public void testCreateCompressorFailureWithFallback()
+    {
+        // This tests a scenario when provider fails to get a compressor,
+        // if fallback is enabled, the registry should still return a compressor from the default provider instead of failing the request.
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString(),
+                                            TestCompressionProvider.FAIL_CREATE, "true");
+        config.compressor_providers.put(SnappyCompressor.class.getSimpleName(), new ParameterizedClass(TestCompressionProvider.class.getName(), params));
+
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+        ICompressor compressor = CompressorRegistry.instance.getCompressor(SnappyCompressor.class, Map.of());
+        assertThat(compressor).isInstanceOf(SnappyCompressor.class);
+    }
+
+    @Test
+    public void testCreateCompressorFailureWithoutFallbackThrows()
+    {
+        // if fallback is disabled and provider fails to get a compressor, the registry should throw an exception
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString(),
+                                            TestCompressionProvider.FAIL_CREATE, "true");
+        ParameterizedClass parameterizedClass = new ParameterizedClass(TestCompressionProvider.class.getName(), params);
+        config.compressor_providers.put(SnappyCompressor.class.getSimpleName(), parameterizedClass);
+
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+
+        assertThatExceptionOfType(IllegalStateException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.getCompressor(SnappyCompressor.class, Map.of()))
+        .withMessageContaining("compressor instantiation failed");
+    }
+
+    @Test
+    public void testCustomProviderAndCompressorWithSerializedAs() throws Exception
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString());
+        config.compressor_providers.put(SnappyCompressor.class.getSimpleName(), new ParameterizedClass(TestCompressionProvider.class.getName(), params));
+
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+        assertThat(provider).isNotNull();
+        assertThat(provider.isHealthy()).isTrue();
+        ICompressor compressor = CompressorRegistry.instance.getCompressor(SnappyCompressor.class, Map.of());
+        assertThat(compressor.getClass()).isEqualTo(SerializedTestCompressor.class);
+    }
+
+    @Test
+    public void testCustomProviderAndCompressorWithoutSerializedAs() throws Exception
+    {
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString(),
+                                            TestCompressionProvider.FAIL_SERIALIZED_AS, "true");
+        config.compressor_providers.put(SnappyCompressor.class.getSimpleName(), new ParameterizedClass(TestCompressionProvider.class.getName(), params));
+
+        DatabaseDescriptor.setConfig(config);
+        DatabaseDescriptor.applyCompressorProviders();
+
+        AbstractCompressionProvider provider = CompressorRegistry.instance.getProvider(SnappyCompressor.class);
+        assertThat(provider).isNotNull();
+        assertThat(provider.isHealthy()).isTrue();
+        assertThatExceptionOfType(RuntimeException.class)
+        .isThrownBy(() -> CompressorRegistry.instance.getCompressor(SnappyCompressor.class, Map.of()))
+        .withMessageContaining("You need to override serializedAs() method of your custom compressor and return " +
+                               "base compressor class it is the substitute for.");
+    }
+
+}

test/unit/org/apache/cassandra/schema/CompressionParamsTest.java

@@ -18,10 +18,20 @@
 
 package org.apache.cassandra.schema;
 
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableMap;
+
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.compress.ICompressor;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -71,4 +81,67 @@ public class CompressionParamsTest
         .as("Noop compression should not enable dictionary compression")
         .isFalse();
     }
+
+    @Test
+    public void testNonBuiltInCompressorClassResolvesViaDefaultProvider()
+    {
+        // A 3rd-party / user-supplied compressor class is not in CompressorType, so the registry
+        // has no entry for it. CompressionParams.createCompressor must fall back to the default
+        // provider, which reflectively invokes the class's static create(Map) factory - otherwise
+        // every existing pluggable compressor outside org.apache.cassandra.io.compress NPEs.
+        Map<String, String> opts = ImmutableMap.of("class", CustomTestCompressor.class.getName(),
+                                                   "chunk_length_in_kb", "8");
+        CompressionParams params = CompressionParams.fromMap(opts);
+
+        assertThat(params.getSstableCompressor()).isInstanceOf(CustomTestCompressor.class);
+        assertThat(params.klass()).isEqualTo(CustomTestCompressor.class);
+    }
+
+    public static class CustomTestCompressor implements ICompressor
+    {
+        public static CustomTestCompressor create(Map<String, String> options)
+        {
+            return new CustomTestCompressor();
+        }
+
+        @Override
+        public int initialCompressedBufferLength(int chunkLength)
+        {
+            return chunkLength;
+        }
+
+        @Override
+        public int uncompress(byte[] in, int io, int il, byte[] out, int oo) throws IOException
+        {
+            return 0;
+        }
+
+        @Override
+        public void compress(ByteBuffer in, ByteBuffer out) throws IOException
+        {
+        }
+
+        @Override
+        public void uncompress(ByteBuffer in, ByteBuffer out) throws IOException
+        {
+        }
+
+        @Override
+        public BufferType preferredBufferType()
+        {
+            return BufferType.OFF_HEAP;
+        }
+
+        @Override
+        public boolean supports(BufferType bufferType)
+        {
+            return true;
+        }
+
+        @Override
+        public Set<String> supportedOptions()
+        {
+            return Collections.emptySet();
+        }
+    }
 }

test/unit/org/apache/cassandra/service/StartupChecksTest.java

@@ -53,12 +53,16 @@ import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.config.Config.DiskAccessMode;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.config.StartupChecksConfiguration;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.StartupException;
+import org.apache.cassandra.io.compress.AbstractCompressionProvider;
+import org.apache.cassandra.io.compress.CompressorRegistry;
+import org.apache.cassandra.io.compress.SnappyCompressor;
 import org.apache.cassandra.io.filesystem.ForwardingFileSystem;
 import org.apache.cassandra.io.filesystem.ForwardingFileSystemProvider;
 import org.apache.cassandra.io.filesystem.ForwardingPath;
@@ -66,6 +70,8 @@ import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.service.DataResurrectionCheck.Heartbeat;
 import org.apache.cassandra.utils.Clock;
+import org.apache.cassandra.utils.CompressionProviderHelper.CompatibleSnappyProvider;
+import org.apache.cassandra.utils.CompressionProviderHelper.IncompatibleSnappyProvider;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.SystemInfo;
 
@@ -75,6 +81,7 @@ import static org.apache.cassandra.io.util.FileUtils.createTempFile;
 import static org.apache.cassandra.service.DataResurrectionCheck.HEARTBEAT_FILE_CONFIG_PROPERTY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
@@ -151,7 +158,8 @@ public class StartupChecksTest
         new File(sstableDir).deleteRecursive();
         Path snapshotDir = sstableDir.resolve("snapshots");
         Files.createDirectories(snapshotDir);
-        copyInvalidLegacySSTables(snapshotDir); startupChecks.verify(options);
+        copyInvalidLegacySSTables(snapshotDir);
+        startupChecks.verify(options);
 
         // and in a backups directory
         new File(sstableDir).deleteRecursive();
@@ -218,16 +226,17 @@ public class StartupChecksTest
         Path legacySSTableRoot = Paths.get(TEST_INVALID_LEGACY_SSTABLE_ROOT.getString(),
                                            "Keyspace1",
                                            "Standard1");
-        for (String filename : new String[]{"Keyspace1-Standard1-ic-0-TOC.txt",
+        for (String filename : new String[]{ "Keyspace1-Standard1-ic-0-TOC.txt",
                                              "Keyspace1-Standard1-ic-0-Digest.sha1",
-                                            "legacyleveled.json"})
+                                             "legacyleveled.json" })
             Files.copy(Paths.get(legacySSTableRoot.toString(), filename), targetDir.resolve(filename));
     }
 
     @Test
     public void testDataResurrectionCheck() throws Exception
     {
-        DataResurrectionCheck check = new DataResurrectionCheck() {
+        DataResurrectionCheck check = new DataResurrectionCheck()
+        {
             @Override
             List<String> getKeyspaces()
             {
@@ -254,7 +263,8 @@ public class StartupChecksTest
     @Test
     public void testDataResurrectionCheckLastModifiedFallback() throws Exception
     {
-        DataResurrectionCheck check = new DataResurrectionCheck() {
+        DataResurrectionCheck check = new DataResurrectionCheck()
+        {
             @Override
             List<String> getKeyspaces()
             {
@@ -287,7 +297,8 @@ public class StartupChecksTest
         }
     }
 
-    private void verifySuccess(StartupChecks tests) {
+    private void verifySuccess(StartupChecks tests)
+    {
         try
         {
             tests.verify(options);
@@ -376,7 +387,8 @@ public class StartupChecksTest
 
         ServiceLoader<StartupCheck> loader = mock(ServiceLoader.class);
         doReturn(List.of(externalCheck).iterator()).when(loader).iterator();
-        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class)) {
+        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class))
+        {
             serviceLoader.when(() -> ServiceLoader.load(StartupCheck.class)).thenReturn(loader);
 
             StartupChecks checks = new StartupChecks().withDefaultTests().withServiceLoaderTests();
@@ -426,7 +438,8 @@ public class StartupChecksTest
         // two times! We model loading of two checks with same name
         doReturn(List.of(externalCheck, externalCheck).iterator()).when(loader).iterator();
 
-        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class)) {
+        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class))
+        {
             serviceLoader.when(() -> ServiceLoader.load(StartupCheck.class)).thenReturn(loader);
 
             try
@@ -478,7 +491,8 @@ public class StartupChecksTest
         // two times! We model loading of two checks with same name
         doReturn(List.of(externalCheck, externalCheck).iterator()).when(loader).iterator();
 
-        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class)) {
+        try (MockedStatic<ServiceLoader> serviceLoader = Mockito.mockStatic(ServiceLoader.class))
+        {
             serviceLoader.when(() -> ServiceLoader.load(StartupCheck.class)).thenReturn(loader);
 
             try
@@ -594,7 +608,6 @@ public class StartupChecksTest
                                                     "Standard1"));
         for (File f : legacySSTableRoot.tryList())
             Files.copy(f.toPath(), targetDir.resolve(f.name()));
-
     }
 
     private void verifyFailure(StartupChecks tests, String message)
@@ -615,7 +628,60 @@ public class StartupChecksTest
     {
         // Non-existent directories should be skipped, not added to unsupported list
         List<String> unsupported = StartupChecks.findDirectIOUnsupportedLocations(
-            new String[] { "/this/path/does/not/exist/for/testing" });
+        new String[]{ "/this/path/does/not/exist/for/testing" });
         assertThat(unsupported).isEmpty();
     }
+
+    @Test
+    public void testCompatibleCompressionProvider()
+    {
+        // This test will go through the smoke test verifing with a valid custom provider,
+        // providing a compressor compatible with Snappy
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString());
+        Map<String, ParameterizedClass> providerOptions = Map.of(
+        SnappyCompressor.class.getSimpleName(),
+        new ParameterizedClass(CompatibleSnappyProvider.class.getName(), params));
+
+        CompressorRegistry.instance.reset();
+        CompressorRegistry.instance.registerProviders(providerOptions);
+        try
+        {
+            StartupChecks.checkCustomCompressionProviders.execute(options);
+        }
+        catch (Throwable t)
+        {
+            fail("This exception should not be thrown since the provider is compatible " +
+                 "with the compressor it is supposed to create: " + t.getMessage());
+        }
+        finally
+        {
+            CompressorRegistry.instance.reset(); // only registry state touched — safe to reset
+        }
+    }
+
+    @Test
+    public void testIncompatibleCompressionProvider()
+    {
+        // This test is trying to simulate a failure scenario by providing a custom provider which is not compatible
+        // with the compressor it is supposed to create. In this case, we are providing a compressor which is compatible
+        // with Snappy, but we are registering it as provider for NoopCompressor.
+        // The smoke test should fail and throw an exception alerting that using this provider may lead to data corruption.
+        Map<String, String> params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString());
+        Map<String, ParameterizedClass> providerOptions = Map.of(
+        SnappyCompressor.class.getSimpleName(),
+        new ParameterizedClass(IncompatibleSnappyProvider.class.getName(), params));
+
+        CompressorRegistry.instance.reset();
+        CompressorRegistry.instance.registerProviders(providerOptions);
+        try
+        {
+            assertThatThrownBy(() -> StartupChecks.checkCustomCompressionProviders.execute(options))
+            .isInstanceOf(StartupException.class)
+            .hasMessageContaining("The following custom compression providers failed smoke test");
+        }
+        finally
+        {
+            CompressorRegistry.instance.reset();
+        }
+    }
 }

test/unit/org/apache/cassandra/utils/CompressionProviderHelper.java

@@ -0,0 +1,226 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.cassandra.io.compress.AbstractCompressionProvider;
+import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.compress.ICompressor;
+import org.apache.cassandra.io.compress.SnappyCompressor;
+
+/**
+ * Shared test compression provider and compressor tests.
+ * Provides reusable providers and compressors used across
+ * {@code CompressorRegistryTest}, {@code StartupChecksTest}, and any future tests
+ * that exercise the compression provider interface.
+ */
+public final class CompressionProviderHelper
+{
+    private CompressionProviderHelper() {}
+
+    // -------------------------------------------------------------------------
+    // Providers
+    // -------------------------------------------------------------------------
+
+    // Test compression provider for testing various edge cases around initialization, health checks, and
+    // compressor creation.
+    public static class TestCompressionProvider extends AbstractCompressionProvider
+    {
+        // ParameterizedClass.parameters are used to pass different test scenarios .
+        // resolveProvider() only strips FAIL_ON_MISSING_PROVIDER, so get to init().
+        public static final String FAIL_INIT             = "fail_init";
+        public static final String FAIL_HEALTH           = "fail_health";
+        public static final String FAIL_HEALTH_EXCEPTION = "fail_health_exception";
+        public static final String FAIL_CREATE           = "fail_create";
+        public static final String FAIL_SERIALIZED_AS    = "fail_serialized_as";
+
+        private boolean flag(String key)
+        {
+            return Boolean.parseBoolean(getParameters().getOrDefault(key, "false"));
+        }
+
+        @Override
+        public void init(Map<String, String> parameters)
+        {
+            super.init(parameters);          //use this to store and provide test variants
+            if (flag(FAIL_INIT))
+                throw new RuntimeException("init failed: something went wrong with startup");
+        }
+
+        @Override
+        public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options) throws IllegalStateException
+        {
+            if (flag(FAIL_CREATE))
+            {
+                throw new IllegalStateException("compressor instantiation failed");
+            }
+            return flag(FAIL_SERIALIZED_AS)
+                ? PlainTestCompressor.create(options)
+                : SerializedTestCompressor.create(options);
+        }
+
+        @Override
+        public boolean isHealthy()
+        {
+            if (flag(FAIL_HEALTH_EXCEPTION))
+                throw new RuntimeException("health check failed");
+            if (flag(FAIL_HEALTH)) return false;
+            return true;
+        }
+    }
+
+    public static class CompatibleSnappyProvider extends AbstractCompressionProvider
+    {
+        @Override
+        public boolean isHealthy()
+        {
+            return true;
+        }
+
+        @Override
+        public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options)
+        {
+            return new MySnappyCompressor();
+        }
+    }
+
+    public static class IncompatibleSnappyProvider extends AbstractCompressionProvider
+    {
+        @Override
+        public boolean isHealthy()
+        {
+            return true;
+        }
+
+        @Override
+        public ICompressor createCompressor(Class<?> compressorClass, Map<String, String> options)
+        {
+            return new IncompatibleSnappyCompressor();
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // Compressors
+    // -------------------------------------------------------------------------
+
+    abstract static class CustomTestCompressor implements ICompressor
+    {
+        @Override
+        public int initialCompressedBufferLength(int chunkLength)
+        {
+            return chunkLength;
+        }
+
+        @Override
+        public int uncompress(byte[] in, int io, int il, byte[] out, int oo) throws IOException
+        {
+            return 0;
+        }
+
+        @Override
+        public void compress(ByteBuffer in, ByteBuffer out) throws IOException
+        {
+        }
+
+        @Override
+        public void uncompress(ByteBuffer in, ByteBuffer out) throws IOException
+        {
+        }
+
+        @Override
+        public BufferType preferredBufferType()
+        {
+            return BufferType.OFF_HEAP;
+        }
+
+        @Override
+        public boolean supports(BufferType bufferType)
+        {
+            return true;
+        }
+
+        @Override
+        public Set<String> supportedOptions()
+        {
+            return Collections.emptySet();
+        }
+    }
+
+    // This test compressor does not override serializedAs(). serialized call should return PlainTestCompressor.class
+    public static class PlainTestCompressor extends CustomTestCompressor
+    {
+        public static PlainTestCompressor create(Map<String, String> options)
+        {
+            return new PlainTestCompressor();
+        }
+    }
+
+    // This test compressor overrides serializedAs(), it is compatible with SnappyCompressor
+    public static class SerializedTestCompressor extends CustomTestCompressor
+    {
+        public static SerializedTestCompressor create(Map<String, String> options)
+        {
+            return new SerializedTestCompressor();
+        }
+
+        @Override
+        public Class<? extends ICompressor> serializedAs()
+        {
+            return SnappyCompressor.class;
+        }
+    }
+
+   // This test compressor overrides serializedAs(), this will create a compressor compatible with SnappyCompressor
+    public static class MySnappyCompressor extends SnappyCompressor
+    {
+        public static MySnappyCompressor create(Map<String, String> options)
+        {
+            return new MySnappyCompressor();
+        }
+
+        @Override
+        public Class<? extends ICompressor> serializedAs()
+        {
+            return SnappyCompressor.class;
+        }
+    }
+
+    // This compressor also is compatible with Snappy but will produce data which is not compatible
+    public static class IncompatibleSnappyCompressor extends SnappyCompressor
+    {
+        @Override
+        public Class<? extends ICompressor> serializedAs()
+        {
+            return SnappyCompressor.class;
+        }
+        @Override
+        public void compress(ByteBuffer src, ByteBuffer dest)
+        {
+            // Write raw uncompressed bytes instead of a valid Snappy stream.
+            // Snappy will not be able to decompress
+            // This is what the smoke test's step 1 cross-check is designed to catch.
+            dest.put(src);
+        }
+    }
+}