diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index c8149dd031..a35088fffa 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -309,6 +309,10 @@ jobs:
     if: needs.conditional.outputs.ci-store == 'true'
     runs-on: ubuntu-latest
     timeout-minutes: 150
+    strategy:
+      matrix:
+        variant: ["persistent-user-sessions,persistent-user-sessions-no-cache", "persistent-user-sessions"]
+      fail-fast: false
     steps:
       - uses: actions/checkout@v4
 
@@ -316,11 +320,11 @@ jobs:
         name: Integration test setup
         uses: ./.github/actions/integration-test-setup
 
-      - name: Run base tests
+      - name: Run base tests without cache
         run: |
           TESTS=`testsuite/integration-arquillian/tests/base/testsuites/suite.sh persistent-sessions`
           echo "Tests: $TESTS"
-          ./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Dauth.server.features=persistent-user-sessions,persistent-user-sessions-no-cache -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
+          ./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Dauth.server.features=${{ matrix.variant }} -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
 
       - name: Upload JVM Heapdumps
         if: always()
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java
index fcbdd5579a..4df31db468 100755
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java
@@ -114,9 +114,9 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
     protected final RemoteCacheInvoker remoteCacheInvoker;
     protected final InfinispanKeyGenerator keyGenerator;
 
-    protected final SessionFunction offlineSessionCacheEntryLifespanAdjuster;
+    protected final SessionFunction<UserSessionEntity> offlineSessionCacheEntryLifespanAdjuster;
 
-    protected final SessionFunction offlineClientSessionCacheEntryLifespanAdjuster;
+    protected final SessionFunction<AuthenticatedClientSessionEntity> offlineClientSessionCacheEntryLifespanAdjuster;
 
     public PersistentUserSessionProvider(KeycloakSession session,
                                          RemoteCacheInvoker remoteCacheInvoker,
@@ -206,12 +206,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
 
     @Override
     public AuthenticatedClientSessionModel createClientSession(RealmModel realm, ClientModel client, UserSessionModel userSession) {
-        final UUID clientSessionId;
-        if (userSession.isOffline()) {
-            clientSessionId = keyGenerator.generateKeyUUID(session, clientSessionCache);
-        } else {
-            clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSession.getId(), client.getId());
-        }
+        final UUID clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSession.getId(), client.getId());
         AuthenticatedClientSessionEntity entity = new AuthenticatedClientSessionEntity(clientSessionId);
         entity.setRealmId(realm.getId());
         entity.setClientId(client.getId());
@@ -244,7 +239,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
         SessionUpdateTask<AuthenticatedClientSessionEntity> createClientSessionTask = Tasks.addIfAbsentSync();
         clientSessionUpdateTx.addTask(clientSessionId, createClientSessionTask, entity, persistenceState);
 
-        SessionUpdateTask registerClientSessionTask = new RegisterClientSessionTask(client.getId(), clientSessionId);
+        SessionUpdateTask<UserSessionEntity> registerClientSessionTask = new RegisterClientSessionTask(client.getId(), clientSessionId);
         userSessionUpdateTx.addTask(userSession.getId(), registerClientSessionTask);
 
         return adapter;
@@ -797,99 +792,71 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
         return getUserSessionsStream(realm, client, first, max, true);
     }
 
-
     @Override
     public void importUserSessions(Collection<UserSessionModel> persistentUserSessions, boolean offline) {
         if (persistentUserSessions == null || persistentUserSessions.isEmpty()) {
             return;
         }
+        persistentUserSessions.forEach(userSessionModel -> importUserSession(userSessionModel, offline));
+    }
 
+    public SessionEntityWrapper<UserSessionEntity> importUserSession(UserSessionModel persistentUserSession, boolean offline) {
         Map<UUID, SessionEntityWrapper<AuthenticatedClientSessionEntity>> clientSessionsById = new HashMap<>();
 
-        Map<String, SessionEntityWrapper<UserSessionEntity>> sessionsById = persistentUserSessions.stream()
-                .map((UserSessionModel persistentUserSession) -> {
+        UserSessionEntity userSessionEntityToImport = createUserSessionEntityInstance(persistentUserSession);
 
-                    UserSessionEntity userSessionEntityToImport = createUserSessionEntityInstance(persistentUserSession);
+        for (Map.Entry<String, AuthenticatedClientSessionModel> entry : persistentUserSession.getAuthenticatedClientSessions().entrySet()) {
+            String clientUUID = entry.getKey();
+            AuthenticatedClientSessionModel clientSession = entry.getValue();
+            AuthenticatedClientSessionEntity clientSessionToImport = createAuthenticatedClientSessionInstance(userSessionEntityToImport.getId(), clientSession,
+                    userSessionEntityToImport.getRealmId(), clientUUID, offline);
+            clientSessionToImport.setUserSessionId(userSessionEntityToImport.getId());
 
-                    for (Map.Entry<String, AuthenticatedClientSessionModel> entry : persistentUserSession.getAuthenticatedClientSessions().entrySet()) {
-                        String clientUUID = entry.getKey();
-                        AuthenticatedClientSessionModel clientSession = entry.getValue();
-                        AuthenticatedClientSessionEntity clientSessionToImport = createAuthenticatedClientSessionInstance(userSessionEntityToImport.getId(), clientSession,
-                                userSessionEntityToImport.getRealmId(), clientUUID, offline);
-                        clientSessionToImport.setUserSessionId(userSessionEntityToImport.getId());
+            // Update timestamp to same value as userSession. LastSessionRefresh of userSession from DB will have correct value
+            clientSessionToImport.setTimestamp(userSessionEntityToImport.getLastSessionRefresh());
 
-                        // Update timestamp to same value as userSession. LastSessionRefresh of userSession from DB will have correct value
-                        clientSessionToImport.setTimestamp(userSessionEntityToImport.getLastSessionRefresh());
+            clientSessionsById.put(clientSessionToImport.getId(), new SessionEntityWrapper<>(clientSessionToImport));
 
-                        clientSessionsById.put(clientSessionToImport.getId(), new SessionEntityWrapper<>(clientSessionToImport));
+            // Update userSession entity with the clientSession
+            AuthenticatedClientSessionStore clientSessions = userSessionEntityToImport.getAuthenticatedClientSessions();
+            clientSessions.put(clientUUID, clientSessionToImport.getId());
+        }
 
-                        // Update userSession entity with the clientSession
-                        AuthenticatedClientSessionStore clientSessions = userSessionEntityToImport.getAuthenticatedClientSessions();
-                        clientSessions.put(clientUUID, clientSessionToImport.getId());
-                    }
+        SessionEntityWrapper<UserSessionEntity>  wrappedUserSessionEntity = new SessionEntityWrapper<>(userSessionEntityToImport);
 
-                    return userSessionEntityToImport;
-                })
-                .map(SessionEntityWrapper::new)
-                .collect(Collectors.toMap(sessionEntityWrapper -> sessionEntityWrapper.getEntity().getId(), Function.identity()));
+        Map<String, SessionEntityWrapper<UserSessionEntity>> sessionsById =
+                    Stream.of(wrappedUserSessionEntity).collect(Collectors.toMap(sessionEntityWrapper -> sessionEntityWrapper.getEntity().getId(), Function.identity()));
 
         // Directly put all entities to the infinispan cache
         Cache<String, SessionEntityWrapper<UserSessionEntity>> cache = CacheDecorators.skipCacheLoadersIfRemoteStoreIsEnabled(getCache(offline));
 
-        boolean importWithExpiration = sessionsById.size() == 1;
-        if (importWithExpiration) {
-            importSessionsWithExpiration(sessionsById, cache,
-                    offline ? offlineSessionCacheEntryLifespanAdjuster : SessionTimeouts::getUserSessionLifespanMs,
-                    offline ? SessionTimeouts::getOfflineSessionMaxIdleMs : SessionTimeouts::getUserSessionMaxIdleMs);
-        } else {
-            Retry.executeWithBackoff((int iteration) -> {
-                cache.putAll(sessionsById);
-            }, 10, 10);
+        sessionsById = importSessionsWithExpiration(sessionsById, cache,
+                offline ? offlineSessionCacheEntryLifespanAdjuster : SessionTimeouts::getUserSessionLifespanMs,
+                offline ? SessionTimeouts::getOfflineSessionMaxIdleMs : SessionTimeouts::getUserSessionMaxIdleMs);
+
+        if (sessionsById.isEmpty()) {
+            return null;
         }
 
         // put all entities to the remoteCache (if exists)
         RemoteCache remoteCache = InfinispanUtil.getRemoteCache(cache);
         if (remoteCache != null) {
-            Map<String, SessionEntityWrapper<UserSessionEntity>> sessionsByIdForTransport = sessionsById.values().stream()
+            Map<String, SessionEntityWrapper<UserSessionEntity>> sessionsByIdForTransport = Stream.of(wrappedUserSessionEntity)
                     .map(SessionEntityWrapper::forTransport)
                     .collect(Collectors.toMap(sessionEntityWrapper -> sessionEntityWrapper.getEntity().getId(), Function.identity()));
 
-            if (importWithExpiration) {
-                importSessionsWithExpiration(sessionsByIdForTransport, remoteCache,
-                        offline ? offlineSessionCacheEntryLifespanAdjuster : SessionTimeouts::getUserSessionLifespanMs,
-                        offline ? SessionTimeouts::getOfflineSessionMaxIdleMs : SessionTimeouts::getUserSessionMaxIdleMs);
-            } else {
-                Retry.executeWithBackoff((int iteration) -> {
-
-                    try {
-                        remoteCache.putAll(sessionsByIdForTransport);
-                    } catch (HotRodClientException re) {
-                        if (log.isDebugEnabled()) {
-                            log.debugf(re, "Failed to put import %d sessions to remoteCache. Iteration '%s'. Will try to retry the task",
-                                    sessionsByIdForTransport.size(), iteration);
-                        }
-
-                        // Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
-                        throw re;
-                    }
-
-                }, 10, 10);
-            }
+            importSessionsWithExpiration(sessionsByIdForTransport, remoteCache,
+                    offline ? offlineSessionCacheEntryLifespanAdjuster : SessionTimeouts::getUserSessionLifespanMs,
+                    offline ? SessionTimeouts::getOfflineSessionMaxIdleMs : SessionTimeouts::getUserSessionMaxIdleMs);
         }
 
         // Import client sessions
         Cache<UUID, SessionEntityWrapper<AuthenticatedClientSessionEntity>> clientSessCache =
                 CacheDecorators.skipCacheLoadersIfRemoteStoreIsEnabled(offline ? offlineClientSessionCache : clientSessionCache);
 
-        if (importWithExpiration) {
-            importSessionsWithExpiration(clientSessionsById, clientSessCache,
-                    offline ? offlineClientSessionCacheEntryLifespanAdjuster : SessionTimeouts::getClientSessionLifespanMs,
-                    offline ? SessionTimeouts::getOfflineClientSessionMaxIdleMs : SessionTimeouts::getClientSessionMaxIdleMs);
-        } else {
-            Retry.executeWithBackoff((int iteration) -> {
-                clientSessCache.putAll(clientSessionsById);
-            }, 10, 10);
-        }
+        importSessionsWithExpiration(clientSessionsById, clientSessCache,
+                offline ? offlineClientSessionCacheEntryLifespanAdjuster : SessionTimeouts::getClientSessionLifespanMs,
+                offline ? SessionTimeouts::getOfflineClientSessionMaxIdleMs : SessionTimeouts::getClientSessionMaxIdleMs);
 
         // put all entities to the remoteCache (if exists)
         RemoteCache remoteCacheClientSessions = InfinispanUtil.getRemoteCache(clientSessCache);
@@ -898,38 +865,22 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
                     .map(SessionEntityWrapper::forTransport)
                     .collect(Collectors.toMap(sessionEntityWrapper -> sessionEntityWrapper.getEntity().getId(), Function.identity()));
 
-            if (importWithExpiration) {
-                importSessionsWithExpiration(sessionsByIdForTransport, remoteCacheClientSessions,
-                        offline ? offlineClientSessionCacheEntryLifespanAdjuster : SessionTimeouts::getClientSessionLifespanMs,
-                        offline ? SessionTimeouts::getOfflineClientSessionMaxIdleMs : SessionTimeouts::getClientSessionMaxIdleMs);
-            } else {
-                Retry.executeWithBackoff((int iteration) -> {
-
-                    try {
-                        remoteCacheClientSessions.putAll(sessionsByIdForTransport);
-                    } catch (HotRodClientException re) {
-                        if (log.isDebugEnabled()) {
-                            log.debugf(re, "Failed to put import %d client sessions to remoteCache. Iteration '%s'. Will try to retry the task",
-                                    sessionsByIdForTransport.size(), iteration);
-                        }
-
-                        // Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
-                        throw re;
-                    }
-
-                }, 10, 10);
-            }
+            importSessionsWithExpiration(sessionsByIdForTransport, remoteCacheClientSessions,
+                    offline ? offlineClientSessionCacheEntryLifespanAdjuster : SessionTimeouts::getClientSessionLifespanMs,
+                    offline ? SessionTimeouts::getOfflineClientSessionMaxIdleMs : SessionTimeouts::getClientSessionMaxIdleMs);
         }
+
+        return sessionsById.entrySet().stream().findFirst().map(Map.Entry::getValue).orElse(null);
     }
 
-    private <T extends SessionEntity> void importSessionsWithExpiration(Map<? extends Object, SessionEntityWrapper<T>> sessionsById,
-                                                                        BasicCache cache, SessionFunction<T> lifespanMsCalculator,
-                                                                        SessionFunction<T> maxIdleTimeMsCalculator) {
-        sessionsById.forEach((id, sessionEntityWrapper) -> {
+    private <T extends SessionEntity, K> Map<K, SessionEntityWrapper<T>> importSessionsWithExpiration(Map<K, SessionEntityWrapper<T>> sessionsById,
+                                                                              BasicCache<K, SessionEntityWrapper<T>> cache, SessionFunction<T> lifespanMsCalculator,
+                                                                              SessionFunction<T> maxIdleTimeMsCalculator) {
+        return sessionsById.entrySet().stream().map(entry -> {
 
-            T sessionEntity = sessionEntityWrapper.getEntity();
+            T sessionEntity = entry.getValue().getEntity();
             RealmModel currentRealm = session.realms().getRealm(sessionEntity.getRealmId());
-            ClientModel client = sessionEntityWrapper.getClientIfNeeded(currentRealm);
+            ClientModel client = entry.getValue().getClientIfNeeded(currentRealm);
             long lifespan = lifespanMsCalculator.apply(currentRealm, client, sessionEntity);
             long maxIdle = maxIdleTimeMsCalculator.apply(currentRealm, client, sessionEntity);
 
@@ -939,7 +890,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
                     Retry.executeWithBackoff((int iteration) -> {
 
                         try {
-                            cache.put(id, sessionEntityWrapper, lifespan, TimeUnit.MILLISECONDS, maxIdle, TimeUnit.MILLISECONDS);
+                            cache.putIfAbsent(entry.getKey(), entry.getValue(), lifespan, TimeUnit.MILLISECONDS, maxIdle, TimeUnit.MILLISECONDS);
                         } catch (HotRodClientException re) {
                             if (log.isDebugEnabled()) {
                                 log.debugf(re, "Failed to put import %d sessions to remoteCache. Iteration '%s'. Will try to retry the task",
@@ -952,10 +903,13 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
 
                     }, 10, 10);
                 } else {
-                    cache.put(id, sessionEntityWrapper, lifespan, TimeUnit.MILLISECONDS, maxIdle, TimeUnit.MILLISECONDS);
+                    cache.putIfAbsent(entry.getKey(), entry.getValue(), lifespan, TimeUnit.MILLISECONDS, maxIdle, TimeUnit.MILLISECONDS);
                 }
+                return entry;
+            } else {
+                return null;
             }
-        });
+        }).filter(Objects::nonNull).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
     private UserSessionEntity createUserSessionEntityInstance(UserSessionModel userSession) {
@@ -1021,7 +975,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
         AuthenticatedClientSessionStore clientSessions = sessionToImportInto.getEntity().getAuthenticatedClientSessions();
         clientSessions.put(clientSession.getClient().getId(), clientSessionId);
 
-        SessionUpdateTask registerClientSessionTask = new RegisterClientSessionTask(clientSession.getClient().getId(), clientSessionId);
+        SessionUpdateTask<UserSessionEntity> registerClientSessionTask = new RegisterClientSessionTask(clientSession.getClient().getId(), clientSessionId);
         userSessionUpdateTx.addTask(sessionToImportInto.getId(), registerClientSessionTask);
 
         return new AuthenticatedClientSessionAdapter(session, this, entity, clientSession.getClient(), sessionToImportInto, clientSessionUpdateTx, offline);
@@ -1030,12 +984,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
 
     private AuthenticatedClientSessionEntity createAuthenticatedClientSessionInstance(String userSessionId, AuthenticatedClientSessionModel clientSession,
                                                                                       String realmId, String clientId, boolean offline) {
-        final UUID clientSessionId;
-        if (offline) {
-            clientSessionId = keyGenerator.generateKeyUUID(session, clientSessionCache);
-        } else {
-            clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSessionId, clientId);
-        }
+        final UUID clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSessionId, clientId);
         AuthenticatedClientSessionEntity entity = new AuthenticatedClientSessionEntity(clientSessionId);
         entity.setRealmId(realmId);
 
@@ -1084,7 +1033,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
             clientSessionUpdateTx.addTask(clientSession.getId(), null, clientSession, UserSessionModel.SessionPersistenceState.PERSISTENT);
         }
 
-        return sessionTx.get(realm, userSessionEntity.getId());
+        return userSessionUpdateTx.get(userSessionEntity.getId());
 
     }
 
@@ -1128,6 +1077,7 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
     }
 
     public static UUID createClientSessionUUID(String userSessionId, String clientId) {
+        // This allows creating a UUID that is constant even if the entry is reloaded from the database
         return UUID.nameUUIDFromBytes((userSessionId + clientId).getBytes(StandardCharsets.UTF_8));
     }
 }
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/ClientSessionPersistentChangelogBasedTransaction.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/ClientSessionPersistentChangelogBasedTransaction.java
index 640a212a8d..0360119834 100644
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/ClientSessionPersistentChangelogBasedTransaction.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/ClientSessionPersistentChangelogBasedTransaction.java
@@ -104,16 +104,7 @@ public class ClientSessionPersistentChangelogBasedTransaction extends Persistent
 
     private AuthenticatedClientSessionEntity createAuthenticatedClientSessionInstance(String userSessionId, AuthenticatedClientSessionModel clientSession,
                                                                                       String realmId, String clientId) {
-        UUID clientSessionId = null;
-        if (clientSession.getId() != null) {
-            clientSessionId = UUID.fromString(clientSession.getId());
-        } else {
-            if (offline) {
-                clientSessionId = keyGenerator.generateKeyUUID(kcSession, cache);
-            } else {
-                clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSessionId, clientId);
-            }
-        }
+        UUID clientSessionId = PersistentUserSessionProvider.createClientSessionUUID(userSessionId, clientId);
 
         AuthenticatedClientSessionEntity entity = new AuthenticatedClientSessionEntity(clientSessionId);
         entity.setRealmId(realmId);
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/InfinispanChangelogBasedTransaction.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/InfinispanChangelogBasedTransaction.java
index 3cb5b2aad9..a062d4ad1f 100644
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/InfinispanChangelogBasedTransaction.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/InfinispanChangelogBasedTransaction.java
@@ -38,6 +38,8 @@ import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheInvoker;
 import org.keycloak.connections.infinispan.InfinispanUtil;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
 
 /**
@@ -73,7 +75,7 @@ public class InfinispanChangelogBasedTransaction<K, V extends SessionEntity> ext
     public void addTask(K key, SessionUpdateTask<V> task) {
         SessionUpdatesList<V> myUpdates = updates.get(key);
         if (myUpdates == null) {
-            if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && (Objects.equals(cacheName, USER_SESSION_CACHE_NAME) || Objects.equals(cacheName, CLIENT_SESSION_CACHE_NAME))) {
+            if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && (Objects.equals(cacheName, USER_SESSION_CACHE_NAME) || Objects.equals(cacheName, CLIENT_SESSION_CACHE_NAME) || Objects.equals(cacheName, OFFLINE_USER_SESSION_CACHE_NAME) || Objects.equals(cacheName, OFFLINE_CLIENT_SESSION_CACHE_NAME))) {
                 throw new IllegalStateException("Can't load from cache");
             }
 
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/JpaChangesPerformer.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/JpaChangesPerformer.java
index 68cd4f377e..7b6ffaa19e 100644
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/JpaChangesPerformer.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/JpaChangesPerformer.java
@@ -47,6 +47,11 @@ import java.util.Map;
 import java.util.UUID;
 import java.util.function.Consumer;
 
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
+
 public class JpaChangesPerformer<K, V extends SessionEntity> implements SessionChangesPerformer<K, V> {
 
     private final KeycloakSession session;
@@ -69,8 +74,8 @@ public class JpaChangesPerformer<K, V extends SessionEntity> implements SessionC
 
     private TriConsumer<KeycloakSession, Map.Entry<K, SessionUpdatesList<V>>, MergedUpdate<V>> processor() {
         return switch (cacheName) {
-            case "sessions", "offlineSessions" -> this::processUserSessionUpdate;
-            case "clientSessions", "offlineClientSessions" -> this::processClientSessionUpdate;
+            case USER_SESSION_CACHE_NAME, OFFLINE_USER_SESSION_CACHE_NAME -> this::processUserSessionUpdate;
+            case CLIENT_SESSION_CACHE_NAME, OFFLINE_CLIENT_SESSION_CACHE_NAME -> this::processClientSessionUpdate;
             default -> throw new IllegalStateException("Unexpected value: " + cacheName);
         };
     }
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/PersistentSessionsChangelogBasedTransaction.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/PersistentSessionsChangelogBasedTransaction.java
index 9007ce3b11..d623a7220c 100644
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/PersistentSessionsChangelogBasedTransaction.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/PersistentSessionsChangelogBasedTransaction.java
@@ -30,6 +30,8 @@ import java.util.List;
 import java.util.Map;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
 
 public class PersistentSessionsChangelogBasedTransaction<K, V extends SessionEntity> extends InfinispanChangelogBasedTransaction<K, V> {
@@ -45,7 +47,8 @@ public class PersistentSessionsChangelogBasedTransaction<K, V extends SessionEnt
             throw new IllegalStateException("Persistent user sessions are not enabled");
         }
 
-        if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && (cache.getName().equals(USER_SESSION_CACHE_NAME) || cache.getName().equals(CLIENT_SESSION_CACHE_NAME))) {
+        if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) &&
+            (cache.getName().equals(USER_SESSION_CACHE_NAME) || cache.getName().equals(CLIENT_SESSION_CACHE_NAME) || cache.getName().equals(OFFLINE_USER_SESSION_CACHE_NAME) || cache.getName().equals(OFFLINE_CLIENT_SESSION_CACHE_NAME))) {
             changesPerformers = List.of(
                     new JpaChangesPerformer<>(session, cache.getName(), offline)
             );
diff --git a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/UserSessionPersistentChangelogBasedTransaction.java b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/UserSessionPersistentChangelogBasedTransaction.java
index 12ecd2d900..ea6b5e8fbf 100644
--- a/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/UserSessionPersistentChangelogBasedTransaction.java
+++ b/model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/UserSessionPersistentChangelogBasedTransaction.java
@@ -35,6 +35,8 @@ import java.util.Collections;
 import java.util.Objects;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
 
 public class UserSessionPersistentChangelogBasedTransaction extends PersistentSessionsChangelogBasedTransaction<String, UserSessionEntity> {
@@ -48,7 +50,7 @@ public class UserSessionPersistentChangelogBasedTransaction extends PersistentSe
         SessionUpdatesList<UserSessionEntity> myUpdates = updates.get(key);
         if (myUpdates == null) {
             SessionEntityWrapper<UserSessionEntity> wrappedEntity = null;
-            if (!((Objects.equals(cache.getName(), USER_SESSION_CACHE_NAME) || Objects.equals(cache.getName(), CLIENT_SESSION_CACHE_NAME)) && Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE))) {
+            if (!((Objects.equals(cache.getName(), USER_SESSION_CACHE_NAME) || Objects.equals(cache.getName(), CLIENT_SESSION_CACHE_NAME) || Objects.equals(cache.getName(), OFFLINE_USER_SESSION_CACHE_NAME) || Objects.equals(cache.getName(), OFFLINE_CLIENT_SESSION_CACHE_NAME)) && Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE))) {
                 wrappedEntity = cache.get(key);
             }
             if (wrappedEntity == null) {
@@ -111,15 +113,12 @@ public class UserSessionPersistentChangelogBasedTransaction extends PersistentSe
             return null;
         }
 
-        if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && (cache.getName().equals(USER_SESSION_CACHE_NAME) || cache.getName().equals(CLIENT_SESSION_CACHE_NAME))) {
+        if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && (cache.getName().equals(USER_SESSION_CACHE_NAME) || cache.getName().equals(CLIENT_SESSION_CACHE_NAME) || cache.getName().equals(OFFLINE_USER_SESSION_CACHE_NAME) || cache.getName().equals(OFFLINE_CLIENT_SESSION_CACHE_NAME))) {
             return ((PersistentUserSessionProvider) kcSession.getProvider(UserSessionProvider.class)).wrapPersistentEntity(persistentUserSession.getRealm(), offline, persistentUserSession);
         }
 
         LOG.debugf("Attempting to import user-session for sessionId=%s offline=%s", sessionId, offline);
-        kcSession.sessions().importUserSessions(Collections.singleton(persistentUserSession), offline);
-        LOG.debugf("user-session imported, trying another lookup for sessionId=%s offline=%s", sessionId, offline);
-
-        SessionEntityWrapper<UserSessionEntity> ispnUserSessionEntity = cache.get(sessionId);
+        SessionEntityWrapper<UserSessionEntity> ispnUserSessionEntity = ((PersistentUserSessionProvider) kcSession.getProvider(UserSessionProvider.class)).importUserSession(persistentUserSession, offline);;
 
         if (ispnUserSessionEntity != null) {
             LOG.debugf("user-session found after import for sessionId=%s offline=%s", sessionId, offline);
diff --git a/model/jpa/src/main/resources/META-INF/jpa-changelog-25.0.0.xml b/model/jpa/src/main/resources/META-INF/jpa-changelog-25.0.0.xml
index 96ca8110f3..38f4d5f1f2 100644
--- a/model/jpa/src/main/resources/META-INF/jpa-changelog-25.0.0.xml
+++ b/model/jpa/src/main/resources/META-INF/jpa-changelog-25.0.0.xml
@@ -17,7 +17,7 @@
   -->
 <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="25.0.0-xxx">
+    <changeSet author="keycloak" id="25.0.0-28265-tables">
         <addColumn tableName="OFFLINE_USER_SESSION">
             <!-- length(broker_session_id) + length(realm_id) <= 1700 for mssql -->
             <column name="BROKER_SESSION_ID" type="VARCHAR(1024)" />
@@ -26,6 +26,13 @@
         <addColumn tableName="OFFLINE_CLIENT_SESSION">
             <column name="VERSION" type="INT" defaultValueNumeric="0" />
         </addColumn>
+        <modifySql dbms="mssql">
+            <!-- ensure that existing rows also get the new values on mssql -->
+            <!-- https://github.com/liquibase/liquibase/issues/4644 -->
+            <replace replace="DEFAULT 0" with="DEFAULT 0 WITH VALUES" />
+        </modifySql>
+    </changeSet>
+    <changeSet author="keycloak" id="25.0.0-28265-index-creation">
         <createIndex tableName="OFFLINE_USER_SESSION" indexName="IDX_OFFLINE_USS_BY_LAST_SESSION_REFRESH" >
             <!-- optimize this index for range queries for expire sessions -->
             <!-- it should also distribute hot segments across realms and online/offline -->
@@ -33,17 +40,14 @@
             <column name="OFFLINE_FLAG" />
             <column name="LAST_SESSION_REFRESH" />
         </createIndex>
+    </changeSet>
+    <changeSet author="keycloak" id="25.0.0-28265-index-cleanup">
         <dropIndex tableName="OFFLINE_USER_SESSION" indexName="IDX_OFFLINE_USS_CREATEDON" />
         <dropIndex tableName="OFFLINE_USER_SESSION" indexName="IDX_OFFLINE_USS_PRELOAD" />
         <dropIndex tableName="OFFLINE_USER_SESSION" indexName="IDX_OFFLINE_USS_BY_USERSESS" />
         <dropIndex tableName="OFFLINE_CLIENT_SESSION" indexName="IDX_OFFLINE_CSS_PRELOAD" />
-        <modifySql dbms="mssql">
-            <!-- ensure that existing rows also get the new values on mssql -->
-            <!-- https://github.com/liquibase/liquibase/issues/4644 -->
-            <replace replace="DEFAULT 0" with="DEFAULT 0 WITH VALUES" />
-        </modifySql>
     </changeSet>
-    <changeSet author="keycloak" id="25.0.0-xxx-2-mysql">
+    <changeSet author="keycloak" id="25.0.0-28265-index-2-mysql">
         <preConditions onSqlOutput="TEST" onFail="MARK_RAN">
             <or>
                 <dbms type="mysql"/>
@@ -57,7 +61,7 @@
             <column name="REALM_ID" />
         </createIndex>
     </changeSet>
-    <changeSet author="keycloak" id="25.0.0-xxx-2-not-mysql">
+    <changeSet author="keycloak" id="25.0.0-28265-index-2-not-mysql">
         <preConditions onSqlOutput="TEST" onFail="MARK_RAN">
             <not>
                 <or>
diff --git a/quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/legacy/infinispan/CacheManagerFactory.java b/quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/legacy/infinispan/CacheManagerFactory.java
index 1254b5d5ed..a6be9b136d 100644
--- a/quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/legacy/infinispan/CacheManagerFactory.java
+++ b/quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/legacy/infinispan/CacheManagerFactory.java
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
 
 import io.micrometer.core.instrument.Metrics;
 import org.infinispan.client.hotrod.impl.ConfigurationProperties;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.configuration.cache.PersistenceConfigurationBuilder;
 import org.infinispan.configuration.global.GlobalConfiguration;
 import org.infinispan.configuration.parsing.ConfigurationBuilderHolder;
@@ -41,6 +42,7 @@ import org.jgroups.protocols.TCP_NIO2;
 import org.jgroups.protocols.UDP;
 import org.jgroups.util.TLS;
 import org.jgroups.util.TLSClientAuth;
+import org.keycloak.common.Profile;
 import org.keycloak.quarkus.runtime.configuration.Configuration;
 
 import javax.net.ssl.SSLContext;
@@ -55,7 +57,10 @@ import static org.keycloak.config.CachingOptions.CACHE_REMOTE_HOST_PROPERTY;
 import static org.keycloak.config.CachingOptions.CACHE_REMOTE_PASSWORD_PROPERTY;
 import static org.keycloak.config.CachingOptions.CACHE_REMOTE_PORT_PROPERTY;
 import static org.keycloak.config.CachingOptions.CACHE_REMOTE_USERNAME_PROPERTY;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.DISTRIBUTED_REPLICATED_CACHE_NAMES;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
 import static org.wildfly.security.sasl.util.SaslMechanismInformation.Names.SCRAM_SHA_512;
 
@@ -103,11 +108,22 @@ public class CacheManagerFactory {
     private DefaultCacheManager startCacheManager() {
         ConfigurationBuilderHolder builder = new ParserRegistry().parse(config);
 
-        if (builder.getNamedConfigurationBuilders().get(USER_SESSION_CACHE_NAME).clustering().cacheMode().isClustered()) {
+        if (builder.getNamedConfigurationBuilders().entrySet().stream().anyMatch(c -> c.getValue().clustering().cacheMode().isClustered())) {
             configureTransportStack(builder);
             configureRemoteStores(builder);
         }
 
+        DISTRIBUTED_REPLICATED_CACHE_NAMES.forEach(cacheName -> {
+            if (!Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) && Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS) &&
+                (cacheName.equals(USER_SESSION_CACHE_NAME) || cacheName.equals(CLIENT_SESSION_CACHE_NAME) || cacheName.equals(OFFLINE_USER_SESSION_CACHE_NAME) || cacheName.equals(OFFLINE_CLIENT_SESSION_CACHE_NAME))) {
+                ConfigurationBuilder configurationBuilder = builder.getNamedConfigurationBuilders().get(cacheName);
+                if (configurationBuilder.memory().maxSize() == null && configurationBuilder.memory().maxCount() == -1) {
+                    logger.infof("Persistent user sessions enabled and no memory limit found in configuration. Setting max entries for %s to 10000 entries", cacheName);
+                    configurationBuilder.memory().maxCount(10000);
+                }
+            }
+        });
+
         if (metricsEnabled) {
             builder.getGlobalConfigurationBuilder().addModule(MicrometerMeterRegisterConfigurationBuilder.class);
             builder.getGlobalConfigurationBuilder().module(MicrometerMeterRegisterConfigurationBuilder.class).meterRegistry(Metrics.globalRegistry);
@@ -223,6 +239,11 @@ public class CacheManagerFactory {
             }
 
             DISTRIBUTED_REPLICATED_CACHE_NAMES.forEach(cacheName -> {
+                if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE) &&
+                    (cacheName.equals(USER_SESSION_CACHE_NAME) || cacheName.equals(CLIENT_SESSION_CACHE_NAME) || cacheName.equals(OFFLINE_USER_SESSION_CACHE_NAME) || cacheName.equals(OFFLINE_CLIENT_SESSION_CACHE_NAME))) {
+                    return;
+                }
+
                 PersistenceConfigurationBuilder persistenceCB = builder.getNamedConfigurationBuilders().get(cacheName).persistence();
 
                 //if specified via command line -> cannot be defined in the xml file
diff --git a/server-spi/src/main/java/org/keycloak/models/UserSessionProvider.java b/server-spi/src/main/java/org/keycloak/models/UserSessionProvider.java
index 9ff68a5559..df30bae6cf 100755
--- a/server-spi/src/main/java/org/keycloak/models/UserSessionProvider.java
+++ b/server-spi/src/main/java/org/keycloak/models/UserSessionProvider.java
@@ -198,7 +198,11 @@ public interface UserSessionProvider extends Provider {
      */
     Stream<UserSessionModel> getOfflineUserSessionsStream(RealmModel realm, ClientModel client, Integer firstResult, Integer maxResults);
 
-    /** Triggered by persister during pre-load. It imports authenticatedClientSessions too **/
+    /** Triggered by persister during pre-load. It imports authenticatedClientSessions too.
+     *
+     * @deprecated Deprecated as offline session preloading was removed in KC25. This method will be removed in KC27.
+     */
+    @Deprecated(forRemoval = true)
     void importUserSessions(Collection<UserSessionModel> persistentUserSessions, boolean offline);
 
     void close();
diff --git a/testsuite/integration-arquillian/tests/base/testsuites/persistent-sessions-suite b/testsuite/integration-arquillian/tests/base/testsuites/persistent-sessions-suite
index bd134a8056..a46032a52b 100644
--- a/testsuite/integration-arquillian/tests/base/testsuites/persistent-sessions-suite
+++ b/testsuite/integration-arquillian/tests/base/testsuites/persistent-sessions-suite
@@ -19,4 +19,8 @@ ConcurrentLoginTest
 RefreshTokenTest
 OfflineTokenTest
 AccessTokenTest
-LogoutTest
\ No newline at end of file
+LogoutTest
+ClientStorageTest
+UserInfoTest
+LightWeightAccessTokenTest
+TokenIntrospectionTest
\ No newline at end of file
diff --git a/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/OfflineSessionPersistenceTest.java b/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/OfflineSessionPersistenceTest.java
index 5ded5fbef3..423cc322d7 100644
--- a/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/OfflineSessionPersistenceTest.java
+++ b/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/OfflineSessionPersistenceTest.java
@@ -335,6 +335,7 @@ public class OfflineSessionPersistenceTest extends KeycloakModelTest {
     private String createOfflineClientSession(String offlineUserSessionId, String clientId) {
         return withRealm(realmId, (session, realm) -> {
             UserSessionModel offlineUserSession = session.sessions().getOfflineUserSession(realm, offlineUserSessionId);
+            assertThat("Can't retrieve offline session for " + offlineUserSessionId, offlineUserSession, Matchers.notNullValue());
             ClientModel client = session.clients().getClientById(realm, clientId);
             AuthenticatedClientSessionModel clientSession = session.sessions().createClientSession(realm, client, offlineUserSession);
             return session.sessions().createOfflineClientSession(clientSession, offlineUserSession).getId();
diff --git a/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionProviderOfflineModelTest.java b/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionProviderOfflineModelTest.java
index cd63439bec..0fd58d6985 100644
--- a/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionProviderOfflineModelTest.java
+++ b/testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionProviderOfflineModelTest.java
@@ -481,6 +481,8 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
     public void testOfflineSessionLifespanOverride() {
         // skip the test for CrossDC
         Assume.assumeFalse(Objects.equals(CONFIG.scope("connectionsInfinispan.default").get("remoteStoreEnabled"), "true"));
+        // As we don't put things in the embedded cache, the test will fail as the number of entries will always be 0.
+        Assume.assumeFalse(Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS_NO_CACHE));
 
         createOfflineSessions("user1", 2, new LinkedList<>(), new LinkedList<>());