diff --git a/core/src/main/scala/akka/persistence/jdbc/testkit/internal/SchemaUtilsImpl.scala b/core/src/main/scala/akka/persistence/jdbc/testkit/internal/SchemaUtilsImpl.scala
index 521725a55..f6a0be236 100644
--- a/core/src/main/scala/akka/persistence/jdbc/testkit/internal/SchemaUtilsImpl.scala
+++ b/core/src/main/scala/akka/persistence/jdbc/testkit/internal/SchemaUtilsImpl.scala
@@ -30,17 +30,20 @@ import slick.jdbc.SQLServerProfile
 @InternalApi
 private[jdbc] object SchemaUtilsImpl {
 
-  def legacy(config: Config): Boolean =
-    config.getString("jdbc-journal.dao") != "akka.persistence.jdbc.journal.dao.DefaultJournalDao"
+  def legacy(configKey: String, config: Config): Boolean =
+    config.getConfig(configKey).getString("dao") != "akka.persistence.jdbc.journal.dao.DefaultJournalDao"
 
   /**
    * INTERNAL API
    */
   @InternalApi
-  private[jdbc] def dropIfExists(logger: Logger)(implicit actorSystem: ClassicActorSystemProvider): Future[Done] = {
-    val slickDb: SlickDatabase = loadSlickDatabase("jdbc-journal")
+  private[jdbc] def dropIfExists(configKey: String, logger: Logger)(
+      implicit actorSystem: ClassicActorSystemProvider): Future[Done] = {
+    val slickDb: SlickDatabase = loadSlickDatabase(configKey)
     val (fileToLoad, separator) =
-      dropScriptFor(slickProfileToSchemaType(slickDb.profile), legacy(actorSystem.classicSystem.settings.config))
+      dropScriptFor(
+        slickProfileToSchemaType(slickDb.profile),
+        legacy(configKey, actorSystem.classicSystem.settings.config))
 
     val blockingEC = actorSystem.classicSystem.dispatchers.lookup(Dispatchers.DefaultBlockingDispatcherId)
     Future(applyScriptWithSlick(fromClasspathAsString(fileToLoad), separator, logger, slickDb.database))(blockingEC)
@@ -50,12 +53,14 @@ private[jdbc] object SchemaUtilsImpl {
    * INTERNAL API
    */
   @InternalApi
-  private[jdbc] def createIfNotExists(logger: Logger)(
+  private[jdbc] def createIfNotExists(configKey: String, logger: Logger)(
       implicit actorSystem: ClassicActorSystemProvider): Future[Done] = {
 
-    val slickDb: SlickDatabase = loadSlickDatabase("jdbc-journal")
+    val slickDb: SlickDatabase = loadSlickDatabase(configKey)
     val (fileToLoad, separator) =
-      createScriptFor(slickProfileToSchemaType(slickDb.profile), legacy(actorSystem.classicSystem.settings.config))
+      createScriptFor(
+        slickProfileToSchemaType(slickDb.profile),
+        legacy(configKey, actorSystem.classicSystem.settings.config))
 
     val blockingEC = actorSystem.classicSystem.dispatchers.lookup(Dispatchers.DefaultBlockingDispatcherId)
     Future(applyScriptWithSlick(fromClasspathAsString(fileToLoad), separator, logger, slickDb.database))(blockingEC)
diff --git a/core/src/main/scala/akka/persistence/jdbc/testkit/javadsl/SchemaUtils.scala b/core/src/main/scala/akka/persistence/jdbc/testkit/javadsl/SchemaUtils.scala
index 2bea3cc3b..b7829e4b9 100644
--- a/core/src/main/scala/akka/persistence/jdbc/testkit/javadsl/SchemaUtils.scala
+++ b/core/src/main/scala/akka/persistence/jdbc/testkit/javadsl/SchemaUtils.scala
@@ -34,7 +34,26 @@ object SchemaUtils {
    */
   @ApiMayChange
   def dropIfExists(actorSystem: ClassicActorSystemProvider): CompletionStage[Done] =
-    SchemaUtilsImpl.dropIfExists(logger)(actorSystem).toJava
+    dropIfExists(configKey = "jdbc-journal", actorSystem)
+
+  /**
+   * Drops the schema for both the journal and the snapshot table using the default schema definition.
+   *
+   * For information about the different schemas and supported databases consult
+   * https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#database-schema
+   *
+   * This utility method is intended to be used for testing only.
+   * For production, it's recommended to run any DDL statements before the system is started.
+   *
+   * This method will automatically detects the configured database using the settings from `configKey` config.
+   * If configured with `use-shared-db`, it will use the `akka-persistence-jdbc.shared-databases` definition instead.
+   * See https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#sharing-the-database-connection-pool-between-the-journals for details.
+   *
+   * @param configKey the database journal configuration key to use.
+   */
+  @ApiMayChange
+  def dropIfExists(configKey: String, actorSystem: ClassicActorSystemProvider): CompletionStage[Done] =
+    SchemaUtilsImpl.dropIfExists(configKey, logger)(actorSystem).toJava
 
   /**
    * Creates the schema for both the journal and the snapshot table using the default schema definition.
@@ -51,7 +70,26 @@ object SchemaUtils {
    */
   @ApiMayChange
   def createIfNotExists(actorSystem: ClassicActorSystemProvider): CompletionStage[Done] =
-    SchemaUtilsImpl.createIfNotExists(logger)(actorSystem).toJava
+    createIfNotExists("jdbc-journal", actorSystem)
+
+  /**
+   * Creates the schema for both the journal and the snapshot table using the default schema definition.
+   *
+   * For information about the different schemas and supported databases consult
+   * https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#database-schema
+   *
+   * This utility method is intended to be used for testing only.
+   * For production, it's recommended to create run DDL statements before the system is started.
+   *
+   * This method will automatically detects the configured database using the settings from `configKey` config.
+   * If configured with `use-shared-db`, it will use the `akka-persistence-jdbc.shared-databases` definition instead.
+   * See https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#sharing-the-database-connection-pool-between-the-journals for details.
+   *
+   * @param configKey the database journal configuration key to use.
+   */
+  @ApiMayChange
+  def createIfNotExists(configKey: String, actorSystem: ClassicActorSystemProvider): CompletionStage[Done] =
+    SchemaUtilsImpl.createIfNotExists(configKey, logger)(actorSystem).toJava
 
   /**
    * This method can be used to load alternative DDL scripts.
diff --git a/core/src/main/scala/akka/persistence/jdbc/testkit/scaladsl/SchemaUtils.scala b/core/src/main/scala/akka/persistence/jdbc/testkit/scaladsl/SchemaUtils.scala
index 2f996f8ef..59e5a0479 100644
--- a/core/src/main/scala/akka/persistence/jdbc/testkit/scaladsl/SchemaUtils.scala
+++ b/core/src/main/scala/akka/persistence/jdbc/testkit/scaladsl/SchemaUtils.scala
@@ -32,7 +32,26 @@ object SchemaUtils {
    */
   @ApiMayChange
   def dropIfExists()(implicit actorSystem: ClassicActorSystemProvider): Future[Done] =
-    SchemaUtilsImpl.dropIfExists(logger)
+    dropIfExists(configKey = "jdbc-journal")
+
+  /**
+   * Drops the schema for both the journal and the snapshot table using the default schema definition.
+   *
+   * For information about the different schemas and supported databases consult
+   * https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#database-schema
+   *
+   * This utility method is intended to be used for testing only.
+   * For production, it's recommended to run any DDL statements before the system is started.
+   *
+   * This method will automatically detects the configured database using the settings from `configKey` config.
+   * If configured with `use-shared-db`, it will use the `akka-persistence-jdbc.shared-databases` definition instead.
+   * See https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#sharing-the-database-connection-pool-between-the-journals for details.
+   *
+   * @param configKey the database journal configuration key to use.
+   */
+  @ApiMayChange
+  def dropIfExists(configKey: String)(implicit actorSystem: ClassicActorSystemProvider): Future[Done] =
+    SchemaUtilsImpl.dropIfExists(configKey, logger)
 
   /**
    * Creates the schema for both the journal and the snapshot table using the default schema definition.
@@ -49,7 +68,26 @@ object SchemaUtils {
    */
   @ApiMayChange
   def createIfNotExists()(implicit actorSystem: ClassicActorSystemProvider): Future[Done] =
-    SchemaUtilsImpl.createIfNotExists(logger)
+    createIfNotExists(configKey = "jdbc-journal")
+
+  /**
+   * Creates the schema for both the journal and the snapshot table using the default schema definition.
+   *
+   * For information about the different schemas and supported databases consult
+   * https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#database-schema
+   *
+   * This utility method is intended to be used for testing only.
+   * For production, it's recommended to run any DDL statements before the system is started.
+   *
+   * This method will automatically detects the configured database using the settings from `configKey` config.
+   * If configured with `use-shared-db`, it will use the `akka-persistence-jdbc.shared-databases` definition instead.
+   * See https://doc.akka.io/docs/akka-persistence-jdbc/current/index.html#sharing-the-database-connection-pool-between-the-journals for details.
+   *
+   * @param configKey the database journal configuration key to use.
+   */
+  @ApiMayChange
+  def createIfNotExists(configKey: String)(implicit actorSystem: ClassicActorSystemProvider): Future[Done] =
+    SchemaUtilsImpl.createIfNotExists(configKey, logger)
 
   /**
    * This method can be used to load alternative DDL scripts.
diff --git a/core/src/test/scala/akka/persistence/jdbc/util/DropCreate.scala b/core/src/test/scala/akka/persistence/jdbc/util/DropCreate.scala
index 307c11c5f..c721799e9 100644
--- a/core/src/test/scala/akka/persistence/jdbc/util/DropCreate.scala
+++ b/core/src/test/scala/akka/persistence/jdbc/util/DropCreate.scala
@@ -25,7 +25,7 @@ private[jdbc] trait DropCreate {
   def db: Database
   def config: Config
 
-  def newDao: Boolean = !SchemaUtilsImpl.legacy(config)
+  def newDao: Boolean = !SchemaUtilsImpl.legacy("jdbc-journal", config)
 
   /**
    * INTERNAL API