Verify compression provider type before initialization

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

@@ -744,8 +744,9 @@ public class FBUtilities
             if (!className.contains("."))
                 className = "org.apache.cassandra.io.compress." + className;
 
-            Class<?> compressionProviderClass = FBUtilities.classForName(className, "compression service provider");
+            Class<? extends AbstractCompressionProvider> compressionProviderClass =
-            return (AbstractCompressionProvider) compressionProviderClass.getConstructor().newInstance();
+                FBUtilities.classForNameWithoutInitialization(className, "compression service provider", AbstractCompressionProvider.class);
+            return compressionProviderClass.getConstructor().newInstance();
         }
         catch (ConfigurationException e)
         {

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

@@ -61,6 +61,7 @@ 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.io.compress.AbstractCompressionProvider;
 import org.apache.cassandra.security.AbstractCryptoProvider;
 import org.apache.cassandra.security.ISslContextFactory;
 
@@ -141,6 +142,17 @@ public class FBUtilitiesTest
         assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
     }
 
+    @Test
+    public void testNewCompressionProviderRejectsWrongTypeWithoutInitializing()
+    {
+        ClassLoadingTestSupport.assertNotInitialized(ClassLoadingTestNonAssignable.class);
+        assertThatThrownBy(() -> FBUtilities.newCompressionProvider(ClassLoadingTestNonAssignable.class.getName()))
+        .isInstanceOf(ConfigurationException.class)
+        .hasStackTraceContaining("must extend or implement " + AbstractCompressionProvider.class.getName());
+
+        assertThat(ClassLoadingTestSupport.wasInitialized(ClassLoadingTestNonAssignable.class)).isFalse();
+    }
+
     @Test
     public void testCompareByteSubArrays()
     {