This is an automated email from the ASF dual-hosted git repository.
smiklosovic pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new c68dd59d4a Implementation of CEP-49: Hardware-accelerated compression
c68dd59d4a is described below
commit c68dd59d4aab78598d97aab0be08b723ffd53bb3
Author: Kokoori, Shylaja <[email protected]>
AuthorDate: Thu Dec 4 09:41:09 2025 -0800
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 <[email protected]>
---
CHANGES.txt | 1 +
NEWS.txt | 4 +
conf/cassandra.yaml | 18 +
conf/cassandra_latest.yaml | 18 +
.../pages/managing/operating/compression.adoc | 47 ++
src/java/org/apache/cassandra/config/Config.java | 1 +
.../cassandra/config/DatabaseDescriptor.java | 16 +
.../io/compress/AbstractCompressionProvider.java | 95 ++++
.../cassandra/io/compress/CompressionMetadata.java | 2 +-
.../cassandra/io/compress/CompressorRegistry.java | 305 ++++++++++++
.../io/compress/DefaultCompressionProvider.java | 106 ++++
.../apache/cassandra/io/compress/ICompressor.java | 15 +
.../apache/cassandra/schema/CompressionParams.java | 53 +-
.../apache/cassandra/service/StartupChecks.java | 158 ++++++
.../org/apache/cassandra/utils/FBUtilities.java | 21 +
.../distributed/test/CompressorProviderTest.java | 552 +++++++++++++++++++++
.../config/DatabaseDescriptorRefTest.java | 4 +
.../io/compress/CompressorRegistryTest.java | 311 ++++++++++++
.../cassandra/schema/CompressionParamsTest.java | 73 +++
.../cassandra/service/StartupChecksTest.java | 90 +++-
.../cassandra/utils/CompressionProviderHelper.java | 226 +++++++++
21 files changed, 2057 insertions(+), 59 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 87ee90e3a7..73e8de6c7e 100644
--- a/CHANGES.txt
+++ b/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)
diff --git a/NEWS.txt b/NEWS.txt
index d9207d89a6..028266355f 100644
--- a/NEWS.txt
+++ b/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
---------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 422ad0a482..0420e1c9ec 100644
--- a/conf/cassandra.yaml
+++ b/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
diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml
index ee68c70082..de6d872718 100644
--- a/conf/cassandra_latest.yaml
+++ b/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
diff --git a/doc/modules/cassandra/pages/managing/operating/compression.adoc
b/doc/modules/cassandra/pages/managing/operating/compression.adoc
index 134c256934..04ab4d13eb 100644
--- a/doc/modules/cassandra/pages/managing/operating/compression.adoc
+++ b/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
diff --git a/src/java/org/apache/cassandra/config/Config.java
b/src/java/org/apache/cassandra/config/Config.java
index 7cac3e8ed7..d6f68cbe1f 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/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");
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index e03f13723f..a1ef3fb789 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/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.
*/
diff --git
a/src/java/org/apache/cassandra/io/compress/AbstractCompressionProvider.java
b/src/java/org/apache/cassandra/io/compress/AbstractCompressionProvider.java
new file mode 100644
index 0000000000..2194b018ca
--- /dev/null
+++ b/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()));
+ }
+}
diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
index 2370d25ef2..ab2ae87024 100644
--- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
+++ b/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())
{
diff --git a/src/java/org/apache/cassandra/io/compress/CompressorRegistry.java
b/src/java/org/apache/cassandra/io/compress/CompressorRegistry.java
new file mode 100644
index 0000000000..5267d691d2
--- /dev/null
+++ b/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);
+ }
+}
diff --git
a/src/java/org/apache/cassandra/io/compress/DefaultCompressionProvider.java
b/src/java/org/apache/cassandra/io/compress/DefaultCompressionProvider.java
new file mode 100644
index 0000000000..8ac09d4dee
--- /dev/null
+++ b/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());
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/io/compress/ICompressor.java
b/src/java/org/apache/cassandra/io/compress/ICompressor.java
index 950ae03e3d..88b1a06eb0 100644
--- a/src/java/org/apache/cassandra/io/compress/ICompressor.java
+++ b/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();
+ }
}
diff --git a/src/java/org/apache/cassandra/schema/CompressionParams.java
b/src/java/org/apache/cassandra/schema/CompressionParams.java
index 0992f118ff..905e6aa039 100644
--- a/src/java/org/apache/cassandra/schema/CompressionParams.java
+++ b/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
- {
- 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());
- }
+ return registry.getCompressor(compressorClass, compressionOptions);
}
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())
{
diff --git a/src/java/org/apache/cassandra/service/StartupChecks.java
b/src/java/org/apache/cassandra/service/StartupChecks.java
index 86ab41688e..97a18da636 100644
--- a/src/java/org/apache/cassandra/service/StartupChecks.java
+++ b/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
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java
b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 924dcd6118..a0ec48390c 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/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.
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/CompressorProviderTest.java
b/test/distributed/org/apache/cassandra/distributed/test/CompressorProviderTest.java
new file mode 100644
index 0000000000..cb755a834c
--- /dev/null
+++
b/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");
+ }
+ }
+}
diff --git
a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
index 33b5c6f513..1704936544 100644
--- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
+++ b/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",
diff --git
a/test/unit/org/apache/cassandra/io/compress/CompressorRegistryTest.java
b/test/unit/org/apache/cassandra/io/compress/CompressorRegistryTest.java
new file mode 100644
index 0000000000..c043a0a3b6
--- /dev/null
+++ b/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.");
+ }
+
+}
diff --git a/test/unit/org/apache/cassandra/schema/CompressionParamsTest.java
b/test/unit/org/apache/cassandra/schema/CompressionParamsTest.java
index 70f1fbbc19..05b98e86aa 100644
--- a/test/unit/org/apache/cassandra/schema/CompressionParamsTest.java
+++ b/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();
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/service/StartupChecksTest.java
b/test/unit/org/apache/cassandra/service/StartupChecksTest.java
index 34fc287643..8ce618b094 100644
--- a/test/unit/org/apache/cassandra/service/StartupChecksTest.java
+++ b/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",
-
"Keyspace1-Standard1-ic-0-Digest.sha1",
- "legacyleveled.json"})
+ for (String filename : new String[]{
"Keyspace1-Standard1-ic-0-TOC.txt",
+
"Keyspace1-Standard1-ic-0-Digest.sha1",
+ "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();
+ }
+ }
}
diff --git
a/test/unit/org/apache/cassandra/utils/CompressionProviderHelper.java
b/test/unit/org/apache/cassandra/utils/CompressionProviderHelper.java
new file mode 100644
index 0000000000..5a4b99afd0
--- /dev/null
+++ b/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);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
