diff --git a/core/src/main/java/org/apache/druid/utils/ConnectionUriUtils.java b/core/src/main/java/org/apache/druid/utils/ConnectionUriUtils.java
index 0cd201dbece..d0bfd98d107 100644
--- a/core/src/main/java/org/apache/druid/utils/ConnectionUriUtils.java
+++ b/core/src/main/java/org/apache/druid/utils/ConnectionUriUtils.java
@@ -43,15 +43,13 @@ public final class ConnectionUriUtils
   public static final String MARIADB_PREFIX = "jdbc:mariadb:";
 
   public static final String POSTGRES_DRIVER = "org.postgresql.Driver";
-  public static final String MYSQL_DRIVER = "com.mysql.jdbc.Driver";
   public static final String MYSQL_NON_REGISTERING_DRIVER = "com.mysql.jdbc.NonRegisteringDriver";
-  public static final String MARIADB_DRIVER = "org.mariadb.jdbc.Driver";
 
   /**
    * This method checks {@param actualProperties} against {@param allowedProperties} if they are not system properties.
    * A property is regarded as a system property if its name starts with a prefix in {@param systemPropertyPrefixes}.
    * See org.apache.druid.server.initialization.JDBCAccessSecurityConfig for more details.
-   *
+   * <p>
    * If a non-system property that is not allowed is found, this method throws an {@link IllegalArgumentException}.
    */
   public static void throwIfPropertiesAreNotAllowed(
@@ -76,16 +74,16 @@ public final class ConnectionUriUtils
    * This method tries to determine the correct type of database for a given JDBC connection string URI, then load the
    * driver using reflection to parse the uri parameters, returning the set of keys which can be used for JDBC
    * parameter whitelist validation.
-   *
+   * <p>
    * uris starting with {@link #MYSQL_PREFIX} will first try to use the MySQL Connector/J driver (5.x), then fallback
    * to MariaDB Connector/J (version 2.x) which also accepts jdbc:mysql uris. This method does not attempt to use
    * MariaDB Connector/J 3.x alpha driver (at the time of these javadocs, it only handles the jdbc:mariadb prefix)
-   *
+   * <p>
    * uris starting with {@link #POSTGRES_PREFIX} will use the postgresql driver to parse the uri
-   *
+   * <p>
    * uris starting with {@link #MARIADB_PREFIX} will first try to use MariaDB Connector/J driver (2.x) then fallback to
    * MariaDB Connector/J 3.x driver.
-   *
+   * <p>
    * If the uri does not match any of these schemes, this method will return an empty set if unknown uris are allowed,
    * or throw an exception if not.
    */
@@ -185,7 +183,11 @@ public final class ConnectionUriUtils
     Class<?> driverClass = Class.forName(MYSQL_NON_REGISTERING_DRIVER);
     Method parseUrl = driverClass.getMethod("parseURL", String.class, Properties.class);
     // almost the same as postgres, but is an instance level method
-    Properties properties = (Properties) parseUrl.invoke(driverClass.getConstructor().newInstance(), connectionUri, null);
+    Properties properties = (Properties) parseUrl.invoke(
+        driverClass.getConstructor().newInstance(),
+        connectionUri,
+        null
+    );
 
     if (properties == null) {
       throw new IAE("Invalid URL format for MySQL: [%s]", connectionUri);
diff --git a/extensions-core/mysql-metadata-storage/pom.xml b/extensions-core/mysql-metadata-storage/pom.xml
index f66d3825725..e00f8367789 100644
--- a/extensions-core/mysql-metadata-storage/pom.xml
+++ b/extensions-core/mysql-metadata-storage/pom.xml
@@ -121,6 +121,11 @@
             <version>${mariadb.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>
diff --git a/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnector.java b/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnector.java
index a88a9ed4721..07f48f36cd3 100644
--- a/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnector.java
+++ b/extensions-core/mysql-metadata-storage/src/main/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnector.java
@@ -26,6 +26,7 @@ import com.fasterxml.jackson.annotation.JsonTypeName;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.apache.druid.metadata.MetadataStorageConnectorConfig;
 import org.apache.druid.metadata.SQLFirehoseDatabaseConnector;
+import org.apache.druid.metadata.storage.mysql.MySQLConnectorDriverConfig;
 import org.apache.druid.server.initialization.JdbcAccessSecurityConfig;
 import org.apache.druid.utils.ConnectionUriUtils;
 import org.skife.jdbi.v2.DBI;
@@ -47,7 +48,8 @@ public class MySQLFirehoseDatabaseConnector extends SQLFirehoseDatabaseConnector
   public MySQLFirehoseDatabaseConnector(
       @JsonProperty("connectorConfig") MetadataStorageConnectorConfig connectorConfig,
       @JsonProperty("driverClassName") @Nullable String driverClassName,
-      @JacksonInject JdbcAccessSecurityConfig securityConfig
+      @JacksonInject JdbcAccessSecurityConfig securityConfig,
+      @JacksonInject MySQLConnectorDriverConfig mySQLConnectorDriverConfig
   )
   {
     this.connectorConfig = connectorConfig;
@@ -56,10 +58,8 @@ public class MySQLFirehoseDatabaseConnector extends SQLFirehoseDatabaseConnector
     datasource.setDriverClassLoader(getClass().getClassLoader());
     if (driverClassName != null) {
       datasource.setDriverClassName(driverClassName);
-    } else if (connectorConfig.getConnectURI().startsWith(ConnectionUriUtils.MARIADB_PREFIX)) {
-      datasource.setDriverClassName(ConnectionUriUtils.MARIADB_DRIVER);
     } else {
-      datasource.setDriverClassName(ConnectionUriUtils.MYSQL_DRIVER);
+      datasource.setDriverClassName(mySQLConnectorDriverConfig.getDriverClassName());
     }
     this.dbi = new DBI(datasource);
   }
diff --git a/extensions-core/mysql-metadata-storage/src/test/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnectorTest.java b/extensions-core/mysql-metadata-storage/src/test/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnectorTest.java
index b46cb268562..035aec7d6aa 100644
--- a/extensions-core/mysql-metadata-storage/src/test/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnectorTest.java
+++ b/extensions-core/mysql-metadata-storage/src/test/java/org/apache/druid/firehose/sql/MySQLFirehoseDatabaseConnectorTest.java
@@ -27,23 +27,33 @@ import nl.jqno.equalsverifier.EqualsVerifier;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.metadata.MetadataStorageConnectorConfig;
+import org.apache.druid.metadata.storage.mysql.MySQLConnectorDriverConfig;
 import org.apache.druid.metadata.storage.mysql.MySQLMetadataStorageModule;
 import org.apache.druid.server.initialization.JdbcAccessSecurityConfig;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
 
 import java.util.Set;
 
+@RunWith(MockitoJUnitRunner.class)
 public class MySQLFirehoseDatabaseConnectorTest
 {
-  private static final ObjectMapper MAPPER = new DefaultObjectMapper();
   private static final JdbcAccessSecurityConfig INJECTED_CONF = newSecurityConfigEnforcingAllowList(ImmutableSet.of());
 
-  static {
-    MAPPER.registerModules(new MySQLMetadataStorageModule().getJacksonModules());
-    MAPPER.setInjectableValues(new InjectableValues.Std().addValue(JdbcAccessSecurityConfig.class, INJECTED_CONF));
+  @Mock
+  private MySQLConnectorDriverConfig mySQLConnectorDriverConfig;
+
+  @Before
+  public void setup()
+  {
+    Mockito.doReturn("com.mysql.jdbc.Driver").when(mySQLConnectorDriverConfig).getDriverClassName();
   }
 
   @Rule
@@ -52,6 +62,13 @@ public class MySQLFirehoseDatabaseConnectorTest
   @Test
   public void testSerde() throws JsonProcessingException
   {
+    ObjectMapper mapper = new DefaultObjectMapper();
+    mapper.registerModules(new MySQLMetadataStorageModule().getJacksonModules());
+    mapper.setInjectableValues(new InjectableValues.Std().addValue(JdbcAccessSecurityConfig.class, INJECTED_CONF)
+                                                         .addValue(
+                                                             MySQLConnectorDriverConfig.class,
+                                                             mySQLConnectorDriverConfig
+                                                         ));
     MetadataStorageConnectorConfig connectorConfig = new MetadataStorageConnectorConfig()
     {
       @Override
@@ -63,18 +80,23 @@ public class MySQLFirehoseDatabaseConnectorTest
     MySQLFirehoseDatabaseConnector connector = new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        INJECTED_CONF
+        INJECTED_CONF,
+        mySQLConnectorDriverConfig
+    );
+    MySQLFirehoseDatabaseConnector andBack = mapper.readValue(
+        mapper.writeValueAsString(connector),
+        MySQLFirehoseDatabaseConnector.class
     );
-    MySQLFirehoseDatabaseConnector andBack = MAPPER.readValue(MAPPER.writeValueAsString(connector), MySQLFirehoseDatabaseConnector.class);
     Assert.assertEquals(connector, andBack);
 
     // test again with classname
     connector = new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         "some.class.name.Driver",
-        INJECTED_CONF
+        INJECTED_CONF,
+        mySQLConnectorDriverConfig
     );
-    andBack = MAPPER.readValue(MAPPER.writeValueAsString(connector), MySQLFirehoseDatabaseConnector.class);
+    andBack = mapper.readValue(mapper.writeValueAsString(connector), MySQLFirehoseDatabaseConnector.class);
     Assert.assertEquals(connector, andBack);
   }
 
@@ -105,7 +127,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -126,7 +149,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -150,7 +174,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -173,7 +198,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -196,12 +222,12 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
 
-
   @Test
   public void testFailOnlyInvalidProperty()
   {
@@ -222,7 +248,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -246,7 +273,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -270,7 +298,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -304,7 +333,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        securityConfig
+        securityConfig,
+        mySQLConnectorDriverConfig
     );
   }
 
@@ -326,7 +356,8 @@ public class MySQLFirehoseDatabaseConnectorTest
     new MySQLFirehoseDatabaseConnector(
         connectorConfig,
         null,
-        new JdbcAccessSecurityConfig()
+        new JdbcAccessSecurityConfig(),
+        mySQLConnectorDriverConfig
     );
   }