JDBC_PING as default discovery protocol

Closes #29399

- Add ProviderFactory#dependsOn to allow dependencies between
  ProviderFactories to be explicitly defined
- Disable Infinispan default shutdownhook disabled to ensure lifecycle
  is managed exclusively by Keycloak
- Remove Infinispan shutdown hook in KeycloakRecorder and manage
  EmbeddedCacheManager lifecycle only in DefaultInfinispanConnectionProviderFactory#close

Signed-off-by: Ryan Emerson <remerson@redhat.com>
Signed-off-by: Alexander Schwartz <aschwart@redhat.com>
Co-authored-by: Alexander Schwartz <aschwart@redhat.com>

.github/workflows/ci.yml

@@ -596,6 +596,17 @@ jobs:
           echo "Tests: $TESTS"
           ./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Pdb-${{ matrix.db }} "-Dwebdriver.chrome.driver=$CHROMEWEBDRIVER/chromedriver" -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
 
+      - name: Run cluster JDBC_PING2 smoke test
+        run: |
+          ./mvnw test ${{ env.SUREFIRE_RETRY }} \
+            -Pauth-server-cluster-quarkus \
+            -Pdb-${{ matrix.db }} \
+            -Dtest=RealmInvalidationClusterTest \
+            -Dsession.cache.owners=2 \
+            -Dauth.server.quarkus.cluster.stack=jdbc-ping \
+            -pl testsuite/integration-arquillian/tests/base \
+            2>&1 | misc/log/trimmer.sh
+
       - name: Upload JVM Heapdumps
         if: always()
         uses: ./.github/actions/upload-heapdumps

docs/documentation/upgrading/topics/changes/changes-26_1_0.adoc

@@ -28,3 +28,20 @@ To disable the virtual threads, add one of the Java system properties combinatio
 * `-Dorg.infinispan.threads.virtual=false`: disables virtual thread in both Infinispan and JGroups.
 * `-Djgroups.thread.virtual=false`: disables virtual threads only in JGroups.
 * `-Dorg.infinispan.threads.virtual=false -Djgroups.thread.virtual=true`: disables virtual threads only in Infinispan.
+
+= Default transport stack changed to JDBC_PING2 for distributed caches
+
+Previous versions of {project_name} used as a default UDP multicast to discover other nodes to form a cluster and to synchronize the replicated caches of {project_name}.
+This required multicast to be available and to be configured correctly, which is usually not the case in cloud environments.
+
+Starting with this version, the default changes to a configuration of JDBC_PING2 which uses {project_name}'s database to discover other nodes.
+As this removes the need for multicast network capabilities, this is a simplification and a drop-in replacement.
+
+To enable the previous behavior, choose the transport stack `udp`.
+
+The {project_name} Operator will continue to configure `kubernetes` as a transport stack.
+
+= Defining dependencies between provider factories
+
+When developing extensions for {project_name}, developers can now specify dependencies between provider factories classes by implementing the method `dependsOn()` in the `ProviderFactory` interface.
+See the Javadoc for a detailed description.

docs/guides/server/caching.adoc

@@ -219,7 +219,7 @@ To apply a specific cache stack, enter this command:
 
 <@kc.start parameters="--cache-stack=<stack>"/>
 
-The default stack is set to `udp` when distributed caches are enabled.
+The default stack is set to `jdbc-ping` when distributed caches are enabled.
 
 === Available transport stacks
 
@@ -229,17 +229,19 @@ The following table shows transport stacks that are available without any furthe
 |===
 |Stack name|Transport protocol|Discovery
 
-|tcp|TCP|MPING (uses UDP multicast).
+|`tcp`|TCP|MPING (uses UDP multicast).
-|udp|UDP|UDP multicast
+|`udp`|UDP|UDP multicast
+|`jdbc-ping`|UDP|JDBC_PING2
 |===
 
+
 The following table shows transport stacks that are available using the `--cache-stack` runtime option and a minimum configuration:
 
 [%autowidth]
 |===
 |Stack name|Transport protocol|Discovery
 
-|kubernetes|TCP|DNS_PING (requires `-Djgroups.dns.query=<headless-service-FQDN>` to be added to JAVA_OPTS or JAVA_OPTS_APPEND environment variable).
+|`kubernetes`|TCP|DNS_PING (requires `-Djgroups.dns.query=<headless-service-FQDN>` to be added to JAVA_OPTS or JAVA_OPTS_APPEND environment variable).
 |===
 
 === Additional transport stacks
@@ -252,9 +254,9 @@ Instead, when you have a distributed cache setup running on AWS EC2 instances, y
 |===
 |Stack name|Transport protocol|Discovery
 
-|ec2|TCP|NATIVE_S3_PING
+|`ec2`|TCP|NATIVE_S3_PING
-|google|TCP|GOOGLE_PING2
+|`google`|TCP|GOOGLE_PING2
-|azure|TCP|AZURE_PING
+|`azure`|TCP|AZURE_PING
 |===
 
 Cloud vendor specific stacks have additional dependencies for {project_name}.

model/infinispan/pom.xml

@@ -43,6 +43,10 @@
             <groupId>org.keycloak</groupId>
             <artifactId>keycloak-model-storage</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.keycloak</groupId>
+            <artifactId>keycloak-model-jpa</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.keycloak</groupId>
             <artifactId>keycloak-model-storage-private</artifactId>

model/infinispan/src/main/java/org/keycloak/connections/infinispan/DefaultInfinispanConnectionProviderFactory.java

@@ -20,6 +20,7 @@ package org.keycloak.connections.infinispan;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.ServiceLoader;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -46,6 +47,7 @@ import org.keycloak.cluster.ClusterEvent;
 import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.cluster.ManagedCacheManagerProvider;
 import org.keycloak.connections.infinispan.remote.RemoteInfinispanConnectionProvider;
+import org.keycloak.connections.jpa.JpaConnectionProvider;
 import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.marshalling.Marshalling;
 import org.keycloak.models.KeycloakSession;
@@ -56,6 +58,7 @@ import org.keycloak.models.cache.infinispan.events.RealmUpdatedEvent;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.provider.InvalidationHandler.ObjectType;
+import org.keycloak.provider.Provider;
 import org.keycloak.provider.ProviderEvent;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.ACTION_TOKEN_CACHE;
@@ -112,7 +115,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
 
     @Override
     public InfinispanConnectionProvider create(KeycloakSession session) {
-        lazyInit();
+        lazyInit(session);
 
         return InfinispanUtils.isRemoteInfinispan() ?
                 new RemoteInfinispanConnectionProvider(cacheManager, remoteCacheManager, topologyInfo) :
@@ -160,15 +163,12 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
     public void close() {
         logger.debug("Closing provider");
         runWithWriteLockOnCacheManager(() -> {
-            if (cacheManager != null && !containerManaged) {
+            if (cacheManager != null) {
                 cacheManager.stop();
             }
             if (remoteCacheProvider != null) {
                 remoteCacheProvider.stop();
             }
-            if (remoteCacheManager != null && !containerManaged) {
-                remoteCacheManager.stop();
-            }
         });
     }
 
@@ -191,7 +191,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         });
     }
 
-    protected void lazyInit() {
+    protected void lazyInit(KeycloakSession keycloakSession) {
         if (cacheManager == null) {
             synchronized (this) {
                 if (cacheManager == null) {
@@ -207,7 +207,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
                             throw new RuntimeException("Multiple " + org.keycloak.cluster.ManagedCacheManagerProvider.class + " providers found.");
                         }
 
-                        managedCacheManager = provider.getEmbeddedCacheManager(config);
+                        managedCacheManager = provider.getEmbeddedCacheManager(keycloakSession, config);
                         if (InfinispanUtils.isRemoteInfinispan()) {
                             rcm = provider.getRemoteCacheManager(config);
                         }
@@ -489,4 +489,9 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
             }
         });
     }
+
+    @Override
+    public Set<Class<? extends Provider>> dependsOn() {
+        return Set.of(JpaConnectionProvider.class);
+    }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/InfinispanSingleUseObjectProviderFactory.java

@@ -22,6 +22,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
@@ -43,6 +44,7 @@ import org.keycloak.models.session.RevokedTokenPersisterProvider;
 import org.keycloak.models.sessions.infinispan.entities.SingleUseObjectValueEntity;
 import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.provider.EnvironmentDependentProviderFactory;
+import org.keycloak.provider.Provider;
 import org.keycloak.provider.ProviderConfigProperty;
 import org.keycloak.provider.ProviderConfigurationBuilder;
 import org.keycloak.provider.ServerInfoAwareProviderFactory;
@@ -65,6 +67,11 @@ public class InfinispanSingleUseObjectProviderFactory implements SingleUseObject
     private volatile boolean initialized;
     private boolean persistRevokedTokens;
 
+    @Override
+    public Set<Class<? extends Provider>> dependsOn() {
+        return Set.of(InfinispanConnectionProvider.class);
+    }
+
     @Override
     public InfinispanSingleUseObjectProvider create(KeycloakSession session) {
         initialize(session);

model/jpa/src/main/resources/META-INF/jpa-changelog-26.1.0.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!--
+  ~ * Copyright 2024 Red Hat, Inc. and/or its affiliates
+  ~ * and other contributors as indicated by the @author tags.
+  ~ *
+  ~ * Licensed 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.
+  -->
+<databaseChangeLog xmlns="http://www.liquibase.org/xml/ns/dbchangelog" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.1.xsd">
+
+    <changeSet author="keycloak" id="29399-jdbc-ping-default">
+        <createTable tableName="JGROUPS_PING">
+            <column name="address" type="VARCHAR(200)">
+                <constraints nullable="false" />
+            </column>
+            <column name="name" type="VARCHAR(200)" />
+            <column name="cluster_name" type="VARCHAR(200)">
+                <constraints nullable="false" />
+            </column>
+            <column name="ip" type="VARCHAR(200)">
+                <constraints nullable="false" />
+            </column>
+            <column name="coord" type="BOOLEAN"/>
+        </createTable>
+        <addPrimaryKey columnNames="address" constraintName="CONSTRAINT_JGROUPS_PING" tableName="JGROUPS_PING"/>
+    </changeSet>
+
+</databaseChangeLog>

model/jpa/src/main/resources/META-INF/jpa-changelog-master.xml

@@ -84,5 +84,6 @@
     <include file="META-INF/jpa-changelog-24.0.2.xml"/>
     <include file="META-INF/jpa-changelog-25.0.0.xml"/>
     <include file="META-INF/jpa-changelog-26.0.0.xml"/>
+    <include file="META-INF/jpa-changelog-26.1.0.xml"/>
 
 </databaseChangeLog>

quarkus/config-api/src/main/java/org/keycloak/config/database/Database.java

@@ -117,15 +117,15 @@ public final class Database {
                 "org.h2.jdbcx.JdbcDataSource",
                 "org.h2.Driver",
                 "org.hibernate.dialect.H2Dialect",
-                new Function<String, String>() {
+                new Function<>() {
                     @Override
                     public String apply(String alias) {
                         if ("dev-file".equalsIgnoreCase(alias)) {
-                            return addH2NonKeywords("jdbc:h2:file:${kc.home.dir:${kc.db-url-path:" + escapeReplacements(System.getProperty("user.home")) + "}}" + escapeReplacements(File.separator) + "${kc.data.dir:data}"
+                            return amendH2("jdbc:h2:file:${kc.home.dir:${kc.db-url-path:" + escapeReplacements(System.getProperty("user.home")) + "}}" + escapeReplacements(File.separator) + "${kc.data.dir:data}"
                                   + escapeReplacements(File.separator) + "h2" + escapeReplacements(File.separator)
                                   + "keycloakdb${kc.db-url-properties:}");
                         }
-                        return addH2NonKeywords("jdbc:h2:mem:keycloakdb${kc.db-url-properties:}");
+                        return amendH2("jdbc:h2:mem:keycloakdb${kc.db-url-properties:}");
                     }
 
                     private String escapeReplacements(String snippet) {
@@ -155,6 +155,26 @@ public final class Database {
                         }
                         return jdbcUrl;
                     }
+
+                    /**
+                     * Required so that the H2 db instance is closed only when the Agroal connection pool is closed during
+                     * Keycloak shutdown. We cannot rely on the default H2 ShutdownHook as this can result in the DB being
+                     * closed before dependent resources, e.g. JDBC_PING2, are shutdown gracefully. This solution also
+                     * requires the Agroal min-pool connection size to be at least 1.
+                     */
+                    private String addH2CloseOnExit(String jdbcUrl) {
+                        if (!jdbcUrl.contains("DB_CLOSE_ON_EXIT=")) {
+                            jdbcUrl = jdbcUrl + ";DB_CLOSE_ON_EXIT=FALSE";
+                        }
+                        if (!jdbcUrl.contains("DB_CLOSE_DELAY=")) {
+                            jdbcUrl = jdbcUrl + ";DB_CLOSE_DELAY=0";
+                        }
+                        return jdbcUrl;
+                    }
+
+                    private String amendH2(String jdbcUrl) {
+                        return addH2CloseOnExit(addH2NonKeywords(jdbcUrl));
+                    }
                 },
                 asList("liquibase.database.core.H2Database"),
                 "dev-mem", "dev-file"

quarkus/deployment/src/main/java/org/keycloak/quarkus/deployment/CacheBuildSteps.java

@@ -23,7 +23,6 @@ import io.quarkus.deployment.annotations.BuildStep;
 import io.quarkus.deployment.annotations.Consume;
 import io.quarkus.deployment.annotations.ExecutionTime;
 import io.quarkus.deployment.annotations.Record;
-import io.quarkus.deployment.builditem.ShutdownContextBuildItem;
 import io.quarkus.deployment.logging.LoggingSetupBuildItem;
 import jakarta.enterprise.context.ApplicationScoped;
 import org.keycloak.quarkus.runtime.KeycloakRecorder;
@@ -40,11 +39,11 @@ public class CacheBuildSteps {
     @Consume(LoggingSetupBuildItem.class)
     @Record(ExecutionTime.RUNTIME_INIT)
     @BuildStep
-    void configureInfinispan(KeycloakRecorder recorder, BuildProducer<SyntheticBeanBuildItem> syntheticBeanBuildItems, ShutdownContextBuildItem shutdownContext) {
+    void configureInfinispan(KeycloakRecorder recorder, BuildProducer<SyntheticBeanBuildItem> syntheticBeanBuildItems) {
         syntheticBeanBuildItems.produce(SyntheticBeanBuildItem.configure(CacheManagerFactory.class)
                 .scope(ApplicationScoped.class)
                 .unremovable()
                 .setRuntimeInit()
-                .runtimeValue(recorder.createCacheInitializer(shutdownContext)).done());
+                .runtimeValue(recorder.createCacheInitializer()).done());
     }
 }

quarkus/deployment/src/main/java/org/keycloak/quarkus/deployment/KeycloakProcessor.java

@@ -485,7 +485,7 @@ class KeycloakProcessor {
             }
         }
 
-        recorder.configSessionFactory(factories, defaultProviders, preConfiguredProviders, loadThemesFromClassPath(), Environment.isRebuild());
+        recorder.configSessionFactory(factories, defaultProviders, preConfiguredProviders, loadThemesFromClassPath());
     }
 
     private List<ClasspathThemeProviderFactory.ThemesRepresentation> loadThemesFromClassPath() {

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/KeycloakRecorder.java

@@ -121,14 +121,13 @@ public class KeycloakRecorder {
             Map<Spi, Map<Class<? extends Provider>, Map<String, Class<? extends ProviderFactory>>>> factories,
             Map<Class<? extends Provider>, String> defaultProviders,
             Map<String, ProviderFactory> preConfiguredProviders,
-            List<ClasspathThemeProviderFactory.ThemesRepresentation> themes, boolean reaugmented) {
+            List<ClasspathThemeProviderFactory.ThemesRepresentation> themes) {
-        QuarkusKeycloakSessionFactory.setInstance(new QuarkusKeycloakSessionFactory(factories, defaultProviders, preConfiguredProviders, themes, reaugmented));
+        QuarkusKeycloakSessionFactory.setInstance(new QuarkusKeycloakSessionFactory(factories, defaultProviders, preConfiguredProviders, themes));
     }
 
-    public RuntimeValue<CacheManagerFactory> createCacheInitializer(ShutdownContext shutdownContext) {
+    public RuntimeValue<CacheManagerFactory> createCacheInitializer() {
         try {
             CacheManagerFactory cacheManagerFactory = new CacheManagerFactory(getInfinispanConfigFile());
-            shutdownContext.addShutdownTask(cacheManagerFactory::shutdown);
             return new RuntimeValue<>(cacheManagerFactory);
         } catch (Exception e) {
             throw new RuntimeException(e);

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/configuration/mappers/DatabasePropertyMappers.java

@@ -69,6 +69,7 @@ final class DatabasePropertyMappers {
                         .build(),
                 fromOption(DatabaseOptions.DB_POOL_MIN_SIZE)
                         .to("quarkus.datasource.jdbc.min-size")
+                        .transformer(DatabasePropertyMappers::transformMinPoolSize)
                         .paramLabel("size")
                         .build(),
                 fromOption(DatabaseOptions.DB_POOL_MAX_SIZE)
@@ -118,4 +119,11 @@ final class DatabasePropertyMappers {
         return Database.getDialect(db).orElse(null);
     }
 
+    /**
+     * For H2 databases we must ensure that the min-pool size is at least one so that the DB is not shutdown until the
+     * Agroal connection pool is closed on Keycloak shutdown.
+     */
+    private static String transformMinPoolSize(String min, ConfigSourceInterceptorContext context) {
+        return isDevModeDatabase(context) && (min == null || "0".equals(min)) ? "1" : min;
+    }
 }

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/integration/QuarkusKeycloakSessionFactory.java

@@ -17,7 +17,6 @@
 
 package org.keycloak.quarkus.runtime.integration;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -48,20 +47,13 @@ public final class QuarkusKeycloakSessionFactory extends DefaultKeycloakSessionF
     }
 
     private static QuarkusKeycloakSessionFactory INSTANCE;
-    private final Boolean reaugmented;
-    private final Map<Spi, Map<Class<? extends Provider>, Map<String, Class<? extends ProviderFactory>>>> factories;
-    private Map<String, ProviderFactory> preConfiguredProviders;
 
     public QuarkusKeycloakSessionFactory(
             Map<Spi, Map<Class<? extends Provider>, Map<String, Class<? extends ProviderFactory>>>> factories,
             Map<Class<? extends Provider>, String> defaultProviders,
             Map<String, ProviderFactory> preConfiguredProviders,
-            List<ClasspathThemeProviderFactory.ThemesRepresentation> themes,
+            List<ClasspathThemeProviderFactory.ThemesRepresentation> themes) {
-            Boolean reaugmented) {
         this.provider = defaultProviders;
-        this.factories = factories;
-        this.preConfiguredProviders = preConfiguredProviders;
-        this.reaugmented = reaugmented;
         serverStartupTimestamp = System.currentTimeMillis();
         spis = factories.keySet();
 
@@ -88,25 +80,11 @@ public final class QuarkusKeycloakSessionFactory extends DefaultKeycloakSessionF
     }
 
     private QuarkusKeycloakSessionFactory() {
-        reaugmented = false;
-        factories = Collections.emptyMap();
     }
 
     @Override
     public void init() {
-        // Component factory must be initialized first, so that postInit in other factories can use component factories
+        initProviderFactories();
-        updateComponentFactoryProviderFactory();
-        if (componentFactoryPF != null) {
-            componentFactoryPF.postInit(this);
-        }
-        for (Map<String, ProviderFactory> f : factoriesMap.values()) {
-            for (ProviderFactory factory : f.values()) {
-                if (factory != componentFactoryPF) {
-                    factory.postInit(this);
-                }
-            }
-        }
-
         AdminPermissions.registerListener(this);
         // make the session factory ready for hot deployment
         ProviderManagerRegistry.SINGLETON.setDeployer(this);

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/infinispan/CacheManagerFactory.java

@@ -20,20 +20,26 @@ package org.keycloak.quarkus.runtime.storage.infinispan;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import io.agroal.api.AgroalDataSource;
 import io.micrometer.core.instrument.Metrics;
+import io.quarkus.arc.Arc;
+import jakarta.persistence.EntityManager;
+
 import org.infinispan.client.hotrod.RemoteCache;
 import org.infinispan.client.hotrod.RemoteCacheManager;
 import org.infinispan.client.hotrod.RemoteCacheManagerAdmin;
 import org.infinispan.client.hotrod.impl.ConfigurationProperties;
-import org.infinispan.commons.api.Lifecycle;
 import org.infinispan.commons.dataconversion.MediaType;
 import org.infinispan.commons.internal.InternalCacheNames;
 import org.infinispan.commons.util.concurrent.CompletableFutures;
@@ -42,6 +48,7 @@ import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.configuration.cache.HashConfiguration;
 import org.infinispan.configuration.cache.PersistenceConfigurationBuilder;
 import org.infinispan.configuration.global.GlobalConfiguration;
+import org.infinispan.configuration.global.ShutdownHookBehavior;
 import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
 import org.infinispan.configuration.parsing.ParserRegistry;
 import org.infinispan.manager.DefaultCacheManager;
@@ -50,8 +57,11 @@ import org.infinispan.persistence.remote.configuration.ExhaustedAction;
 import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
 import org.infinispan.protostream.descriptors.FileDescriptor;
 import org.infinispan.query.remote.client.ProtobufMetadataManagerConstants;
+import org.infinispan.remoting.transport.jgroups.EmbeddedJGroupsChannelConfigurator;
 import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
 import org.jboss.logging.Logger;
+import org.jgroups.conf.ProtocolConfiguration;
+import org.jgroups.protocols.JDBC_PING2;
 import org.jgroups.protocols.TCP_NIO2;
 import org.jgroups.protocols.UDP;
 import org.jgroups.util.TLS;
@@ -60,10 +70,13 @@ import org.keycloak.common.Profile;
 import org.keycloak.common.util.MultiSiteUtils;
 import org.keycloak.config.CachingOptions;
 import org.keycloak.config.MetricsOptions;
+import org.keycloak.connections.jpa.JpaConnectionProvider;
+import org.keycloak.connections.jpa.util.JpaUtils;
 import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.marshalling.KeycloakIndexSchemaUtil;
 import org.keycloak.marshalling.KeycloakModelSchema;
 import org.keycloak.marshalling.Marshalling;
+import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.sessions.infinispan.query.ClientSessionQueries;
 import org.keycloak.models.sessions.infinispan.query.UserSessionQueries;
 import org.keycloak.models.sessions.infinispan.remote.RemoteInfinispanAuthenticationSessionProviderFactory;
@@ -71,7 +84,9 @@ import org.keycloak.models.sessions.infinispan.remote.RemoteUserLoginFailureProv
 import org.keycloak.quarkus.runtime.configuration.Configuration;
 
 import javax.net.ssl.SSLContext;
+import javax.sql.DataSource;
 
+import static org.infinispan.configuration.global.TransportConfiguration.STACK;
 import static org.keycloak.config.CachingOptions.CACHE_EMBEDDED_MTLS_KEYSTORE_FILE_PROPERTY;
 import static org.keycloak.config.CachingOptions.CACHE_EMBEDDED_MTLS_KEYSTORE_PASSWORD_PROPERTY;
 import static org.keycloak.config.CachingOptions.CACHE_EMBEDDED_MTLS_TRUSTSTORE_FILE_PROPERTY;
@@ -95,11 +110,12 @@ public class CacheManagerFactory {
 
     private static final Logger logger = Logger.getLogger(CacheManagerFactory.class);
 
-    private final CompletableFuture<DefaultCacheManager> cacheManagerFuture;
     private final CompletableFuture<RemoteCacheManager> remoteCacheManagerFuture;
+    private final String config;
+    private volatile DefaultCacheManager cacheManager;
 
     public CacheManagerFactory(String config) {
-        this.cacheManagerFuture = startEmbeddedCacheManager(config);
+        this.config = config;
         if (InfinispanUtils.isRemoteInfinispan()) {
             logger.debug("Remote Cache feature is enabled");
             this.remoteCacheManagerFuture = CompletableFuture.supplyAsync(this::startRemoteCacheManager);
@@ -109,20 +125,20 @@ public class CacheManagerFactory {
         }
     }
 
-    public DefaultCacheManager getOrCreateEmbeddedCacheManager() {
+    public DefaultCacheManager getOrCreateEmbeddedCacheManager(KeycloakSession keycloakSession) {
-        return join(cacheManagerFuture);
+        if (cacheManager == null) {
+           synchronized (this) {
+              if (cacheManager == null)
+                 cacheManager = startEmbeddedCacheManager(keycloakSession);
+           }
+        }
+        return cacheManager;
     }
 
     public RemoteCacheManager getOrCreateRemoteCacheManager() {
         return join(remoteCacheManagerFuture);
     }
 
-    public void shutdown() {
-        logger.debug("Shutdown embedded and remote cache managers");
-        cacheManagerFuture.thenAccept(CacheManagerFactory::close);
-        remoteCacheManagerFuture.thenAccept(CacheManagerFactory::close);
-    }
-
     private static <T> T join(Future<T> future) {
         try {
             return future.get(getStartTimeout(), TimeUnit.SECONDS);
@@ -134,12 +150,6 @@ public class CacheManagerFactory {
         }
     }
 
-    private static void close(Lifecycle lifecycle) {
-        if (lifecycle != null) {
-            lifecycle.stop();
-        }
-    }
-
     private RemoteCacheManager startRemoteCacheManager() {
         logger.info("Starting Infinispan remote cache manager (Hot Rod Client)");
         String cacheRemoteHost = requiredStringProperty(CACHE_REMOTE_HOST_PROPERTY);
@@ -278,10 +288,14 @@ public class CacheManagerFactory {
         admin.reindexCache(cacheName);
     }
 
-    private CompletableFuture<DefaultCacheManager> startEmbeddedCacheManager(String config) {
+    private DefaultCacheManager startEmbeddedCacheManager(KeycloakSession keycloakSession) {
         logger.info("Starting Infinispan embedded cache manager");
         ConfigurationBuilderHolder builder = new ParserRegistry().parse(config);
 
+        // We must disable the Infinispan default ShutdownHook as we manage the EmbeddedCacheManager lifecycle explicitly
+        // with #shutdown and multiple calls to EmbeddedCacheManager#stop can lead to Exceptions being thrown
+        builder.getGlobalConfigurationBuilder().shutdown().hookBehavior(ShutdownHookBehavior.DONT_REGISTER);
+
         if (Configuration.isTrue(MetricsOptions.METRICS_ENABLED)) {
             builder.getGlobalConfigurationBuilder().addModule(MicrometerMeterRegisterConfigurationBuilder.class);
             builder.getGlobalConfigurationBuilder().module(MicrometerMeterRegisterConfigurationBuilder.class).meterRegistry(Metrics.globalRegistry);
@@ -310,7 +324,7 @@ public class CacheManagerFactory {
         } else {
             // embedded mode!
             if (builder.getNamedConfigurationBuilders().entrySet().stream().anyMatch(c -> c.getValue().clustering().cacheMode().isClustered())) {
-                configureTransportStack(builder);
+                configureTransportStack(builder, keycloakSession);
                 configureRemoteStores(builder);
             }
             configureCacheMaxCount(builder, CachingOptions.CLUSTERED_MAX_COUNT_CACHES);
@@ -320,8 +334,7 @@ public class CacheManagerFactory {
         configureCacheMaxCount(builder, CachingOptions.LOCAL_MAX_COUNT_CACHES);
         checkForRemoteStores(builder);
 
-        var start = isStartEagerly();
+        return new DefaultCacheManager(builder, isStartEagerly());
-        return CompletableFuture.supplyAsync(() -> new DefaultCacheManager(builder, start));
     }
 
     private static boolean isRemoteTLSEnabled() {
@@ -357,12 +370,37 @@ public class CacheManagerFactory {
         return Integer.getInteger("kc.cache-ispn-start-timeout", 120);
     }
 
-    private static void configureTransportStack(ConfigurationBuilderHolder builder) {
+    private void configureTransportStack(ConfigurationBuilderHolder builder, KeycloakSession keycloakSession) {
         String transportStack = Configuration.getRawValue("kc.cache-stack");
 
+        var jdbcStackName = "jdbc-ping";
         var transportConfig = builder.getGlobalConfigurationBuilder().transport();
-        if (transportStack != null && !transportStack.isBlank()) {
+        var stackXmlAttribute = transportConfig.defaultTransport().attributes().attribute(STACK);
+        if (transportStack != null && !transportStack.isBlank() && !jdbcStackName.equals(transportStack)) {
             transportConfig.defaultTransport().stack(transportStack);
+        } else if (!stackXmlAttribute.isModified() || jdbcStackName.equals(stackXmlAttribute.get())){
+            EntityManager em = keycloakSession.getProvider(JpaConnectionProvider.class).getEntityManager();
+            var tableName = JpaUtils.getTableNameForNativeQuery("JGROUPS_PING", em);
+            var attributes = Map.of(
+                  // Leave initialize_sql blank as table is already created by Keycloak
+                  "initialize_sql", "",
+                  // Explicitly specify clear and select_all SQL to ensure "cluster_name" column is used, as the default
+                  // "cluster" cannot be used with Oracle DB as it's a reserved word.
+                  "clear_sql", String.format("DELETE from %s WHERE cluster_name=?", tableName),
+                  "delete_single_sql", String.format("DELETE from %s WHERE address=?", tableName),
+                  "insert_single_sql", String.format("INSERT INTO %s values (?, ?, ?, ?, ?)", tableName),
+                  "select_all_pingdata_sql", String.format("SELECT address, name, ip, coord FROM %s WHERE cluster_name=?", tableName),
+                  "remove_all_data_on_view_change", "true",
+                  "register_shutdown_hook", "false",
+                  "stack.combine", "REPLACE",
+                  "stack.position", "PING"
+            );
+            var stack = List.of(new ProtocolConfiguration(JDBC_PING2.class.getSimpleName(), attributes));
+            builder.addJGroupsStack(new EmbeddedJGroupsChannelConfigurator(jdbcStackName, stack, null), "udp");
+
+            Supplier<DataSource> dataSourceSupplier = Arc.container().select(AgroalDataSource.class)::get;
+            transportConfig.addProperty(JGroupsTransport.DATA_SOURCE, dataSourceSupplier);
+            transportConfig.defaultTransport().stack(jdbcStackName);
         }
 
         if (Configuration.isTrue(CachingOptions.CACHE_EMBEDDED_MTLS_ENABLED_PROPERTY)) {
@@ -378,7 +416,7 @@ public class CacheManagerFactory {
                     .setClientAuth(TLSClientAuth.NEED)
                     .setProtocols(new String[]{"TLSv1.3"});
             transportConfig.addProperty(JGroupsTransport.SOCKET_FACTORY, tls.createSocketFactory());
-            Logger.getLogger(CacheManagerFactory.class).info("MTLS enabled for communications for embedded caches");
+            logger.info("MTLS enabled for communications for embedded caches");
         }
     }
 

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/infinispan/QuarkusCacheManagerProvider.java

@@ -20,6 +20,7 @@ package org.keycloak.quarkus.runtime.storage.infinispan;
 import io.quarkus.arc.Arc;
 import org.keycloak.Config;
 import org.keycloak.cluster.ManagedCacheManagerProvider;
+import org.keycloak.models.KeycloakSession;
 
 /**
  * @author <a href="mailto:psilva@redhat.com">Pedro Igor</a>
@@ -27,8 +28,8 @@ import org.keycloak.cluster.ManagedCacheManagerProvider;
 public final class QuarkusCacheManagerProvider implements ManagedCacheManagerProvider {
 
     @Override
-    public <C> C getEmbeddedCacheManager(Config.Scope config) {
+    public <C> C getEmbeddedCacheManager(KeycloakSession keycloakSession, Config.Scope config) {
-        return (C) Arc.container().instance(CacheManagerFactory.class).get().getOrCreateEmbeddedCacheManager();
+        return (C) Arc.container().instance(CacheManagerFactory.class).get().getOrCreateEmbeddedCacheManager(keycloakSession);
     }
 
     @Override

quarkus/runtime/src/main/resources/cache-ispn.xml

@@ -22,7 +22,7 @@
         xmlns="urn:infinispan:config:15.0">
 
     <cache-container name="keycloak">
-        <transport lock-timeout="60000" stack="udp"/>
+        <transport lock-timeout="60000"/>
         <local-cache name="realms" simple-cache="true">
             <encoding>
                 <key media-type="application/x-java-object"/>

quarkus/runtime/src/test/java/org/keycloak/quarkus/runtime/configuration/test/ConfigurationTest.java

@@ -221,12 +221,12 @@ public class ConfigurationTest extends AbstractConfigurationTest {
                 .toString()
                 .replaceFirst(isWindows() ? "file:///" : "file://", "");
 
-        assertEquals("jdbc:h2:file:" + userHomeUri + "data/h2/keycloakdb;NON_KEYWORDS=VALUE", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
+        assertEquals("jdbc:h2:file:" + userHomeUri + "data/h2/keycloakdb;NON_KEYWORDS=VALUE;DB_CLOSE_ON_EXIT=FALSE;DB_CLOSE_DELAY=0", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
 
         ConfigArgsConfigSource.setCliArgs("--db=dev-mem");
         config = createConfig();
         assertEquals(H2Dialect.class.getName(), config.getConfigValue("kc.db-dialect").getValue());
-        assertEquals("jdbc:h2:mem:keycloakdb;NON_KEYWORDS=VALUE", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
+        assertEquals("jdbc:h2:mem:keycloakdb;NON_KEYWORDS=VALUE;DB_CLOSE_ON_EXIT=FALSE;DB_CLOSE_DELAY=0", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
         assertEquals("h2", config.getConfigValue("quarkus.datasource.db-kind").getValue());
 
         ConfigArgsConfigSource.setCliArgs("--db=dev-mem", "--db-username=other");
@@ -304,13 +304,13 @@ public class ConfigurationTest extends AbstractConfigurationTest {
         ConfigArgsConfigSource.setCliArgs("--db=dev-file");
         SmallRyeConfig config = createConfig();
         assertEquals(H2Dialect.class.getName(), config.getConfigValue("kc.db-dialect").getValue());
-        assertEquals("jdbc:h2:file:test-dir/data/h2/keycloakdb;;test=test;test1=test1;NON_KEYWORDS=VALUE", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
+        assertEquals("jdbc:h2:file:test-dir/data/h2/keycloakdb;;test=test;test1=test1;NON_KEYWORDS=VALUE;DB_CLOSE_ON_EXIT=FALSE;DB_CLOSE_DELAY=0", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
         assertEquals("xa", config.getConfigValue("quarkus.datasource.jdbc.transactions").getValue());
 
         ConfigArgsConfigSource.setCliArgs("");
         config = createConfig();
         assertEquals(H2Dialect.class.getName(), config.getConfigValue("kc.db-dialect").getValue());
-        assertEquals("jdbc:h2:file:test-dir/data/h2/keycloakdb;;test=test;test1=test1;NON_KEYWORDS=VALUE", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
+        assertEquals("jdbc:h2:file:test-dir/data/h2/keycloakdb;;test=test;test1=test1;NON_KEYWORDS=VALUE;DB_CLOSE_ON_EXIT=FALSE;DB_CLOSE_DELAY=0", config.getConfigValue("quarkus.datasource.jdbc.url").getValue());
 
         System.setProperty("kc.db-url-properties", "?test=test&test1=test1");
         ConfigArgsConfigSource.setCliArgs("--db=mariadb");

server-spi-private/src/main/java/org/keycloak/cluster/ManagedCacheManagerProvider.java

@@ -18,6 +18,7 @@
 package org.keycloak.cluster;
 
 import org.keycloak.Config;
+import org.keycloak.models.KeycloakSession;
 
 /**
  * A Service Provider Interface (SPI) that allows to plug-in an embedded or remote cache manager instance.
@@ -26,7 +27,7 @@ import org.keycloak.Config;
  */
 public interface ManagedCacheManagerProvider {
 
-    <C> C getEmbeddedCacheManager(Config.Scope config);
+    <C> C getEmbeddedCacheManager(KeycloakSession keycloakSession, Config.Scope config);
 
     /**
      * @return A RemoteCacheManager if the features {@link org.keycloak.common.Profile.Feature#CLUSTERLESS} or {@link org.keycloak.common.Profile.Feature#MULTI_SITE}  is enabled, {@code null} otherwise.

server-spi/src/main/java/org/keycloak/provider/ProviderFactory.java

@@ -19,6 +19,8 @@ package org.keycloak.provider;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Set;
+
 import org.keycloak.Config;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
@@ -67,4 +69,15 @@ public interface ProviderFactory<T extends Provider> {
     default List<ProviderConfigProperty> getConfigMetadata() {
         return Collections.emptyList();
     }
+
+    /**
+     * Optional method used to declare that a ProviderFactory has a dependency on one or more Providers. If a Provider
+     * is declared here, it is guaranteed that the dependencies {@link #postInit} method will be executed
+     * before this ProviderFactory's {@link #postInit}. Similarly, it's guaranteed that {@link #close()} will be
+     * called on this {@link ProviderFactory} before {@link #close()} is called on any of the dependent ProviderFactory
+     * implementations.
+     */
+    default Set<Class<? extends Provider>> dependsOn() {
+        return Collections.emptySet();
+    }
 }

services/src/main/java/org/keycloak/services/DefaultKeycloakSessionFactory.java

@@ -16,6 +16,7 @@
  */
 package org.keycloak.services;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -24,8 +25,10 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.Stack;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
 import java.util.stream.Stream;
@@ -66,13 +69,6 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
     // TODO: Likely should be changed to int and use Time.currentTime() to be compatible with all our "time" reps
     protected long serverStartupTimestamp;
 
-    /**
-     * Timeouts are used as time boundary for obtaining models from an external storage. Default value is set
-     * to 3000 milliseconds and it's configurable.
-     */
-    private Long clientStorageProviderTimeout;
-    private Long roleStorageProviderTimeout;
-
     protected ComponentFactoryProviderFactory componentFactoryPF;
 
     @Override
@@ -117,18 +113,7 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
                 }
             }
             checkProvider();
-            // Component factory must be initialized first, so that postInit in other factories can use component factories
+            initProviderFactories();
-            updateComponentFactoryProviderFactory();
-            if (componentFactoryPF != null) {
-                componentFactoryPF.postInit(this);
-            }
-            for (Map<String, ProviderFactory> factories : factoriesMap.values()) {
-                for (ProviderFactory factory : factories.values()) {
-                    if (factory != componentFactoryPF) {
-                        factory.postInit(this);
-                    }
-                }
-            }
             // make the session factory ready for hot deployment
             ProviderManagerRegistry.SINGLETON.setDeployer(this);
         }
@@ -136,6 +121,54 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
         AdminPermissions.registerListener(this);
     }
 
+    protected void initProviderFactories() {
+        initProviderFactories(true, factoriesMap);
+    }
+
+    protected void initProviderFactories(boolean updateComponentFactory, Map<Class<? extends Provider>, Map<String, ProviderFactory>> factories) {
+        if (updateComponentFactory) {
+            // Component factory must be initialized first, so that postInit in other factories can use component factories
+            updateComponentFactoryProviderFactory();
+            if (componentFactoryPF != null) {
+                componentFactoryPF.postInit(this);
+            }
+        }
+
+        Set<Class<? extends Provider>> initializedProviders = new HashSet<>();
+        Stack<ProviderFactory> recursionPrevention = new Stack<>();
+
+        for(Map.Entry<Class<? extends Provider>, Map<String, ProviderFactory>>  f : factories.entrySet()) {
+            if (initializedProviders.contains(f.getKey())) {
+                continue;
+            }
+            initializeProviders(f.getKey(), factories, initializedProviders, recursionPrevention);
+        }
+    }
+
+    private void initializeProviders(Class<? extends Provider> provider, Map<Class<? extends Provider>, Map<String, ProviderFactory>> factories, Set<Class<? extends Provider>> intializedProviders, Stack<ProviderFactory> recursionPrevention) {
+        for (ProviderFactory<?> factory : factories.get(provider).values()) {
+            if (factory == componentFactoryPF)
+                continue;
+
+            for (Class<? extends Provider> providerDep : factory.dependsOn()) {
+                if (recursionPrevention.contains(factory)) {
+                    List<String> stackForException = recursionPrevention.stream().map(providerFactory -> providerFactory.getClass().getName()).toList();
+                    throw new RuntimeException("Detected a recursive dependency on provider " + providerDep.getName() +
+                          " while the initialization of the following provider factories is ongoing: " + stackForException);
+                }
+                Map<String, ProviderFactory> f = factories.get(providerDep);
+                if (f == null) {
+                    throw new RuntimeException("No provider factories exists for provider " + providerDep.getSimpleName());
+                }
+                recursionPrevention.push(factory);
+                initializeProviders(providerDep, factories, intializedProviders, recursionPrevention);
+                recursionPrevention.pop();
+            }
+            factory.postInit(this);
+            intializedProviders.add(provider);
+        }
+    }
+
     protected Map<Class<? extends Provider>, Map<String, ProviderFactory>> getFactoriesCopy() {
         Map<Class<? extends Provider>, Map<String, ProviderFactory>> copy = new HashMap<>();
         for (Map.Entry<Class<? extends Provider>, Map<String, ProviderFactory>> entry : factoriesMap.entrySet()) {
@@ -150,17 +183,22 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
     public void deploy(ProviderManager pm) {
         Map<Class<? extends Provider>, Map<String, ProviderFactory>> copy = getFactoriesCopy();
         Map<Class<? extends Provider>, Map<String, ProviderFactory>> newFactories = loadFactories(pm);
-        List<ProviderFactory> deployed = new LinkedList<>();
+        Map<Class<? extends Provider>, Map<String, ProviderFactory>> deployed = new HashMap<>();
         List<ProviderFactory> undeployed = new LinkedList<>();
 
         for (Map.Entry<Class<? extends Provider>, Map<String, ProviderFactory>> entry : newFactories.entrySet()) {
-            Map<String, ProviderFactory> current = copy.get(entry.getKey());
+            Class<? extends Provider> provider = entry.getKey();
+            Map<String, ProviderFactory> current = copy.get(provider);
             if (current == null) {
-                copy.put(entry.getKey(), entry.getValue());
+                copy.put(provider, entry.getValue());
             } else {
-                for (ProviderFactory f : entry.getValue().values()) {
+                for (Map.Entry<String, ProviderFactory> e : entry.getValue().entrySet()) {
-                    deployed.add(f);
+                    deployed.compute(provider, (k, v) -> {
-                    ProviderFactory old = current.remove(f.getId());
+                        Map<String, ProviderFactory> map = Objects.requireNonNullElseGet(v, HashMap::new);
+                        map.put(e.getKey(), e.getValue());
+                        return map;
+                    });
+                    ProviderFactory old = current.remove(e.getValue().getId());
                     if (old != null) {
                         undeployed.add(old);
                     }
@@ -178,18 +216,7 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
             factory.close();
             cfChanged |= (componentFactoryPF == factory);
         }
-        // Component factory must be initialized first, so that postInit in other factories can use component factories
+        initProviderFactories(cfChanged, deployed);
-        if (cfChanged) {
-            updateComponentFactoryProviderFactory();
-            if (componentFactoryPF != null) {
-                componentFactoryPF.postInit(this);
-            }
-        }
-        for (ProviderFactory factory : deployed) {
-            if (factory != componentFactoryPF) {
-                factory.postInit(this);
-            }
-        }
 
         if (pm.getInfo().hasThemes() || pm.getInfo().hasThemeResources()) {
             themeManagerFactory.clearCache();
@@ -415,11 +442,52 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
     @Override
     public void close() {
         ProviderManagerRegistry.SINGLETON.setDeployer(null);
-        for (Map<String, ProviderFactory> factories : factoriesMap.values()) {
+
-            for (ProviderFactory factory : factories.values()) {
+        // Create a tree-structure to represent reverse relation of ProviderFactory#dependsOn to Providers
-                factory.close();
+        Map<Class<? extends Provider>, Node<Set<ProviderFactory>>> nodes = new HashMap<>();
+        for (Map.Entry<Class<? extends Provider>, Map<String, ProviderFactory>>  f : factoriesMap.entrySet()) {
+            Class<? extends Provider> provider = f.getKey();
+            for (Map.Entry<String, ProviderFactory> entry : f.getValue().entrySet()) {
+                ProviderFactory pf = entry.getValue();
+                Node<Set<ProviderFactory>> node = nodes.computeIfAbsent(provider, k -> new Node<>(new HashSet<>()));
+                // Add ProviderFactory to the associated Provider node
+                node.data.add(pf);
+                // If dependencies exist, make this node a child of the Provider dependencies node so that we can ensure
+                // that the leaves of the tree are closed first
+                pf.dependsOn().forEach(dep -> {
+                    node.parent = nodes.computeIfAbsent((Class<? extends Provider>) dep, k -> new Node<>(new HashSet<>()));
+                    node.parent.children.add(node);
+                });
             }
         }
+        nodes.values().forEach(this::closeProvider);
+    }
+
+    private void closeProvider(Node<Set<ProviderFactory>> node) {
+        for (var it = node.children.iterator(); it.hasNext(); ) {
+            closeProvider(it.next());
+            it.remove();
+        }
+
+        // Provider has no other dependent ProviderFactories, it's ProviderFactories can safely be closed
+        for (var it = node.data.iterator(); it.hasNext(); ) {
+            ProviderFactory pf = it.next();
+            logger.debugf("Closing ProviderFactory: %s", pf.getClass().getName());
+            pf.close();
+            it.remove();
+        }
+    }
+
+    private static class Node<T> {
+        private final T data;
+        private Node<T> parent;
+        private List<Node<T>> children;
+
+        public Node(T data) {
+            this.data = data;
+            this.parent = null;
+            this.children = new ArrayList<>();
+        }
     }
 
     public static boolean isInternal(ProviderFactory<?> factory) {
@@ -427,20 +495,6 @@ public abstract class DefaultKeycloakSessionFactory implements KeycloakSessionFa
         return packageName.startsWith("org.keycloak") && !packageName.startsWith("org.keycloak.examples");
     }
 
-    public long getClientStorageProviderTimeout() {
-        if (clientStorageProviderTimeout == null) {
-            clientStorageProviderTimeout = Config.scope("client").getLong("storageProviderTimeout", 3000L);
-        }
-        return clientStorageProviderTimeout;
-    }
-
-    public long getRoleStorageProviderTimeout() {
-        if (roleStorageProviderTimeout == null) {
-            roleStorageProviderTimeout = Config.scope("role").getLong("storageProviderTimeout", 3000L);
-        }
-        return roleStorageProviderTimeout;
-    }
-
     /**
      * @return timestamp of Keycloak server startup
      */

testsuite/integration-arquillian/tests/base/src/main/java/org/keycloak/testsuite/arquillian/containers/AbstractQuarkusDeployableContainer.java

@@ -195,6 +195,10 @@ public abstract class AbstractQuarkusDeployableContainer implements DeployableCo
         } else {
             commands.add("--cache=ispn");
             commands.add("--cache-config-file=cluster-" + cacheMode + ".xml");
+
+            var stack = System.getProperty("auth.server.quarkus.cluster.stack");
+            if (stack != null)
+                commands.add("--cache-stack=" + stack);
         }
 
         log.debugf("FIPS Mode: %s", configuration.getFipsMode());