Merge branch 'cassandra-5.0' into cassandra-6.0

CHANGES.txt

@@ -76,6 +76,7 @@ Merged from 4.1:
  * Add Paxos v2 option and informatin in cassandra.yaml (CASSANDRA-21316)
  * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290)
 Merged from 4.0:
+ * Verify extension type before initializing reflectively-loaded classes (CASSANDRA-21525)
  * Rename conflicting nodetool import --copy-data short option from -p to -cd (CASSANDRA-20214)
  * Fix PasswordObfuscator failing to obfuscate certain passwords (CASSANDRA-21113)
  * Fix negative memtable allocator ownership when an update is shadowed by an existing row deletion (CASSANDRA-21469)

src/java/org/apache/cassandra/auth/AuthConfig.java

@@ -63,7 +63,7 @@ public final class AuthConfig
 
         /* Authentication, authorization and role management backend, implementing IAuthenticator, I*Authorizer & IRoleManager */
 
-        IAuthenticator authenticator = authInstantiate(conf.authenticator, AllowAllAuthenticator.class);
+        IAuthenticator authenticator = authInstantiate(conf.authenticator, IAuthenticator.class, AllowAllAuthenticator.class);
 
         // the configuration options regarding credentials caching are only guaranteed to
         // work with PasswordAuthenticator, so log a message if some other authenticator
@@ -82,7 +82,7 @@ public final class AuthConfig
 
         // authorizer
 
-        IAuthorizer authorizer = authInstantiate(conf.authorizer, AllowAllAuthorizer.class);
+        IAuthorizer authorizer = authInstantiate(conf.authorizer, IAuthorizer.class, AllowAllAuthorizer.class);
 
         if (!authenticator.requireAuthentication() && authorizer.requireAuthorization())
         {
@@ -94,7 +94,7 @@ public final class AuthConfig
 
         // role manager
 
-        IRoleManager roleManager = authInstantiate(conf.role_manager, CassandraRoleManager.class);
+        IRoleManager roleManager = authInstantiate(conf.role_manager, IRoleManager.class, CassandraRoleManager.class);
 
         if (authenticator instanceof PasswordAuthenticator && !(roleManager instanceof CassandraRoleManager))
             throw new ConfigurationException(authenticator.getClass().getName() + " requires " + CassandraRoleManager.class.getName(), false);
@@ -104,12 +104,13 @@ public final class AuthConfig
         // authenticator
 
         IInternodeAuthenticator internodeAuthenticator = authInstantiate(conf.internode_authenticator,
+                                                                         IInternodeAuthenticator.class,
                                                                          AllowAllInternodeAuthenticator.class);
         DatabaseDescriptor.setInternodeAuthenticator(internodeAuthenticator);
 
         // network authorizer
 
-        INetworkAuthorizer networkAuthorizer = authInstantiate(conf.network_authorizer, AllowAllNetworkAuthorizer.class);
+        INetworkAuthorizer networkAuthorizer = authInstantiate(conf.network_authorizer, INetworkAuthorizer.class, AllowAllNetworkAuthorizer.class);
 
         if (networkAuthorizer.requireAuthorization() && !authenticator.requireAuthentication())
         {
@@ -120,7 +121,7 @@ public final class AuthConfig
 
         // cidr authorizer
 
-        ICIDRAuthorizer cidrAuthorizer = authInstantiate(conf.cidr_authorizer, AllowAllCIDRAuthorizer.class);
+        ICIDRAuthorizer cidrAuthorizer = authInstantiate(conf.cidr_authorizer, ICIDRAuthorizer.class, AllowAllCIDRAuthorizer.class);
 
         if (cidrAuthorizer.requireAuthorization() && !authenticator.requireAuthentication())
         {
@@ -140,11 +141,11 @@ public final class AuthConfig
         DatabaseDescriptor.getInternodeAuthenticator().validateConfiguration();
     }
 
-    private static <T> T authInstantiate(ParameterizedClass authCls, Class<T> defaultCls) {
+    private static <T> T authInstantiate(ParameterizedClass authCls, Class<T> expectedType, Class<? extends T> defaultCls) {
         if (authCls != null && authCls.class_name != null)
         {
             String authPackage = AuthConfig.class.getPackage().getName();
-            return ParameterizedClass.newInstance(authCls, List.of("", authPackage));
+            return ParameterizedClass.newInstance(authCls, List.of("", authPackage), expectedType);
         }
 
         // for now, this has to stay and can not be replaced by ParameterizedClass.newInstance as above

src/java/org/apache/cassandra/auth/MutualTlsAuthenticator.java

@@ -101,7 +101,8 @@ public class MutualTlsAuthenticator implements IAuthenticator
             throw new ConfigurationException(message);
         }
         certificateValidator = ParameterizedClass.newInstance(new ParameterizedClass(certificateValidatorClassName),
-                                                              Arrays.asList("", AuthConfig.class.getPackage().getName()));
+                                                              Arrays.asList("", AuthConfig.class.getPackage().getName()),
+                                                              MutualTlsCertificateValidator.class);
 
         Config config = DatabaseDescriptor.getRawConfig();
         certificateValidityPeriodValidator = new MutualTlsCertificateValidityPeriodValidator(config.client_encryption_options.max_certificate_validity_period);

src/java/org/apache/cassandra/auth/MutualTlsInternodeAuthenticator.java

@@ -104,7 +104,8 @@ public class MutualTlsInternodeAuthenticator implements IInternodeAuthenticator
         }
 
         certificateValidator = ParameterizedClass.newInstance(new ParameterizedClass(certificateValidatorClassName),
-                                                              Arrays.asList("", AuthConfig.class.getPackage().getName()));
+                                                              Arrays.asList("", AuthConfig.class.getPackage().getName()),
+                                                              MutualTlsCertificateValidator.class);
         Config config = DatabaseDescriptor.getRawConfig();
 
         if (parameters.containsKey(TRUSTED_PEER_IDENTITIES))

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

@@ -511,7 +511,7 @@ public class DatabaseDescriptor
         String loaderClass = CONFIG_LOADER.getString();
         ConfigurationLoader loader = loaderClass == null
                                      ? new YamlConfigurationLoader()
-                                     : FBUtilities.construct(loaderClass, "configuration loading");
+                                     : FBUtilities.construct(loaderClass, "configuration loading", ConfigurationLoader.class);
         Config config = loader.loadConfig();
 
         if (!hasLoggedConfig)
@@ -1644,7 +1644,8 @@ public class DatabaseDescriptor
         }
         try
         {
-            Class<?> seedProviderClass = Class.forName(conf.seed_provider.class_name);
+            Class<? extends SeedProvider> seedProviderClass =
+                FBUtilities.classForNameWithoutInitialization(conf.seed_provider.class_name, "seed provider", SeedProvider.class);
             seedProvider = (SeedProvider) seedProviderClass.getConstructor(Map.class).newInstance(conf.seed_provider.parameters);
         }
         // there are about 5 checked exceptions that could be thrown here.
@@ -2021,7 +2022,7 @@ public class DatabaseDescriptor
     {
         if (!snitchClassName.contains("."))
             snitchClassName = "org.apache.cassandra.locator." + snitchClassName;
-        IEndpointSnitch snitch = FBUtilities.construct(snitchClassName, "snitch");
+        IEndpointSnitch snitch = FBUtilities.construct(snitchClassName, "snitch", IEndpointSnitch.class);
         return snitch;
     }
 
@@ -2029,7 +2030,7 @@ public class DatabaseDescriptor
     {
         if (!className.contains("."))
             className = "org.apache.cassandra.locator." + className;
-        NodeProximity sorter = FBUtilities.construct(className, "node proximity measurement");
+        NodeProximity sorter = FBUtilities.construct(className, "node proximity measurement", NodeProximity.class);
         return sorter;
     }
 
@@ -2037,7 +2038,7 @@ public class DatabaseDescriptor
     {
         if (!className.contains("."))
             className = "org.apache.cassandra.locator." + className;
-        InitialLocationProvider provider = FBUtilities.construct(className, "initial location provider");
+        InitialLocationProvider provider = FBUtilities.construct(className, "initial location provider", InitialLocationProvider.class);
         return provider;
     }
 
@@ -2045,7 +2046,7 @@ public class DatabaseDescriptor
     {
         if (!className.contains("."))
             className = "org.apache.cassandra.locator." + className;
-        NodeAddressConfig config = FBUtilities.construct(className, "node address config");
+        NodeAddressConfig config = FBUtilities.construct(className, "node address config", NodeAddressConfig.class);
         return config;
     }
 
@@ -2053,7 +2054,7 @@ public class DatabaseDescriptor
     {
         if (!detectorClassName.contains("."))
             detectorClassName = "org.apache.cassandra.gms." + detectorClassName;
-        IFailureDetector detector = FBUtilities.construct(detectorClassName, "failure detector");
+        IFailureDetector detector = FBUtilities.construct(detectorClassName, "failure detector", IFailureDetector.class);
         return detector;
     }
 

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

@@ -64,7 +64,17 @@ public class ParameterizedClass
              p.containsKey(PARAMETERS) ? (Map<String, String>)((List<?>)p.get(PARAMETERS)).get(0) : null);
     }
 
+    /**
+     * Prefer {@link #newInstance(ParameterizedClass, List, Class)}: passing an {@code expectedType} verifies the
+     * resolved class is the intended extension type before it is instantiated. This overload performs no type
+     * check (it still loads without initialization, so a wrong class name cannot run its static initializer here).
+     */
     static public <K> K newInstance(ParameterizedClass parameterizedClass, List<String> searchPackages)
+    {
+        return newInstance(parameterizedClass, searchPackages, null);
+    }
+
+    static public <K> K newInstance(ParameterizedClass parameterizedClass, List<String> searchPackages, Class<K> expectedType)
     {
         Class<?> providerClass = null;
         if (searchPackages == null || searchPackages.isEmpty())
@@ -76,9 +86,12 @@ public class ParameterizedClass
                 if (!searchPackage.isEmpty() && !searchPackage.endsWith("."))
                     searchPackage = searchPackage + '.';
                 String name = searchPackage + parameterizedClass.class_name;
-                providerClass = Class.forName(name);
+                // Load without initialization so a wrong class name does not run its static initializer here. The
+                // type is verified below (once the search has resolved a class) and the class is only initialized
+                // later, when it is constructed.
+                providerClass = Class.forName(name, false, ParameterizedClass.class.getClassLoader());
             }
-            catch (ClassNotFoundException e)
+            catch (ClassNotFoundException | NoClassDefFoundError e)
             {
                 //no-op
             }
@@ -91,6 +104,13 @@ public class ParameterizedClass
             throw new ConfigurationException(error);
         }
 
+        // Verify the resolved class is the expected extension type before it is initialized/instantiated. Done once,
+        // after the package search, so a wrong-type match under an earlier search package does not abort the search
+        // before a valid class under a later package is found.
+        if (expectedType != null && !expectedType.isAssignableFrom(providerClass))
+            throw new ConfigurationException("Invalid parameterized class " + providerClass.getName() +
+                                             ": must extend or implement " + expectedType.getName());
+
         try
         {
             Constructor<?> mapConstructor = filterConstructor(providerClass, c -> c.getParameterTypes().length == 1 && c.getParameterTypes()[0].equals(Map.class));

src/java/org/apache/cassandra/db/StorageHook.java

@@ -55,7 +55,7 @@ public interface StorageHook
         String className = STORAGE_HOOK.getString();
         if (className != null)
         {
-            return FBUtilities.construct(className, StorageHook.class.getSimpleName());
+            return FBUtilities.construct(className, StorageHook.class.getSimpleName(), StorageHook.class);
         }
 
         return new StorageHook()

src/java/org/apache/cassandra/db/guardrails/GuardrailsConfigProvider.java

@@ -65,7 +65,7 @@ public interface GuardrailsConfigProvider
      */
     static GuardrailsConfigProvider build(String customImpl)
     {
-        return FBUtilities.construct(customImpl, "custom guardrails config provider");
+        return FBUtilities.construct(customImpl, "custom guardrails config provider", GuardrailsConfigProvider.class);
     }
 
     /**

src/java/org/apache/cassandra/db/guardrails/ValueGenerator.java

@@ -143,8 +143,11 @@ public abstract class ValueGenerator<VALUE>
 
         try
         {
+            Class<? extends ValueGenerator> rawGeneratorClass =
+                FBUtilities.classForNameWithoutInitialization(className, "generator", ValueGenerator.class);
+            @SuppressWarnings("unchecked")
             Class<? extends ValueGenerator<VALUE>> generatorClass =
-            FBUtilities.classForName(className, "generator");
+                (Class<? extends ValueGenerator<VALUE>>) rawGeneratorClass;
 
             @SuppressWarnings("unchecked")
             ValueGenerator<VALUE> generator = generatorClass.getConstructor(CustomGuardrailConfig.class)

src/java/org/apache/cassandra/db/guardrails/ValueValidator.java

@@ -127,8 +127,11 @@ public abstract class ValueValidator<VALUE>
 
         try
         {
+            Class<? extends ValueValidator> rawValidatorClass =
+                FBUtilities.classForNameWithoutInitialization(className, "validator", ValueValidator.class);
+            @SuppressWarnings("unchecked")
             Class<? extends ValueValidator<VALUE>> validatorClass =
-            FBUtilities.classForName(className, "validator");
+                (Class<? extends ValueValidator<VALUE>>) rawValidatorClass;
 
             @SuppressWarnings("unchecked")
             ValueValidator<VALUE> validator = validatorClass.getConstructor(CustomGuardrailConfig.class)

src/java/org/apache/cassandra/db/marshal/TypeParser.java

@@ -449,8 +449,7 @@ public class TypeParser
 
     private static AbstractType<?> getAbstractType(String compareWith) throws ConfigurationException
     {
-        String className = compareWith.contains(".") ? compareWith : "org.apache.cassandra.db.marshal." + compareWith;
+        Class<? extends AbstractType<?>> typeClass = getAbstractTypeClass(compareWith);
-        Class<? extends AbstractType<?>> typeClass = FBUtilities.<AbstractType<?>>classForName(className, "abstract-type");
         try
         {
             Field field = typeClass.getDeclaredField("instance");
@@ -463,10 +462,22 @@ public class TypeParser
         }
     }
 
-    private static AbstractType<?> getAbstractType(String compareWith, TypeParser parser) throws SyntaxException, ConfigurationException
+    private static Class<? extends AbstractType<?>> getAbstractTypeClass(String compareWith) throws ConfigurationException
     {
         String className = compareWith.contains(".") ? compareWith : "org.apache.cassandra.db.marshal." + compareWith;
-        Class<? extends AbstractType<?>> typeClass = FBUtilities.<AbstractType<?>>classForName(className, "abstract-type");
+        // Defer class initialization until after confirming this is an AbstractType. The static instance field
+        // access or getInstance(TypeParser) invocation below performs the initialization for valid types.
+        @SuppressWarnings("unchecked")
+        Class<? extends AbstractType<?>> typeClass =
+            (Class<? extends AbstractType<?>>) FBUtilities.classForNameWithoutInitialization(className,
+                                                                                             "abstract-type",
+                                                                                             AbstractType.class);
+        return typeClass;
+    }
+
+    private static AbstractType<?> getAbstractType(String compareWith, TypeParser parser) throws SyntaxException, ConfigurationException
+    {
+        Class<? extends AbstractType<?>> typeClass = getAbstractTypeClass(compareWith);
         if (PseudoUtf8Type.class.isAssignableFrom(typeClass))
         {
             if (StorageService.instance.isDaemonSetupCompleted())

src/java/org/apache/cassandra/diag/DiagnosticEventPersistence.java

@@ -128,6 +128,7 @@ public final class DiagnosticEventPersistence
         LastEventIdBroadcaster.instance().setLastEventId(event.getClass().getName(), store.getLastEventId());
     }
 
+    @SuppressWarnings("unchecked")
     private Class<DiagnosticEvent> getEventClass(String eventClazz) throws ClassNotFoundException, InvalidClassException
     {
         // get class by eventClazz argument name
@@ -135,12 +136,13 @@ public final class DiagnosticEventPersistence
         if (!eventClazz.startsWith("org.apache.cassandra."))
             throw new RuntimeException("Not a Cassandra event class: " + eventClazz);
 
-        Class<DiagnosticEvent> clazz = (Class<DiagnosticEvent>) Class.forName(eventClazz);
+        // Load without initialization so the type can be verified before the class's static initializer runs.
+        Class<?> clazz = Class.forName(eventClazz, false, DiagnosticEventPersistence.class.getClassLoader());
 
         if (!(DiagnosticEvent.class.isAssignableFrom(clazz)))
             throw new InvalidClassException("Event class must be of type DiagnosticEvent");
 
-        return clazz;
+        return (Class<DiagnosticEvent>) clazz.asSubclass(DiagnosticEvent.class);
     }
 
     private DiagnosticEventStore<Long> getStore(Class cls)

src/java/org/apache/cassandra/index/SecondaryIndexManager.java

@@ -921,6 +921,11 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
         return indexes.get(indexName);
     }
 
+    static Class<? extends Index> loadIndexClass(String className)
+    {
+        return FBUtilities.classForNameWithoutInitialization(className, "Index", Index.class);
+    }
+
     private Index createInstance(IndexMetadata indexDef)
     {
         Index newIndex;
@@ -933,7 +938,7 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
 
             try
             {
-                Class<? extends Index> indexClass = FBUtilities.classForName(className, "Index");
+                Class<? extends Index> indexClass = loadIndexClass(className);
                 Constructor<? extends Index> ctor = indexClass.getConstructor(ColumnFamilyStore.class, IndexMetadata.class);
                 newIndex = ctor.newInstance(baseCfs, indexDef);
             }

src/java/org/apache/cassandra/index/sasi/conf/IndexMode.java

@@ -37,6 +37,7 @@ import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder.Mode;
 import org.apache.cassandra.index.sasi.plan.Expression.Op;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.utils.FBUtilities;
 
 public class IndexMode
 {
@@ -60,10 +61,10 @@ public class IndexMode
 
     public final Mode mode;
     public final boolean isAnalyzed, isLiteral;
-    public final Class analyzerClass;
+    public final Class<? extends AbstractAnalyzer> analyzerClass;
     public final long maxCompactionFlushMemoryInBytes;
 
-    private IndexMode(Mode mode, boolean isLiteral, boolean isAnalyzed, Class analyzerClass, long maxMemBytes)
+    private IndexMode(Mode mode, boolean isLiteral, boolean isAnalyzed, Class<? extends AbstractAnalyzer> analyzerClass, long maxMemBytes)
     {
         this.mode = mode;
         this.isLiteral = isLiteral;
@@ -81,7 +82,7 @@ public class IndexMode
             if (isAnalyzed)
             {
                 if (analyzerClass != null)
-                    analyzer = (AbstractAnalyzer) analyzerClass.newInstance();
+                    analyzer = analyzerClass.newInstance();
                 else if (TOKENIZABLE_TYPES.contains(validator))
                     analyzer = new StandardAnalyzer();
             }
@@ -99,21 +100,14 @@ public class IndexMode
         // validate that a valid analyzer class was provided if specified
         if (indexOptions.containsKey(INDEX_ANALYZER_CLASS_OPTION))
         {
-            Class<?> analyzerClass;
+            Class<? extends AbstractAnalyzer> analyzerClass = FBUtilities.classForNameWithoutInitialization(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION),
-            try
+                                                                                                           "analyzer",
-            {
+                                                                                                           AbstractAnalyzer.class);
-                analyzerClass = Class.forName(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
-            }
-            catch (ClassNotFoundException e)
-            {
-                throw new ConfigurationException(String.format("Invalid analyzer class option specified [%s]",
-                                                               indexOptions.get(INDEX_ANALYZER_CLASS_OPTION)));
-            }
 
             AbstractAnalyzer analyzer;
             try
             {
-                analyzer = (AbstractAnalyzer) analyzerClass.newInstance();
+                analyzer = analyzerClass.newInstance();
                 analyzer.validate(indexOptions, cd);
             }
             catch (InstantiationException | IllegalAccessException e)
@@ -148,26 +142,31 @@ public class IndexMode
         }
 
         boolean isAnalyzed = false;
-        Class analyzerClass = null;
+        Class<? extends AbstractAnalyzer> analyzerClass = null;
-        try
-        {
         if (indexOptions.get(INDEX_ANALYZER_CLASS_OPTION) != null)
         {
-                analyzerClass = Class.forName(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
+            try
+            {
+                analyzerClass = FBUtilities.classForNameWithoutInitialization(indexOptions.get(INDEX_ANALYZER_CLASS_OPTION),
+                                                                              "analyzer",
+                                                                              AbstractAnalyzer.class);
                 isAnalyzed = indexOptions.get(INDEX_ANALYZED_OPTION) == null
                              ? true : Boolean.parseBoolean(indexOptions.get(INDEX_ANALYZED_OPTION));
             }
+            catch (ConfigurationException e)
+            {
+                if (!(e.getCause() instanceof ClassNotFoundException))
+                    throw e;
+
+                // Should not happen as we already validated we could instantiate an instance in validateAnalyzer().
+                logger.error("Failed to find specified analyzer class [{}]. Falling back to default analyzer",
+                             indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
+            }
+        }
         else if (indexOptions.get(INDEX_ANALYZED_OPTION) != null)
         {
             isAnalyzed = Boolean.parseBoolean(indexOptions.get(INDEX_ANALYZED_OPTION));
         }
-        }
-        catch (ClassNotFoundException e)
-        {
-            // should not happen as we already validated we could instantiate an instance in validateAnalyzer()
-            logger.error("Failed to find specified analyzer class [{}]. Falling back to default analyzer",
-                         indexOptions.get(INDEX_ANALYZER_CLASS_OPTION));
-        }
 
         boolean isLiteral = false;
         try

src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java

@@ -340,11 +340,11 @@ public abstract class AbstractReplicationStrategy
         if ("org.apache.cassandra.locator.OldNetworkTopologyStrategy".equals(className)) // see CASSANDRA-16301 
             throw new ConfigurationException("The support for the OldNetworkTopologyStrategy has been removed in C* version 4.0. The keyspace strategy should be switch to NetworkTopologyStrategy");
 
-        Class<AbstractReplicationStrategy> strategyClass = FBUtilities.classForName(className, "replication strategy");
+        @SuppressWarnings("unchecked")
-        if (!AbstractReplicationStrategy.class.isAssignableFrom(strategyClass))
+        Class<AbstractReplicationStrategy> strategyClass =
-        {
+            (Class<AbstractReplicationStrategy>) FBUtilities.classForNameWithoutInitialization(className,
-            throw new ConfigurationException(String.format("Specified replication strategy class (%s) is not derived from AbstractReplicationStrategy", className));
+                                                                                             "replication strategy",
-        }
+                                                                                             AbstractReplicationStrategy.class);
         return strategyClass;
     }
 

src/java/org/apache/cassandra/repair/autorepair/AutoRepairConfig.java

@@ -395,7 +395,10 @@ public class AutoRepairConfig implements Serializable
                 className = parameterizedClass.class_name.contains(".") ?
                             parameterizedClass.class_name :
                             "org.apache.cassandra.repair.autorepair." + parameterizedClass.class_name;
-                tokenRangeSplitterClass = FBUtilities.classForName(className, "token_range_splitter");
+                tokenRangeSplitterClass =
+                    FBUtilities.classForNameWithoutInitialization(className,
+                                                                  "token_range_splitter",
+                                                                  IAutoRepairTokenRangeSplitter.class);
             }
             else
             {

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

@@ -311,15 +311,7 @@ public final class CompactionParams
         String className = name.contains(".")
                          ? name
                          : "org.apache.cassandra.db.compaction." + name;
-        Class<AbstractCompactionStrategy> strategyClass = FBUtilities.classForName(className, "compaction strategy");
+        return FBUtilities.classForNameWithoutInitialization(className, "compaction strategy", AbstractCompactionStrategy.class);
-
-        if (!AbstractCompactionStrategy.class.isAssignableFrom(strategyClass))
-        {
-            throw new ConfigurationException(format("Compaction strategy class %s is not derived from AbstractReplicationStrategy",
-                                                    className));
-        }
-
-        return strategyClass;
     }
 
     /*

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

@@ -47,6 +47,7 @@ import org.apache.cassandra.io.compress.ZstdDictionaryCompressor;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
 
 import static java.lang.String.format;
 
@@ -302,7 +303,7 @@ public final class CompressionParams
         return maxCompressedLength;
     }
 
-    private static Class<?> parseCompressorClass(String className) throws ConfigurationException
+    private static Class<? extends ICompressor> parseCompressorClass(String className) throws ConfigurationException
     {
         if (className == null || className.isEmpty())
             return null;
@@ -310,15 +311,17 @@ public final class CompressionParams
         className = className.contains(".") ? className : "org.apache.cassandra.io.compress." + className;
         try
         {
-            return Class.forName(className);
+            return FBUtilities.classForNameWithoutInitialization(className, "compression", ICompressor.class);
         }
-        catch (Exception e)
+        catch (ConfigurationException e)
         {
+            if (e.getCause() instanceof ClassNotFoundException || e.getCause() instanceof NoClassDefFoundError)
                 throw new ConfigurationException("Could not create Compression for type " + className, e);
+            throw e;
         }
     }
 
-    private static ICompressor createCompressor(Class<?> compressorClass, Map<String, String> compressionOptions) throws ConfigurationException
+    private static ICompressor createCompressor(Class<? extends ICompressor> compressorClass, Map<String, String> compressionOptions) throws ConfigurationException
     {
         if (compressorClass == null)
         {

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

@@ -148,9 +148,7 @@ public final class IndexMetadata
             // Get the fully qualified class name:
             String className = getIndexClassName();
 
-            Class<Index> indexerClass = FBUtilities.classForName(className, "custom indexer");
+            Class<? extends Index> indexerClass = FBUtilities.classForNameWithoutInitialization(className, "custom indexer", Index.class);
-            if (!Index.class.isAssignableFrom(indexerClass))
-                throw new ConfigurationException(String.format("Specified Indexer class (%s) does not implement the Indexer interface", className));
             validateCustomIndexOptions(table, indexerClass, options);
         }
     }

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

@@ -228,19 +228,25 @@ public final class MemtableParams
         try
         {
             Memtable.Factory factory;
-            Class<?> clazz = Class.forName(className);
+            Class<?> clazz = Class.forName(className, false, MemtableParams.class.getClassLoader());
             final Map<String, String> parametersCopy = options.parameters != null
                                                        ? new HashMap<>(options.parameters)
                                                        : new HashMap<>();
             try
             {
                 Method factoryMethod = clazz.getDeclaredMethod("factory", Map.class);
+                if (!Memtable.Factory.class.isAssignableFrom(factoryMethod.getReturnType()))
+                    throw new ClassCastException("Memtable factory method on " + className +
+                                                 " must return " + Memtable.Factory.class.getName());
                 factory = (Memtable.Factory) factoryMethod.invoke(null, parametersCopy);
             }
             catch (NoSuchMethodException e)
             {
                 // continue with FACTORY field
                 Field factoryField = clazz.getDeclaredField("FACTORY");
+                if (!Memtable.Factory.class.isAssignableFrom(factoryField.getType()))
+                    throw new ClassCastException("Memtable FACTORY field on " + className +
+                                                 " must be of type " + Memtable.Factory.class.getName());
                 factory = (Memtable.Factory) factoryField.get(null);
             }
             if (!parametersCopy.isEmpty())

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

@@ -286,7 +286,7 @@ public final class ReplicationParams
             Map<String, String> options = new HashMap<>(size);
             for (int i = 0; i < size; i++)
                 options.put(in.readUTF(), in.readUTF());
-            return new ReplicationParams(FBUtilities.classForName(klassName, "ReplicationStrategy"), options);
+            return new ReplicationParams(FBUtilities.classForNameWithoutInitialization(klassName, "ReplicationStrategy", AbstractReplicationStrategy.class), options);
         }
 
         public long serializedSize(ReplicationParams t, Version version)
@@ -322,7 +322,7 @@ public final class ReplicationParams
             Map<String, String> options = new HashMap<>(size);
             for (int i=0; i<size; i++)
                 options.put(in.readUTF(), in.readUTF());
-            return new ReplicationParams(FBUtilities.classForName(klassName, "ReplicationStrategy"), options);
+            return new ReplicationParams(FBUtilities.classForNameWithoutInitialization(klassName, "ReplicationStrategy", AbstractReplicationStrategy.class), options);
         }
 
         public long serializedSize(ReplicationParams t, int version)

src/java/org/apache/cassandra/security/AbstractCryptoProvider.java

@@ -105,7 +105,7 @@ public abstract class AbstractCryptoProvider
                 return;
             }
 
-            FBUtilities.classForName(getProviderClassAsString(), "crypto provider");
+            FBUtilities.classForNameWithoutInitialization(getProviderClassAsString(), "crypto provider", Provider.class);
 
             String providerName = getProviderName();
             int providerPosition = getProviderPosition(providerName);

src/java/org/apache/cassandra/security/CipherFactory.java

@@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.ImmediateExecutor;
 import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.utils.FBUtilities;
 
 import io.netty.util.concurrent.FastThreadLocal;
 
@@ -71,9 +72,10 @@ public class CipherFactory
         try
         {
             secureRandom = SecureRandom.getInstance("SHA1PRNG");
-            Class<KeyProvider> keyProviderClass = (Class<KeyProvider>)Class.forName(options.key_provider.class_name);
+            Class<? extends KeyProvider> keyProviderClass =
-            Constructor ctor = keyProviderClass.getConstructor(TransparentDataEncryptionOptions.class);
+                FBUtilities.classForNameWithoutInitialization(options.key_provider.class_name, "key provider", KeyProvider.class);
-            keyProvider = (KeyProvider)ctor.newInstance(options);
+            Constructor<? extends KeyProvider> ctor = keyProviderClass.getConstructor(TransparentDataEncryptionOptions.class);
+            keyProvider = ctor.newInstance(options);
         }
         catch (Exception e)
         {

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

@@ -147,9 +147,11 @@ public class CacheService implements CacheServiceMBean
                                         ? DatabaseDescriptor.getRowCacheClassName() : "org.apache.cassandra.cache.NopCacheProvider";
         try
         {
-            Class<CacheProvider<RowCacheKey, IRowCacheEntry>> cacheProviderClass =
+            Class<? extends CacheProvider> cacheProviderClass =
-                (Class<CacheProvider<RowCacheKey, IRowCacheEntry>>) Class.forName(cacheProviderClassName);
+                FBUtilities.classForNameWithoutInitialization(cacheProviderClassName, "row cache provider", CacheProvider.class);
-            cacheProvider = cacheProviderClass.newInstance();
+            @SuppressWarnings("unchecked")
+            CacheProvider<RowCacheKey, IRowCacheEntry> typedCacheProvider = cacheProviderClass.newInstance();
+            cacheProvider = typedCacheProvider;
         }
         catch (Exception e)
         {

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

@@ -124,7 +124,7 @@ public class ClientState
         {
             try
             {
-                handler = FBUtilities.construct(customHandlerClass, "QueryHandler");
+                handler = FBUtilities.construct(customHandlerClass, "QueryHandler", QueryHandler.class);
                 logger.info("Using {} as a query handler for native protocol queries (as requested by the {} system property)",
                             customHandlerClass, CUSTOM_QUERY_HANDLER_CLASS.getKey());
             }

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

@@ -78,7 +78,7 @@ public class DiskErrorsHandlerService
         String fsErrorHandlerClass = CassandraRelevantProperties.CUSTOM_DISK_ERROR_HANDLER.getString();
         DiskErrorsHandler fsErrorHandler = fsErrorHandlerClass == null
                                            ? new DefaultDiskErrorsHandler()
-                                           : FBUtilities.construct(fsErrorHandlerClass, "disk error handler");
+                                           : FBUtilities.construct(fsErrorHandlerClass, "disk error handler", DiskErrorsHandler.class);
         DiskErrorsHandlerService.set(fsErrorHandler);
     }
 }

src/java/org/apache/cassandra/service/accord/AccordService.java

@@ -458,7 +458,9 @@ public class AccordService implements IAccordService, Shutdownable
     {
         Invariants.require(localId != null, "static localId must be set before instantiating AccordService");
         logger.info("Starting accord with nodeId {}", localId);
-        AccordAgent agent = FBUtilities.construct(CassandraRelevantProperties.ACCORD_AGENT_CLASS.getString(AccordAgent.class.getName()), "AccordAgent");
+        AccordAgent agent = FBUtilities.construct(CassandraRelevantProperties.ACCORD_AGENT_CLASS.getString(AccordAgent.class.getName()),
+                                                  "AccordAgent",
+                                                  AccordAgent.class);
         agent.setup(localId);
         AccordTimeService time = new AccordTimeService();
         this.scheduler = new AccordScheduler();

src/java/org/apache/cassandra/streaming/StreamHook.java

@@ -37,7 +37,7 @@ public interface StreamHook
         String className = STREAM_HOOK.getString();
         if (className != null)
         {
-            return FBUtilities.construct(className, StreamHook.class.getSimpleName());
+            return FBUtilities.construct(className, StreamHook.class.getSimpleName(), StreamHook.class);
         }
         else
         {

src/java/org/apache/cassandra/tcm/extensions/ExtensionKey.java

@@ -41,7 +41,7 @@ public class ExtensionKey<V, K extends ExtensionValue<V>> extends MetadataKey
 
     public K newValue()
     {
-        return valueType.cast(FBUtilities.construct(valueType.getName(), "extension value"));
+        return FBUtilities.construct(valueType.getName(), "extension value", valueType);
     }
 
     public static final class Serializer implements MetadataSerializer<ExtensionKey<?, ?>>
@@ -58,7 +58,9 @@ public class ExtensionKey<V, K extends ExtensionValue<V>> extends MetadataKey
         {
             String id = in.readUTF();
             String valType = in.readUTF();
-            return new ExtensionKey(id, FBUtilities.classForName(valType, "value type"));
+            Class<? extends ExtensionValue> valueType =
+                FBUtilities.classForNameWithoutInitialization(valType, "value type", ExtensionValue.class);
+            return new ExtensionKey(id, valueType);
         }
 
         @Override

src/java/org/apache/cassandra/tools/nodetool/Sjk.java

@@ -393,6 +393,7 @@ public class Sjk extends AbstractCommand
             List<Class<?>> result = new ArrayList<>();
             try
             {
+                ClassLoader cl = Thread.currentThread().getContextClassLoader();
                 String path = packageName.replace('.', '/');
                 for (String f : findFiles(path))
                 {
@@ -400,7 +401,7 @@ public class Sjk extends AbstractCommand
                     {
                         f = f.substring(0, f.length() - ".class".length());
                         f = f.replace('/', '.');
-                        result.add(Class.forName(f));
+                        result.add(Class.forName(f, false, cl));
                     }
                 }
                 return result;

src/java/org/apache/cassandra/tracing/Tracing.java

@@ -117,7 +117,7 @@ public abstract class Tracing extends ExecutorLocals.Impl
         {
             try
             {
-                tracing = FBUtilities.construct(customTracingClass, "Tracing");
+                tracing = FBUtilities.construct(customTracingClass, "Tracing", Tracing.class);
                 logger.info("Using the {} class to trace queries (as requested by the {} system property)",
                             customTracingClass, CUSTOM_TRACING_CLASS.getKey());
             }

src/java/org/apache/cassandra/triggers/TriggerExecutor.java

@@ -286,13 +286,33 @@ public class TriggerExecutor
         }
     }
 
-    public synchronized void loadTriggerClass(String triggerClass) throws Exception
+    public synchronized Class<? extends ITrigger> loadTriggerClass(String triggerClass) throws Exception
     {
         // Allow loading the class regardless of Config, since this could happen as part of TCM replay via
         // CreateTriggerStatement#apply.
         // Check that triggerClass is available on the classpath, but do not initialize the class since that would
         // execute static blocks.
-        customClassLoader.loadClass(triggerClass).getConstructor();
+        Class<? extends ITrigger> trigger = loadTriggerClassWithoutInitialization(triggerClass);
+        // Validate that the class exposes a public no-argument constructor before it is accepted.
+        trigger.getConstructor();
+        return trigger;
+    }
+
+    private Class<? extends ITrigger> loadTriggerClassWithoutInitialization(String triggerClass) throws Exception
+    {
+        try
+        {
+            return FBUtilities.classForNameWithoutInitialization(triggerClass,
+                                                                 "trigger",
+                                                                 ITrigger.class,
+                                                                 customClassLoader);
+        }
+        catch (ConfigurationException e)
+        {
+            if (e.getCause() instanceof ClassNotFoundException)
+                throw (ClassNotFoundException) e.getCause();
+            throw e;
+        }
     }
 
     public synchronized ITrigger loadTriggerInstance(String triggerClass) throws Exception
@@ -306,6 +326,6 @@ public class TriggerExecutor
         // double check.
         if (cachedTriggers.get(triggerClass) != null)
             return cachedTriggers.get(triggerClass);
-        return (ITrigger) customClassLoader.loadClass(triggerClass).getConstructor().newInstance();
+        return loadTriggerClassWithoutInitialization(triggerClass).getConstructor().newInstance();
     }
 }

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

@@ -63,7 +63,7 @@ public interface Clock
                 try
                 {
                     outcome = "Using custom clock implementation: " + classname;
-                    clock = (Clock) Class.forName(classname).newInstance();
+                    clock = FBUtilities.construct(classname, "clock", Clock.class);
                 }
                 catch (Throwable t)
                 {

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

@@ -678,7 +678,7 @@ public class FBUtilities
             assert comparator.isPresent() : "Expected a comparator for local partitioner";
             return new LocalPartitioner(comparator.get());
         }
-        return FBUtilities.instanceOrConstruct(partitionerClassName, "partitioner");
+        return FBUtilities.instanceOrConstruct(partitionerClassName, "partitioner", IPartitioner.class);
     }
 
     public static IAuditLogger newAuditLogger(String className, Map<String, String> parameters) throws ConfigurationException
@@ -688,8 +688,9 @@ public class FBUtilities
 
         try
         {
-            Class<?> auditLoggerClass = FBUtilities.classForName(className, "Audit logger");
+            Class<? extends IAuditLogger> auditLoggerClass =
-            return (IAuditLogger) auditLoggerClass.getConstructor(Map.class).newInstance(parameters);
+                FBUtilities.classForNameWithoutInitialization(className, "Audit logger", IAuditLogger.class);
+            return auditLoggerClass.getConstructor(Map.class).newInstance(parameters);
         }
         catch (Exception ex)
         {
@@ -704,12 +705,16 @@ public class FBUtilities
 
         try
         {
-            Class<?> sslContextFactoryClass = Class.forName(className);
+            Class<? extends ISslContextFactory> sslContextFactoryClass =
-            return (ISslContextFactory) sslContextFactoryClass.getConstructor(Map.class).newInstance(parameters);
+                FBUtilities.classForNameWithoutInitialization(className, "ISslContextFactory", ISslContextFactory.class);
+            return sslContextFactoryClass.getConstructor(Map.class).newInstance(parameters);
         }
         catch (Exception ex)
         {
-            throw new ConfigurationException("Unable to create instance of ISslContextFactory for " + className, ex);
+            // Surface the underlying load failure (e.g. ClassNotFoundException) as the direct cause rather than the
+            // intermediate ConfigurationException that reports it.
+            Throwable cause = ex instanceof ConfigurationException && ex.getCause() != null ? ex.getCause() : ex;
+            throw new ConfigurationException("Unable to create instance of ISslContextFactory for " + className, cause);
         }
     }
 
@@ -720,8 +725,9 @@ public class FBUtilities
             if (!className.contains("."))
                 className = "org.apache.cassandra.security." + className;
 
-            Class<?> cryptoProviderClass = FBUtilities.classForName(className, "crypto provider class");
+            Class<? extends AbstractCryptoProvider> cryptoProviderClass =
-            return (AbstractCryptoProvider) cryptoProviderClass.getConstructor(Map.class).newInstance(Collections.unmodifiableMap(parameters));
+                FBUtilities.classForNameWithoutInitialization(className, "crypto provider class", AbstractCryptoProvider.class);
+            return cryptoProviderClass.getConstructor(Map.class).newInstance(Collections.unmodifiableMap(parameters));
         }
         catch (Exception e)
         {
@@ -734,6 +740,8 @@ public class FBUtilities
     }
 
     /**
+     * Loads and initializes a class.
+     *
      * @return The Class for the given name.
      * @param classname Fully qualified classname.
      * @param readable Descriptive noun for the role the class plays.
@@ -751,6 +759,53 @@ public class FBUtilities
         }
     }
 
+    /**
+     * Loads a class without initializing it, then verifies it extends or implements the expected base type.
+     *
+     * @return The Class for the given name.
+     * @param classname Fully qualified classname.
+     * @param readable Descriptive noun for the role the class plays.
+     * @param expectedType Required superclass or interface.
+     * @throws ConfigurationException If the class cannot be found or is not assignable to {@code expectedType}.
+     */
+    public static <T> Class<? extends T> classForNameWithoutInitialization(String classname,
+                                                                           String readable,
+                                                                           Class<T> expectedType) throws ConfigurationException
+    {
+        return classForNameWithoutInitialization(classname, readable, expectedType, FBUtilities.class.getClassLoader());
+    }
+
+    /**
+     * Loads a class without initializing it, then verifies it extends or implements the expected base type.
+     *
+     * @return The Class for the given name.
+     * @param classname Fully qualified classname.
+     * @param readable Descriptive noun for the role the class plays.
+     * @param expectedType Required superclass or interface.
+     * @param classLoader ClassLoader to use.
+     * @throws ConfigurationException If the class cannot be found or is not assignable to {@code expectedType}.
+     */
+    public static <T> Class<? extends T> classForNameWithoutInitialization(String classname,
+                                                                           String readable,
+                                                                           Class<T> expectedType,
+                                                                           ClassLoader classLoader) throws ConfigurationException
+    {
+        try
+        {
+            Class<?> klass = Class.forName(classname, false, classLoader);
+            if (!expectedType.isAssignableFrom(klass))
+                throw new ConfigurationException(String.format("Invalid %s class '%s': must extend or implement %s",
+                                                               readable,
+                                                               classname,
+                                                               expectedType.getName()));
+            return klass.asSubclass(expectedType);
+        }
+        catch (ClassNotFoundException | NoClassDefFoundError e)
+        {
+            throw new ConfigurationException(String.format("Unable to find %s class '%s'", readable, classname), e);
+        }
+    }
+
     /**
      * Constructs an instance of the given class, which must have a no-arg or default constructor.
      * @param classname Fully qualified classname.
@@ -760,6 +815,25 @@ public class FBUtilities
     public static <T> T instanceOrConstruct(String classname, String readable) throws ConfigurationException
     {
         Class<T> cls = FBUtilities.classForName(classname, readable);
+        return instanceOrConstruct(cls, classname, readable);
+    }
+
+    /**
+     * Constructs an instance of the given class, or gets its static {@code instance} field, after verifying the
+     * class without initializing it.
+     * @param classname Fully qualified classname.
+     * @param readable Descriptive noun for the role the class plays.
+     * @param expectedType Required superclass or interface.
+     * @throws ConfigurationException If the class cannot be found or is not assignable to {@code expectedType}.
+     */
+    public static <T> T instanceOrConstruct(String classname, String readable, Class<T> expectedType) throws ConfigurationException
+    {
+        Class<? extends T> cls = FBUtilities.classForNameWithoutInitialization(classname, readable, expectedType);
+        return instanceOrConstruct(cls, classname, readable);
+    }
+
+    private static <T> T instanceOrConstruct(Class<? extends T> cls, String classname, String readable) throws ConfigurationException
+    {
         try
         {
             Field instance = cls.getField("instance");
@@ -784,7 +858,20 @@ public class FBUtilities
         return construct(cls, classname, readable);
     }
 
-    private static <T> T construct(Class<T> cls, String classname, String readable) throws ConfigurationException
+    /**
+     * Constructs an instance of the given class after verifying it without initializing it.
+     * @param classname Fully qualified classname.
+     * @param readable Descriptive noun for the role the class plays.
+     * @param expectedType Required superclass or interface.
+     * @throws ConfigurationException If the class cannot be found or is not assignable to {@code expectedType}.
+     */
+    public static <T> T construct(String classname, String readable, Class<T> expectedType) throws ConfigurationException
+    {
+        Class<? extends T> cls = FBUtilities.classForNameWithoutInitialization(classname, readable, expectedType);
+        return construct(cls, classname, readable);
+    }
+
+    private static <T> T construct(Class<? extends T> cls, String classname, String readable) throws ConfigurationException
     {
         try
         {

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

@@ -222,7 +222,7 @@ public class JMXServerUtils
         String authzProxyClass = options.authorizer;
         if (authzProxyClass != null)
         {
-            final InvocationHandler handler = FBUtilities.construct(authzProxyClass, "JMX authz proxy");
+            final InvocationHandler handler = FBUtilities.construct(authzProxyClass, "JMX authz proxy", InvocationHandler.class);
             final Class[] interfaces = { MBeanServerForwarder.class };
 
             Object proxy = Proxy.newProxyInstance(MBeanServerForwarder.class.getClassLoader(), interfaces, handler);

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

@@ -78,7 +78,7 @@ public interface MBeanWrapper
                 return new PlatformMBeanWrapper();
             }
         }
-        return FBUtilities.construct(klass, "mbean");
+        return FBUtilities.construct(klass, "mbean", MBeanWrapper.class);
     }
 
     // Passing true for graceful will log exceptions instead of rethrowing them

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

@@ -94,7 +94,7 @@ public interface MonotonicClock
                 try
                 {
                     logger.debug("Using custom clock implementation: {}", sclock);
-                    return (MonotonicClock) Class.forName(sclock).newInstance();
+                    return FBUtilities.construct(sclock, "monotonic clock", MonotonicClock.class);
                 }
                 catch (Exception e)
                 {
@@ -113,7 +113,8 @@ public interface MonotonicClock
                 try
                 {
                     logger.debug("Using custom clock implementation: {}", sclock);
-                    Class<? extends MonotonicClock> clazz = (Class<? extends MonotonicClock>) Class.forName(sclock);
+                    Class<? extends MonotonicClock> clazz =
+                        FBUtilities.classForNameWithoutInitialization(sclock, "monotonic clock", MonotonicClock.class);
 
                     if (SystemClock.class.equals(clazz) && SystemClock.class.equals(precise.getClass()))
                         return precise;

test/unit/org/apache/cassandra/auth/AuthConfigTest.java

@@ -31,7 +31,10 @@ 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.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 import org.apache.cassandra.utils.MBeanWrapper;
 
 import static org.apache.cassandra.auth.AuthCache.MBEAN_NAME_BASE;
@@ -39,6 +42,7 @@ import static org.apache.cassandra.auth.AuthTestUtils.loadCertificateChain;
 import static org.apache.cassandra.auth.IInternodeAuthenticator.InternodeConnectionDirection.INBOUND;
 import static org.apache.cassandra.config.YamlConfigurationLoaderTest.load;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -160,6 +164,75 @@ public class AuthConfigTest
         assertTrue(DatabaseDescriptor.getRoleManager().alterableOptions().containsAll(authenticator.getAlterableRoleOptions()));
     }
 
+    private static final String PROBE = ClassLoadingTestNonAssignable.class.getName();
+
+    private static Config baseConfig()
+    {
+        // Use the default (AllowAll) auth config so the non-probe auth components do not register JMX caches that
+        // would conflict across the per-field probe tests below.
+        Config config = load("cassandra.yaml");
+        DatabaseDescriptor.unsafeDaemonInitialization(() -> config);
+        return config;
+    }
+
+    private static void assertApplyAuthRejectsProbe()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        AuthConfig.reset();
+        assertThatThrownBy(AuthConfig::applyAuth)
+            .isInstanceOf(ConfigurationException.class)
+            .hasMessageContaining("must extend or implement");
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testAuthenticatorWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.authenticator = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
+    @Test
+    public void testAuthorizerWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.authorizer = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
+    @Test
+    public void testRoleManagerWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.role_manager = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
+    @Test
+    public void testInternodeAuthenticatorWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.internode_authenticator = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
+    @Test
+    public void testNetworkAuthorizerWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.network_authorizer = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
+    @Test
+    public void testCidrAuthorizerWrongTypeRejectedWithoutInitializing()
+    {
+        Config config = baseConfig();
+        config.cidr_authorizer = new ParameterizedClass(PROBE, Collections.emptyMap());
+        assertApplyAuthRejectsProbe();
+    }
+
     private void unregisterCaches()
     {
         safeUnregisterMbean(IDENTITIES_CACHE_MBEAN);

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

@@ -0,0 +1,37 @@
+/*
+ * 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.config;
+
+/**
+ * Search-package fall-through probe. Shares its simple name ({@code ClassLoadingSearchProbe}) with
+ * {@link org.apache.cassandra.utils.ClassLoadingSearchProbe}, but unlike that one it IS a {@link Runnable}, so it
+ * represents the "correct type under a later search package" case. A wrong-type match under an earlier package must
+ * not abort the search before this class is found.
+ */
+public final class ClassLoadingSearchProbe implements Runnable
+{
+    public ClassLoadingSearchProbe()
+    {
+    }
+
+    @Override
+    public void run()
+    {
+    }
+}

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

@@ -47,8 +47,11 @@ import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.PathUtils;
+import org.apache.cassandra.locator.IEndpointSnitch;
 import org.apache.cassandra.security.EncryptionContext;
 import org.apache.cassandra.security.EncryptionContextGenerator;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNLIMITED_CONCURRENT_VALIDATIONS;
 import static org.apache.cassandra.config.CassandraRelevantProperties.CONFIG_LOADER;
@@ -72,6 +75,25 @@ public class DatabaseDescriptorTest
         DatabaseDescriptor.daemonInitialization();
     }
 
+    @Test
+    public void testCreateEndpointSnitchWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> DatabaseDescriptor.createEndpointSnitch(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + IEndpointSnitch.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testCreateEndpointSnitchValidClassResolves()
+    {
+        IEndpointSnitch snitch = DatabaseDescriptor.createEndpointSnitch("SimpleSnitch");
+        assertThat(snitch).isNotNull();
+    }
+
     // this came as a result of CASSANDRA-995
     @Test
     public void testConfigurationLoader() throws Exception

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

@@ -26,7 +26,10 @@ import org.junit.Test;
 import org.apache.cassandra.auth.AllowAllAuthorizer;
 import org.apache.cassandra.auth.IAuthorizer;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
@@ -98,4 +101,49 @@ public class ParameterizedClassTest
         .hasMessageContaining("Simulated failure")
         .isInstanceOf(ConfigurationException.class);
     }
+
+    @Test
+    public void testTypedNewInstanceRejectsWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        ParameterizedClass parameterizedClass = new ParameterizedClass(ClassLoadingTestNonAssignable.class.getName());
+
+        assertThatThrownBy(() -> ParameterizedClass.newInstance(parameterizedClass, null, Runnable.class))
+        .hasMessageContaining("must extend or implement " + Runnable.class.getName())
+        .isInstanceOf(ConfigurationException.class);
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testTwoArgNewInstanceDoesNotInitializeAtLoadTime()
+    {
+        // The 2-arg overload performs no type check, but it must still load without running <clinit>.
+        // ClassLoadingTestNonAssignable has no usable constructor, so instantiation fails -- but only after the
+        // class has been loaded; loading must not have run its static initializer.
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        ParameterizedClass parameterizedClass = new ParameterizedClass(ClassLoadingTestNonAssignable.class.getName());
+
+        assertThatThrownBy(() -> ParameterizedClass.newInstance(parameterizedClass, null))
+        .isInstanceOf(ConfigurationException.class);
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testNewInstanceSearchPackageFallThroughDoesNotAbortOnWrongType()
+    {
+        // org.apache.cassandra.utils.ClassLoadingSearchProbe (wrong type, not a Runnable) resolves under the earlier
+        // search package, while org.apache.cassandra.config.ClassLoadingSearchProbe (correct type, a Runnable)
+        // resolves under the later one. A wrong-type match under the earlier package must not abort the search.
+        ClassLoadingTestSupport.assertNotInitialized(org.apache.cassandra.utils.ClassLoadingSearchProbe.class);
+        ParameterizedClass parameterizedClass = new ParameterizedClass("ClassLoadingSearchProbe");
+
+        Runnable instance = ParameterizedClass.newInstance(parameterizedClass,
+                                                           List.of("org.apache.cassandra.utils",
+                                                                   "org.apache.cassandra.config"),
+                                                           Runnable.class);
+        assertNotNull(instance);
+        assertThat(instance).isInstanceOf(org.apache.cassandra.config.ClassLoadingSearchProbe.class);
+        // The wrong-type class that was probed under the earlier package must not have been initialized.
+        assertThat(ClassLoadingTestSupport.wasInitialized(org.apache.cassandra.utils.ClassLoadingSearchProbe.class)).isFalse();
+    }
 }

test/unit/org/apache/cassandra/db/guardrails/GuardrailsConfigProviderTest.java

@@ -26,11 +26,25 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.GuardrailsOptions;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
 import static java.lang.String.format;
 
 public class GuardrailsConfigProviderTest extends GuardrailTester
 {
+    @Test
+    public void testBuildWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        Assertions.assertThatThrownBy(() -> GuardrailsConfigProvider.build(ClassLoadingTestNonAssignable.class.getName()))
+                  .isInstanceOf(ConfigurationException.class)
+                  .hasMessageContaining("must extend or implement " + GuardrailsConfigProvider.class.getName());
+
+        Assertions.assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
     @Test
     public void testBuildCustom() throws Throwable
     {

test/unit/org/apache/cassandra/db/guardrails/ValueGeneratorTest.java

@@ -26,9 +26,12 @@ import javax.annotation.Nonnull;
 import org.junit.Test;
 
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
 import static java.lang.String.format;
 import static org.apache.cassandra.db.guardrails.ValueGenerator.GENERATOR_CLASS_NAME_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -68,6 +71,21 @@ public class ValueGeneratorTest
         .message().isEqualTo(format("Unable to create instance of generator of class %s: does not contain property 'expecting_true'", BooleanGenerator.class.getName()));
     }
 
+    @Test
+    public void testGeneratorWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        CustomGuardrailConfig config = new CustomGuardrailConfig();
+        config.put(GENERATOR_CLASS_NAME_KEY, ClassLoadingTestNonAssignable.class.getName());
+
+        assertThatThrownBy(() -> ValueGenerator.getGenerator("probe generator", config))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + ValueGenerator.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
     public static class BooleanGenerator extends ValueGenerator<Boolean[]>
     {
         private final boolean expectingTrue;

test/unit/org/apache/cassandra/db/guardrails/ValueValidatorTest.java

@@ -26,10 +26,13 @@ import org.junit.Test;
 
 import org.apache.cassandra.db.guardrails.ValueValidator.ValidationViolation;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
 import static java.lang.Boolean.FALSE;
 import static java.lang.String.format;
 import static org.apache.cassandra.db.guardrails.ValueValidator.VALIDATOR_CLASS_NAME_KEY;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -83,6 +86,21 @@ public class ValueValidatorTest
         .message().isEqualTo(format("Unable to create instance of validator of class %s: does not contain property 'expecting_true'", BooleanValidator.class.getName()));
     }
 
+    @Test
+    public void testValidatorWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        CustomGuardrailConfig config = new CustomGuardrailConfig();
+        config.put(VALIDATOR_CLASS_NAME_KEY, ClassLoadingTestNonAssignable.class.getName());
+
+        assertThatThrownBy(() -> ValueValidator.getValidator("probe validator", config))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + ValueValidator.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
     public static class BooleanValidator extends ValueValidator<Boolean>
     {
         private final boolean expectingTrue;

test/unit/org/apache/cassandra/db/marshal/TypeParserTest.java

@@ -33,7 +33,10 @@ import org.apache.cassandra.dht.OrderPreservingPartitioner;
 import org.apache.cassandra.dht.RandomPartitioner;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.SyntaxException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -47,6 +50,17 @@ public class TypeParserTest
         DatabaseDescriptor.daemonInitialization();
     }
 
+    @Test
+    public void testRejectsNonAbstractTypeWithoutInitializing() throws SyntaxException
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> TypeParser.parse(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractType.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
     @Test
     public void testParse() throws ConfigurationException, SyntaxException
     {

test/unit/org/apache/cassandra/diag/ClassLoadingTestDiagnosticEvent.java

@@ -0,0 +1,44 @@
+/*
+ * 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.diag;
+
+import java.io.Serializable;
+import java.util.Map;
+
+/**
+ * A minimal, valid {@link DiagnosticEvent} subclass used by {@code DiagnosticEventPersistenceTest} to confirm that
+ * the type-checked load path in {@link DiagnosticEventPersistence} resolves a correct subtype. Lives in the
+ * {@code org.apache.cassandra} namespace so it passes the persistence package-prefix guard.
+ */
+public final class ClassLoadingTestDiagnosticEvent extends DiagnosticEvent
+{
+    public enum TestType { TEST }
+
+    @Override
+    public Enum<?> getType()
+    {
+        return TestType.TEST;
+    }
+
+    @Override
+    public Map<String, Serializable> toMap()
+    {
+        return Map.of();
+    }
+}

test/unit/org/apache/cassandra/diag/DiagnosticEventPersistenceTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.diag;
+
+import java.io.InvalidClassException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+import org.junit.Test;
+
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class DiagnosticEventPersistenceTest
+{
+    /**
+     * Reflectively drives the (private) bespoke type-checked load path
+     * {@link DiagnosticEventPersistence#getEventClass(String)}.
+     */
+    private static Class<?> getEventClass(String eventClazz) throws Throwable
+    {
+        Method method = DiagnosticEventPersistence.class.getDeclaredMethod("getEventClass", String.class);
+        method.setAccessible(true);
+        try
+        {
+            return (Class<?>) method.invoke(DiagnosticEventPersistence.instance(), eventClazz);
+        }
+        catch (InvocationTargetException e)
+        {
+            throw e.getCause();
+        }
+    }
+
+    @Test
+    public void testNonDiagnosticEventRejectedWithoutInitializing() throws Throwable
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> getEventClass(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(InvalidClassException.class)
+        .hasMessageContaining("must be of type DiagnosticEvent");
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testValidDiagnosticEventSubclassResolves() throws Throwable
+    {
+        Class<?> resolved = getEventClass(ClassLoadingTestDiagnosticEvent.class.getName());
+        assertThat(resolved).isEqualTo(ClassLoadingTestDiagnosticEvent.class);
+    }
+}

test/unit/org/apache/cassandra/index/SecondaryIndexManagerTest.java

@@ -38,13 +38,17 @@ import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.compaction.CompactionInfo;
 import org.apache.cassandra.db.lifecycle.SSTableSet;
+import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.notifications.SSTableAddedNotification;
 import org.apache.cassandra.schema.IndexMetadata;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 import org.apache.cassandra.utils.JVMStabilityInspector;
 import org.apache.cassandra.utils.KillerForTests;
 import org.apache.cassandra.utils.concurrent.Refs;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -58,6 +62,17 @@ public class SecondaryIndexManagerTest extends CQLTester
         TestingIndex.clear();
     }
 
+    @Test
+    public void rejectsNonIndexClassWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> SecondaryIndexManager.loadIndexClass(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + Index.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
     @Test
     public void createSasiAfterSai()
     {

test/unit/org/apache/cassandra/index/sasi/conf/IndexModeTest.java

@@ -31,10 +31,16 @@ import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.index.sasi.analyzer.AbstractAnalyzer;
+import org.apache.cassandra.index.sasi.analyzer.NonTokenizingAnalyzer;
 import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder.Mode;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 
 public class IndexModeTest
@@ -185,8 +191,8 @@ public class IndexModeTest
     {
         ColumnMetadata cd = ColumnMetadata.regularColumn(cfm, ByteBufferUtil.bytes("TestColumnMetadata"), BytesType.instance, ColumnMetadata.NO_UNIQUE_ID);
 
-        IndexMode result = IndexMode.getMode(cd, Collections.singletonMap("analyzer_class", "java.lang.Object"));
+        IndexMode result = IndexMode.getMode(cd, Collections.singletonMap("analyzer_class", NonTokenizingAnalyzer.class.getName()));
-        Assert.assertEquals(Object.class, result.analyzerClass);
+        Assert.assertEquals(NonTokenizingAnalyzer.class, result.analyzerClass);
         Assert.assertTrue(result.isAnalyzed);
         Assert.assertFalse(result.isLiteral);
         Assert.assertEquals((long)(1073741824 * 0.15), result.maxCompactionFlushMemoryInBytes);
@@ -198,16 +204,55 @@ public class IndexModeTest
     {
         ColumnMetadata cd = ColumnMetadata.regularColumn(cfm, ByteBufferUtil.bytes("TestColumnMetadata"), BytesType.instance, ColumnMetadata.NO_UNIQUE_ID);
 
-        IndexMode result = IndexMode.getMode(cd, ImmutableMap.of("analyzer_class", "java.lang.Object",
+        IndexMode result = IndexMode.getMode(cd, ImmutableMap.of("analyzer_class", NonTokenizingAnalyzer.class.getName(),
                                                                  "analyzed", "false"));
 
-        Assert.assertEquals(Object.class, result.analyzerClass);
+        Assert.assertEquals(NonTokenizingAnalyzer.class, result.analyzerClass);
         Assert.assertFalse(result.isAnalyzed);
         Assert.assertFalse(result.isLiteral);
         Assert.assertEquals((long)(1073741824 * 0.15), result.maxCompactionFlushMemoryInBytes);
         Assert.assertEquals(Mode.PREFIX, result.mode);
     }
 
+    @Test
+    public void test_bytesType_rejectsNonAnalyzerWithoutInitializing()
+    {
+        ColumnMetadata cd = ColumnMetadata.regularColumn(cfm, ByteBufferUtil.bytes("TestColumnMetadata"), BytesType.instance, ColumnMetadata.NO_UNIQUE_ID);
+
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> IndexMode.getMode(cd, Collections.singletonMap("analyzer_class", ClassLoadingTestNonAssignable.class.getName())))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractAnalyzer.class.getName());
+
+        Assert.assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void test_bytesType_missingAnalyzerFallsBack()
+    {
+        ColumnMetadata cd = ColumnMetadata.regularColumn(cfm, ByteBufferUtil.bytes("TestColumnMetadata"), BytesType.instance, ColumnMetadata.NO_UNIQUE_ID);
+
+        IndexMode result = IndexMode.getMode(cd, Collections.singletonMap("analyzer_class", "does.not.ExistAnalyzer"));
+        Assert.assertNull(result.analyzerClass);
+        Assert.assertFalse(result.isAnalyzed);
+        Assert.assertFalse(result.isLiteral);
+        Assert.assertEquals((long)(1073741824 * 0.15), result.maxCompactionFlushMemoryInBytes);
+        Assert.assertEquals(Mode.PREFIX, result.mode);
+    }
+
+    @Test
+    public void test_validateAnalyzer_rejectsNonAnalyzerWithoutInitializing()
+    {
+        ColumnMetadata cd = ColumnMetadata.regularColumn(cfm, ByteBufferUtil.bytes("TestColumnMetadata"), UTF8Type.instance, ColumnMetadata.NO_UNIQUE_ID);
+
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> IndexMode.validateAnalyzer(Collections.singletonMap("analyzer_class", ClassLoadingTestNonAssignable.class.getName()), cd))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractAnalyzer.class.getName());
+
+        Assert.assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
     @Test
     public void test_bytesType_maxCompactionFlushMemoryInBytes()
     {

test/unit/org/apache/cassandra/repair/autorepair/AutoRepairConfigTest.java

@@ -36,7 +36,11 @@ import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.repair.autorepair.AutoRepairConfig.Options;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -506,4 +510,17 @@ public class AutoRepairConfigTest extends CQLTester
         assertEquals(new DurationSpec.IntSecondsBound("3h"), config.global_settings.repair_session_timeout);
         assertEquals(new DurationSpec.IntSecondsBound("24h"), config.global_settings.min_repair_interval);
     }
+
+    @Test
+    public void testTokenRangeSplitterWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        ParameterizedClass pc = new ParameterizedClass(ClassLoadingTestNonAssignable.class.getName(), Collections.emptyMap());
+        assertThatThrownBy(() -> AutoRepairConfig.newAutoRepairTokenRangeSplitter(repairType, pc))
+        .isInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must extend or implement " + IAutoRepairTokenRangeSplitter.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
 }

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

@@ -0,0 +1,44 @@
+/*
+ * 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.schema;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class CompactionParamsTest
+{
+    @Test
+    public void testRejectsNonCompactionStrategyWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> CompactionParams.classFromName(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractCompactionStrategy.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+}

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

@@ -18,12 +18,19 @@
 
 package org.apache.cassandra.schema;
 
+import java.util.Collections;
+
 import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.compress.ICompressor;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class CompressionParamsTest
 {
@@ -71,4 +78,17 @@ public class CompressionParamsTest
         .as("Noop compression should not enable dictionary compression")
         .isFalse();
     }
+
+    @Test
+    public void testRejectsNonCompressorWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> CompressionParams.fromMap(Collections.singletonMap(CompressionParams.CLASS,
+                                                                                    ClassLoadingTestNonAssignable.class.getName())))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + ICompressor.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
 }

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

@@ -20,10 +20,21 @@
  */
 package org.apache.cassandra.schema;
 
+import java.util.Collections;
+
 import org.junit.Assert;
 import org.junit.Test;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.statements.schema.IndexTarget;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.index.Index;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class IndexMetadataTest
 {
@@ -33,4 +44,24 @@ public class IndexMetadataTest
         Assert.assertEquals("aB4__idx", IndexMetadata.generateDefaultIndexName("a B-4@!_+"));
         Assert.assertEquals("34_Ddd_F6_idx", IndexMetadata.generateDefaultIndexName("34_()Ddd", new ColumnIdentifier("#F%6*", true)));
     }
+
+    @Test
+    public void testRejectsNonCustomIndexWithoutInitializing()
+    {
+        TableMetadata table = TableMetadata.builder("ks", "tbl")
+                                           .addPartitionKeyColumn("pk", UTF8Type.instance)
+                                           .build();
+        IndexMetadata index = IndexMetadata.fromSchemaMetadata("idx",
+                                                               IndexMetadata.Kind.CUSTOM,
+                                                               Collections.singletonMap(IndexTarget.CUSTOM_INDEX_OPTION_NAME,
+                                                                                        ClassLoadingTestNonAssignable.class.getName()));
+
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> index.validate(table))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + Index.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
 }

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

@@ -0,0 +1,34 @@
+/*
+ * 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.schema;
+
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+public final class MemtableFactoryInvalidFieldType
+{
+    public static final Object FACTORY = new Object();
+
+    static
+    {
+        ClassLoadingTestSupport.markInitialized(MemtableFactoryInvalidFieldType.class);
+    }
+
+    private MemtableFactoryInvalidFieldType()
+    {
+    }
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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.schema;
+
+import java.util.Map;
+
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+public final class MemtableFactoryInvalidReturnType
+{
+    static
+    {
+        ClassLoadingTestSupport.markInitialized(MemtableFactoryInvalidReturnType.class);
+    }
+
+    private MemtableFactoryInvalidReturnType()
+    {
+    }
+
+    public static Object factory(Map<String, String> parameters)
+    {
+        return new Object();
+    }
+}

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

@@ -18,6 +18,8 @@
 
 package org.apache.cassandra.schema;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -32,11 +34,13 @@ import org.apache.cassandra.config.InheritingClass;
 import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.db.memtable.SkipListMemtableFactory;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 import org.apache.cassandra.utils.ConfigGenBuilder;
 
 import static accord.utils.Property.qt;
 import static org.apache.cassandra.config.YamlConfigurationLoader.fromMap;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -52,6 +56,46 @@ public class MemtableParamsTest
         assertEquals(ImmutableMap.of("default", DEFAULT), map);
     }
 
+    @Test
+    public void testInvalidFactoryMethodDoesNotInitializeClass() throws Exception
+    {
+        ClassLoadingTestSupport.assertNotInitialized(MemtableFactoryInvalidReturnType.class);
+
+        assertThatThrownBy(() -> getMemtableFactory(MemtableFactoryInvalidReturnType.class))
+        .isInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must return");
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(MemtableFactoryInvalidReturnType.class)).isFalse();
+    }
+
+    @Test
+    public void testInvalidFactoryFieldDoesNotInitializeClass() throws Exception
+    {
+        ClassLoadingTestSupport.assertNotInitialized(MemtableFactoryInvalidFieldType.class);
+
+        assertThatThrownBy(() -> getMemtableFactory(MemtableFactoryInvalidFieldType.class))
+        .isInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must be of type");
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(MemtableFactoryInvalidFieldType.class)).isFalse();
+    }
+
+    private static void getMemtableFactory(Class<?> memtableClass) throws Exception
+    {
+        Method method = MemtableParams.class.getDeclaredMethod("getMemtableFactory", ParameterizedClass.class);
+        method.setAccessible(true);
+        try
+        {
+            method.invoke(null, new ParameterizedClass(memtableClass.getName(), null));
+        }
+        catch (InvocationTargetException e)
+        {
+            if (e.getCause() instanceof ConfigurationException)
+                throw (ConfigurationException) e.getCause();
+            throw e;
+        }
+    }
+
     @Test
     public void testDefaultRemapped()
     {

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

@@ -0,0 +1,87 @@
+/*
+ * 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.schema;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import org.junit.Test;
+
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.tcm.serialization.Version;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+public class ReplicationParamsTest
+{
+    @Test
+    public void testRejectsNonReplicationStrategyWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> ReplicationParams.fromMap(Collections.singletonMap(ReplicationParams.CLASS,
+                                                                                    ClassLoadingTestNonAssignable.class.getName())))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractReplicationStrategy.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testSerializerRejectsNonReplicationStrategyWithoutInitializing() throws IOException
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> ReplicationParams.serializer.deserialize(replicationParamsInput(), Version.V8))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractReplicationStrategy.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    @Test
+    public void testMessageSerializerRejectsNonReplicationStrategyWithoutInitializing() throws IOException
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        assertThatThrownBy(() -> ReplicationParams.messageSerializer.deserialize(replicationParamsInput(),
+                                                                                MessagingService.current_version))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractReplicationStrategy.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
+    private static DataInputBuffer replicationParamsInput() throws IOException
+    {
+        try (DataOutputBuffer out = new DataOutputBuffer())
+        {
+            out.writeUTF(ClassLoadingTestNonAssignable.class.getName());
+            out.writeUnsignedVInt32(0);
+            return new DataInputBuffer(out.toByteArray());
+        }
+    }
+}

test/unit/org/apache/cassandra/security/CipherFactoryTest.java

@@ -35,12 +35,34 @@ import org.junit.Before;
 import org.junit.Test;
 
 import org.apache.cassandra.config.TransparentDataEncryptionOptions;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.ClassLoadingTestNonAssignable;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
 public class CipherFactoryTest
 {
+    @Test
+    public void keyProviderWrongTypeRejectedWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+
+        TransparentDataEncryptionOptions options = EncryptionContextGenerator.createEncryptionOptions();
+        options.key_provider.class_name = ClassLoadingTestNonAssignable.class.getName();
+
+        // CipherFactory wraps the load failure; the cause is the type-check ConfigurationException
+        assertThatThrownBy(() -> new CipherFactory(options))
+        .isInstanceOf(RuntimeException.class)
+        .hasCauseInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must extend or implement " + KeyProvider.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
     // http://www.gutenberg.org/files/4300/4300-h/4300-h.htm
     static final String ULYSSEUS = "Stately, plump Buck Mulligan came from the stairhead, bearing a bowl of lather on which a mirror and a razor lay crossed. " +
                                    "A yellow dressinggown, ungirdled, was sustained gently behind him on the mild morning air. He held the bowl aloft and intoned: " +

test/unit/org/apache/cassandra/security/CryptoProviderTest.java

@@ -167,7 +167,7 @@ public class CryptoProviderTest
             DatabaseDescriptor.getRawConfig().crypto_provider = new ParameterizedClass(DefaultCryptoProvider.class.getName(),
                                                                                        of("k1", "v1", "k2", "v2"));
 
-            fbUtilitiesMock.when(() -> FBUtilities.classForName(DefaultCryptoProvider.class.getName(), "crypto provider class"))
+            fbUtilitiesMock.when(() -> FBUtilities.classForNameWithoutInitialization(DefaultCryptoProvider.class.getName(), "crypto provider class", AbstractCryptoProvider.class))
                            .thenThrow(new RuntimeException("exception from test"));
 
             fbUtilitiesMock.when(() -> FBUtilities.newCryptoProvider(anyString(), anyMap())).thenCallRealMethod();

test/unit/org/apache/cassandra/triggers/TriggerExecutorTest.java

@@ -47,9 +47,11 @@ import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TriggerMetadata;
 import org.apache.cassandra.schema.Triggers;
+import org.apache.cassandra.utils.ClassLoadingTestSupport;
 import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
@@ -387,4 +389,28 @@ public class TriggerExecutorTest
             return cmp != 0 ? cmp : m1.key().compareTo(m2.key());
         }
     }
+
+    @Test
+    public void nonTriggerClassRejectedWithoutInitializing() throws Exception
+    {
+        ClassLoadingTestSupport.assertNotInitialized(NonTrigger.class);
+
+        assertThatExceptionOfType(ConfigurationException.class)
+        .isThrownBy(() -> TriggerExecutor.instance.loadTriggerClass(NonTrigger.class.getName()))
+        .withMessageContaining("must extend or implement " + ITrigger.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(NonTrigger.class)).isFalse();
+    }
+
+    public static class NonTrigger
+    {
+        static
+        {
+            ClassLoadingTestSupport.markInitialized(NonTrigger.class);
+        }
+
+        public NonTrigger()
+        {
+        }
+    }
 }

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

@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/**
+ * Search-package fall-through probe. Shares its simple name ({@code ClassLoadingSearchProbe}) with
+ * {@link org.apache.cassandra.config.ClassLoadingSearchProbe}, but is NOT a {@link Runnable}, so it represents the
+ * "wrong type under an earlier search package" case. Its static initializer records initialization so tests can
+ * confirm the search resolves without initializing this class.
+ */
+public final class ClassLoadingSearchProbe
+{
+    static
+    {
+        ClassLoadingTestSupport.markInitialized(ClassLoadingSearchProbe.class);
+    }
+}

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+public final class ClassLoadingTestNonAssignable
+{
+    static
+    {
+        ClassLoadingTestSupport.markInitialized(ClassLoadingTestNonAssignable.class);
+    }
+
+    private ClassLoadingTestNonAssignable()
+    {
+    }
+}

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

@@ -0,0 +1,47 @@
+/*
+ * 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.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class ClassLoadingTestSupport
+{
+    private static final Set<String> initializedClasses = ConcurrentHashMap.newKeySet();
+
+    private ClassLoadingTestSupport()
+    {
+    }
+
+    public static void markInitialized(Class<?> initializedClass)
+    {
+        initializedClasses.add(initializedClass.getName());
+    }
+
+    public static boolean wasInitialized(Class<?> initializedClass)
+    {
+        return initializedClasses.contains(initializedClass.getName());
+    }
+
+    public static void assertNotInitialized(Class<?> initializedClass)
+    {
+        if (wasInitialized(initializedClass))
+            throw new AssertionError(initializedClass.getName() + " has already been initialized");
+    }
+}

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

@@ -25,6 +25,7 @@ import java.nio.charset.CharacterCodingException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -46,6 +47,7 @@ import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.audit.IAuditLogger;
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -59,9 +61,14 @@ import org.apache.cassandra.dht.LocalPartitioner;
 import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.dht.OrderPreservingPartitioner;
 import org.apache.cassandra.dht.RandomPartitioner;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.security.AbstractCryptoProvider;
+import org.apache.cassandra.security.ISslContextFactory;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 
 public class FBUtilitiesTest
@@ -69,6 +76,73 @@ public class FBUtilitiesTest
 
     public static final Logger LOGGER = LoggerFactory.getLogger(FBUtilitiesTest.class);
 
+    @Test
+    public void testTypedClassForNameRejectsWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.classForNameWithoutInitialization(ClassLoadingTestNonAssignable.class.getName(), "test class", Runnable.class))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + Runnable.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void testTypedConstructRejectsWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.construct(ClassLoadingTestNonAssignable.class.getName(), "test class", Runnable.class))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + Runnable.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void testTypedInstanceOrConstructRejectsWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.instanceOrConstruct(ClassLoadingTestNonAssignable.class.getName(), "test class", Runnable.class))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + Runnable.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void testNewAuditLoggerRejectsWrongTypeWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.newAuditLogger(ClassLoadingTestNonAssignable.class.getName(), Collections.emptyMap()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasRootCauseInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must extend or implement " + IAuditLogger.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void testNewSslContextFactoryRejectsWrongTypeWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.newSslContextFactory(ClassLoadingTestNonAssignable.class.getName(), Collections.emptyMap()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must extend or implement " + ISslContextFactory.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
+    @Test
+    public void testNewCryptoProviderRejectsWrongTypeWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.newCryptoProvider(ClassLoadingTestNonAssignable.class.getName(), Collections.emptyMap()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasMessageContaining("must extend or implement " + AbstractCryptoProvider.class.getName());
+
+        assertFalse(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class));
+    }
+
     @Test
     public void testCompareByteSubArrays()
     {

tools/sstableloader/src/org/apache/cassandra/tools/LoaderOptions.java

@@ -50,6 +50,7 @@ import org.apache.cassandra.config.YamlConfigurationLoader;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.config.DataRateSpec.DataRateUnit.MEBIBYTES_PER_SECOND;
 import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions.ClientAuth.REQUIRED;
@@ -715,11 +716,12 @@ public class LoaderOptions
                 {
                     try
                     {
-                        Class authProviderClass = Class.forName(authProviderName);
+                        Class<? extends AuthProvider> authProviderClass =
-                        Constructor constructor = authProviderClass.getConstructor(String.class, String.class);
+                            FBUtilities.classForNameWithoutInitialization(authProviderName, "auth provider", AuthProvider.class);
-                        authProvider = (AuthProvider)constructor.newInstance(user, passwd);
+                        Constructor<? extends AuthProvider> constructor = authProviderClass.getConstructor(String.class, String.class);
+                        authProvider = constructor.newInstance(user, passwd);
                     }
-                    catch (ClassNotFoundException e)
+                    catch (ConfigurationException e)
                     {
                         errorMsg("Unknown auth provider: " + e.getMessage(), getCmdLineOptions());
                     }
@@ -746,9 +748,9 @@ public class LoaderOptions
             {
                 try
                 {
-                    authProvider = (AuthProvider)Class.forName(authProviderName).newInstance();
+                    authProvider = FBUtilities.construct(authProviderName, "auth provider", AuthProvider.class);
                 }
-                catch (ClassNotFoundException | InstantiationException | IllegalAccessException e)
+                catch (ConfigurationException e)
                 {
                     errorMsg("Unknown auth provider: " + e.getMessage(), getCmdLineOptions());
                 }

tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java

@@ -28,6 +28,7 @@ import java.util.Map;
 import com.google.common.base.Function;
 
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.utils.FBUtilities;
 
 /**
  * For specifying replication options
@@ -76,9 +77,9 @@ class OptionReplication extends OptionMulti
             {
                 try
                 {
-                    Class<?> clazz = Class.forName(fullname);
+                    FBUtilities.classForNameWithoutInitialization(fullname,
-                    if (!AbstractReplicationStrategy.class.isAssignableFrom(clazz))
+                                                                  "replication strategy",
-                        throw new IllegalArgumentException(clazz + " is not a replication strategy");
+                                                                  AbstractReplicationStrategy.class);
                     strategy = fullname;
                     break;
                 } catch (Exception ignore)

tools/stress/src/org/apache/cassandra/stress/settings/SettingsMode.java

@@ -33,6 +33,7 @@ import com.datastax.driver.core.ProtocolOptions;
 import com.datastax.driver.core.ProtocolVersion;
 
 import org.apache.cassandra.stress.util.ResultLogger;
+import org.apache.cassandra.utils.FBUtilities;
 
 import static java.lang.String.format;
 import static org.apache.cassandra.stress.settings.SettingsCredentials.CQL_PASSWORD_PROPERTY_KEY;
@@ -91,9 +92,10 @@ public class SettingsMode implements Serializable
             {
                 try
                 {
-                    Class<?> clazz = Class.forName(authProviderClassname);
+                    Class<? extends AuthProvider> clazz =
-                    if (!AuthProvider.class.isAssignableFrom(clazz))
+                        FBUtilities.classForNameWithoutInitialization(authProviderClassname,
-                        throw new IllegalArgumentException(clazz + " is not a valid auth provider");
+                                                                      "auth provider",
+                                                                      AuthProvider.class);
                     // check we can instantiate it
                     if (PlainTextAuthProvider.class.equals(clazz))
                     {