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 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 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 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 parameters = Collections.emptyMap(); + + /** + * Checks if this compression provider is in a healthy state and ready to use. + * + *

This method should perform any necessary health checks, such as verifying + * that required native libraries are loaded, able to do compress/decompress successfully, etc.

+ * + * @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. + * + *

This method is called when a new compressor is needed. The implementation + * should create and configure a compressor based on the provided options.

+ * + * @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 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. + *

+ * 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. + *

+ * 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 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 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 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 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. + *

+ * 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. + *

+ * 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, 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 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 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. + *

+ * 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 providerOptions) + { + if (providerOptions == null) + return; + + Set 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 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 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 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. + *

+ * 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 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 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 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 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 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, 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 failedProviders = new ArrayList<>(); + for (Map.Entry, 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, AbstractCompressionProvider> getCustomProviders() throws StartupException + { + Map, 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. + *

+ * 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 snappyProviderOptions = new HashMap<>(); + snappyProviderOptions.put("class_name", HealthyTestProvider.class.getName()); + snappyProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map 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) () -> 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) () -> 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 compressionOptions = (Map) 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) () -> { + 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 snappyProviderOptions = new HashMap<>(); + snappyProviderOptions.put("class_name", HealthyTestProvider.class.getName()); + snappyProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map 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) () -> 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) () -> 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 compressionOptions = (Map) 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) () -> { + 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 snappyProviderOptionsCustom = new HashMap<>(); + snappyProviderOptionsCustom.put("class_name", HealthyTestProvider.class.getName()); + snappyProviderOptionsCustom.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map snappyProviderOptionsDefault = new HashMap<>(); + snappyProviderOptionsDefault.put("class_name", DefaultCompressionProvider.class.getName()); + snappyProviderOptionsDefault.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map firstNodeProvider = new HashMap<>(); + firstNodeProvider.put(SnappyCompressor.class.getSimpleName(), snappyProviderOptionsCustom); + Map 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) () -> 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) () -> 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 compressionOptions = (Map) 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) () -> { + 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) () -> { + 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 lz4ProviderOptions = new HashMap<>(); + lz4ProviderOptions.put("class_name", UnhealthyTestProvider.class.getName()); + lz4ProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.FALSE.toString())); + + Map 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 lz4ProviderOptions = new HashMap<>(); + lz4ProviderOptions.put("class_name", UnhealthyTestProvider.class.getName()); + lz4ProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map 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 bogusProviderOptions = new HashMap<>(); + bogusProviderOptions.put("class_name", HealthyTestProvider.class.getName()); + bogusProviderOptions.put("parameters", Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString())); + + Map 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. + *

+ * 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 getKeyspaces() { @@ -254,7 +263,8 @@ public class StartupChecksTest @Test public void testDataResurrectionCheckLastModifiedFallback() throws Exception { - DataResurrectionCheck check = new DataResurrectionCheck() { + DataResurrectionCheck check = new DataResurrectionCheck() + { @Override List 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 loader = mock(ServiceLoader.class); doReturn(List.of(externalCheck).iterator()).when(loader).iterator(); - try (MockedStatic serviceLoader = Mockito.mockStatic(ServiceLoader.class)) { + try (MockedStatic 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 = Mockito.mockStatic(ServiceLoader.class)) { + try (MockedStatic 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 = Mockito.mockStatic(ServiceLoader.class)) { + try (MockedStatic 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 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 params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()); + Map 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 params = Map.of(AbstractCompressionProvider.FAIL_ON_MISSING_PROVIDER, Boolean.TRUE.toString()); + Map 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 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 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 options) + { + return new MySnappyCompressor(); + } + } + + public static class IncompatibleSnappyProvider extends AbstractCompressionProvider + { + @Override + public boolean isHealthy() + { + return true; + } + + @Override + public ICompressor createCompressor(Class compressorClass, Map 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 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 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 options) + { + return new SerializedTestCompressor(); + } + + @Override + public Class 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 options) + { + return new MySnappyCompressor(); + } + + @Override + public Class 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 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); + } + } +}