External Infinispan as cache - Part 4 (#30072)

UserSessionProvider implementation to make use of Infinispan remote
cache.

Closes #28755

Signed-off-by: Pedro Ruivo <pruivo@redhat.com>

.github/workflows/ci.yml

@@ -377,7 +377,7 @@ jobs:
 
       - name: Run base tests without cache
         run: |
-          TESTS=`testsuite/integration-arquillian/tests/base/testsuites/suite.sh persistent-sessions`
+          TESTS=`testsuite/integration-arquillian/tests/base/testsuites/suite.sh remote-cache`
           echo "Tests: $TESTS"
           ./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Pinfinispan-server -Dauth.server.feature=${{ matrix.variant }} -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
 

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

@@ -332,15 +332,15 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
                     .stateTransfer().awaitInitialTransfer(awaitInitialTransfer).timeout(30, TimeUnit.SECONDS);
         }
 
-        // Base configuration doesn't contain any remote stores
-        var clusteredConfiguration = builder.build();
-
-        defineClusteredCache(cacheManager, USER_SESSION_CACHE_NAME, clusteredConfiguration);
-        defineClusteredCache(cacheManager, OFFLINE_USER_SESSION_CACHE_NAME, clusteredConfiguration);
-        defineClusteredCache(cacheManager, CLIENT_SESSION_CACHE_NAME, clusteredConfiguration);
-        defineClusteredCache(cacheManager, OFFLINE_CLIENT_SESSION_CACHE_NAME, clusteredConfiguration);
-
         if (InfinispanUtils.isEmbeddedInfinispan()) {
+            // Base configuration doesn't contain any remote stores
+            var clusteredConfiguration = builder.build();
+
+            defineClusteredCache(cacheManager, USER_SESSION_CACHE_NAME, clusteredConfiguration);
+            defineClusteredCache(cacheManager, OFFLINE_USER_SESSION_CACHE_NAME, clusteredConfiguration);
+            defineClusteredCache(cacheManager, CLIENT_SESSION_CACHE_NAME, clusteredConfiguration);
+            defineClusteredCache(cacheManager, OFFLINE_CLIENT_SESSION_CACHE_NAME, clusteredConfiguration);
+
             defineClusteredCache(cacheManager, LOGIN_FAILURE_CACHE_NAME, clusteredConfiguration);
             defineClusteredCache(cacheManager, AUTHENTICATION_SESSIONS_CACHE_NAME, clusteredConfiguration);
 

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

@@ -57,7 +57,6 @@ import org.keycloak.models.AuthenticatedClientSessionModel;
 import org.keycloak.models.ClientModel;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.ModelException;
-import org.keycloak.models.OfflineUserSessionModel;
 import org.keycloak.models.RealmModel;
 import org.keycloak.models.UserModel;
 import org.keycloak.models.UserProvider;
@@ -204,15 +203,7 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
     @Override
     public AuthenticatedClientSessionModel createClientSession(RealmModel realm, ClientModel client, UserSessionModel userSession) {
         final UUID clientSessionId = keyGenerator.generateKeyUUID(session, clientSessionCache);
-        AuthenticatedClientSessionEntity entity = new AuthenticatedClientSessionEntity(clientSessionId);
+        var entity = AuthenticatedClientSessionEntity.create(clientSessionId, realm, client, userSession);
-        entity.setRealmId(realm.getId());
-        entity.setClientId(client.getId());
-        entity.setTimestamp(Time.currentTime());
-        entity.getNotes().put(AuthenticatedClientSessionModel.STARTED_AT_NOTE, String.valueOf(entity.getTimestamp()));
-        entity.getNotes().put(AuthenticatedClientSessionModel.USER_SESSION_STARTED_AT_NOTE, String.valueOf(userSession.getStarted()));
-        if (userSession.isRememberMe()) {
-            entity.getNotes().put(AuthenticatedClientSessionModel.USER_SESSION_REMEMBER_ME_NOTE, "true");
-        }
 
         InfinispanChangelogBasedTransaction<String, UserSessionEntity> userSessionUpdateTx = getTransaction(false);
         InfinispanChangelogBasedTransaction<UUID, AuthenticatedClientSessionEntity> clientSessionUpdateTx = getClientSessionTransaction(false);
@@ -238,8 +229,7 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
             id = keyGenerator.generateKeyString(session, sessionCache);
         }
 
-        UserSessionEntity entity = new UserSessionEntity(id);
+        UserSessionEntity entity = UserSessionEntity.create(id, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
-        updateSessionEntity(entity, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
 
         SessionUpdateTask<UserSessionEntity> createSessionTask = Tasks.addIfAbsentSync();
         sessionTx.addTask(id, createSessionTask, entity, persistenceState);
@@ -251,21 +241,6 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
         return adapter;
     }
 
-    static void updateSessionEntity(UserSessionEntity entity, RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId) {
-        entity.setRealmId(realm.getId());
-        entity.setUser(user.getId());
-        entity.setLoginUsername(loginUsername);
-        entity.setIpAddress(ipAddress);
-        entity.setAuthMethod(authMethod);
-        entity.setRememberMe(rememberMe);
-        entity.setBrokerSessionId(brokerSessionId);
-        entity.setBrokerUserId(brokerUserId);
-
-        int currentTime = Time.currentTime();
-
-        entity.setStarted(currentTime);
-        entity.setLastSessionRefresh(currentTime);
-    }
 
     @Override
     public UserSessionModel getUserSession(RealmModel realm, String id) {
@@ -889,7 +864,7 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
         Map<String, SessionEntityWrapper<UserSessionEntity>> sessionsById = persistentUserSessions.stream()
                 .map((UserSessionModel persistentUserSession) -> {
 
-                    UserSessionEntity userSessionEntityToImport = createUserSessionEntityInstance(persistentUserSession);
+                    UserSessionEntity userSessionEntityToImport = UserSessionEntity.createFromModel(persistentUserSession);
 
                     for (Map.Entry<String, AuthenticatedClientSessionModel> entry : persistentUserSession.getAuthenticatedClientSessions().entrySet()) {
                         String clientUUID = entry.getKey();
@@ -1039,7 +1014,7 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
 
     // Imports just userSession without it's clientSessions
     protected UserSessionAdapter importUserSession(UserSessionModel userSession, boolean offline) {
-        UserSessionEntity entity = createUserSessionEntityInstance(userSession);
+        UserSessionEntity entity = UserSessionEntity.createFromModel(userSession);
 
         InfinispanChangelogBasedTransaction<String, UserSessionEntity> userSessionUpdateTx = getTransaction(offline);
 
@@ -1052,38 +1027,6 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
     }
 
 
-    private UserSessionEntity createUserSessionEntityInstance(UserSessionModel userSession) {
-        UserSessionEntity entity = new UserSessionEntity(userSession.getId());
-        entity.setRealmId(userSession.getRealm().getId());
-
-        entity.setAuthMethod(userSession.getAuthMethod());
-        entity.setBrokerSessionId(userSession.getBrokerSessionId());
-        entity.setBrokerUserId(userSession.getBrokerUserId());
-        entity.setIpAddress(userSession.getIpAddress());
-        entity.setNotes(userSession.getNotes() == null ? new ConcurrentHashMap<>() : userSession.getNotes());
-        entity.setAuthenticatedClientSessions(new AuthenticatedClientSessionStore());
-        entity.setRememberMe(userSession.isRememberMe());
-        entity.setState(userSession.getState());
-        if (userSession instanceof OfflineUserSessionModel) {
-            // this is a hack so that UserModel doesn't have to be available when offline token is imported.
-            // see related JIRA - KEYCLOAK-5350 and corresponding test
-            OfflineUserSessionModel oline = (OfflineUserSessionModel) userSession;
-            entity.setUser(oline.getUserId());
-            // NOTE: Hack
-            // We skip calling entity.setLoginUsername(userSession.getLoginUsername())
-
-        } else {
-            entity.setLoginUsername(userSession.getLoginUsername());
-            entity.setUser(userSession.getUser().getId());
-        }
-
-        entity.setStarted(userSession.getStarted());
-        entity.setLastSessionRefresh(userSession.getLastSessionRefresh());
-
-        return entity;
-    }
-
-
     private AuthenticatedClientSessionAdapter importClientSession(UserSessionAdapter sessionToImportInto, AuthenticatedClientSessionModel clientSession,
                                                                   InfinispanChangelogBasedTransaction<String, UserSessionEntity> userSessionUpdateTx,
                                                                   InfinispanChangelogBasedTransaction<UUID, AuthenticatedClientSessionEntity> clientSessionUpdateTx,

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

@@ -69,13 +69,14 @@ import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.models.utils.ResetTimeOffsetEvent;
+import org.keycloak.provider.EnvironmentDependentProviderFactory;
 import org.keycloak.provider.ProviderConfigProperty;
 import org.keycloak.provider.ProviderConfigurationBuilder;
 import org.keycloak.provider.ProviderEvent;
 import org.keycloak.provider.ProviderEventListener;
 import org.keycloak.provider.ServerInfoAwareProviderFactory;
 
-public class InfinispanUserSessionProviderFactory implements UserSessionProviderFactory, ServerInfoAwareProviderFactory {
+public class InfinispanUserSessionProviderFactory implements UserSessionProviderFactory, ServerInfoAwareProviderFactory, EnvironmentDependentProviderFactory {
 
     private static final Logger log = Logger.getLogger(InfinispanUserSessionProviderFactory.class);
 
@@ -179,11 +180,7 @@ public class InfinispanUserSessionProviderFactory implements UserSessionProvider
                             initializeLastSessionRefreshStore(factory);
                         }
                         registerClusterListeners(session);
-                        // TODO [pruivo] to remove: workaround to run the testsuite.
+                        loadSessionsFromRemoteCaches(session);
-                        if (InfinispanUtils.isEmbeddedInfinispan()) {
-                            loadSessionsFromRemoteCaches(session);
-                        }
-
                     }, preloadTransactionTimeout);
 
                 } else if (event instanceof UserModel.UserRemovedEvent) {
@@ -429,6 +426,11 @@ public class InfinispanUserSessionProviderFactory implements UserSessionProvider
         return InfinispanUtils.PROVIDER_ORDER;
     }
 
+    @Override
+    public boolean isSupported(Config.Scope config) {
+        return InfinispanUtils.isEmbeddedInfinispan();
+    }
+
     @Override
     public Map<String, String> getOperationalInfo() {
         Map<String, String> info = new HashMap<>();

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

@@ -371,7 +371,7 @@ public class UserSessionAdapter<T extends SessionRefreshStore & UserSessionProvi
 
             @Override
             public void runUpdate(UserSessionEntity entity) {
-                InfinispanUserSessionProvider.updateSessionEntity(entity, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
+                UserSessionEntity.updateSessionEntity(entity, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
 
                 entity.setState(null);
                 entity.getNotes().clear();

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/RemoteChangeLogTransaction.java

@@ -18,6 +18,7 @@ package org.keycloak.models.sessions.infinispan.changes.remote;
 
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -26,18 +27,19 @@ import java.util.function.Predicate;
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.Flowable;
 import org.infinispan.client.hotrod.Flag;
+import org.infinispan.client.hotrod.MetadataValue;
 import org.infinispan.client.hotrod.RemoteCache;
 import org.infinispan.commons.util.concurrent.AggregateCompletionStage;
 import org.infinispan.commons.util.concurrent.CompletableFutures;
 import org.infinispan.commons.util.concurrent.CompletionStages;
 import org.keycloak.models.AbstractKeycloakTransaction;
-import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakTransaction;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.Expiration;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.Updater;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.UpdaterFactory;
 
 /**
- * A {@link org.keycloak.models.KeycloakTransaction} implementation that keeps track of changes made to entities stored
+ * A {@link KeycloakTransaction} implementation that keeps track of changes made to entities stored
  * in a Infinispan cache.
  *
  * @param <K> The type of the Infinispan cache key.
@@ -50,13 +52,11 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
     private final Map<K, T> entityChanges;
     private final UpdaterFactory<K, V, T> factory;
     private final RemoteCache<K, V> cache;
-    private final KeycloakSession session;
     private Predicate<V> removePredicate;
 
-    public RemoteChangeLogTransaction(UpdaterFactory<K, V, T> factory, RemoteCache<K, V> cache, KeycloakSession session) {
+    public RemoteChangeLogTransaction(UpdaterFactory<K, V, T> factory, RemoteCache<K, V> cache) {
         this.factory = Objects.requireNonNull(factory);
         this.cache = Objects.requireNonNull(cache);
-        this.session = Objects.requireNonNull(session);
         entityChanges = new ConcurrentHashMap<>(8);
     }
 
@@ -75,6 +75,16 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
         removePredicate = null;
     }
 
+    public void commitAsync(AggregateCompletionStage<Void> stage) {
+        if (state != TransactionState.STARTED) {
+            throw new IllegalStateException("Transaction in illegal state for commit: " + state);
+        }
+
+        doCommit(stage);
+
+        state = TransactionState.FINISHED;
+    }
+
     private void doCommit(AggregateCompletionStage<Void> stage) {
         if (removePredicate != null) {
             // TODO [pruivo] [optimization] with protostream, use delete by query: DELETE FROM ...
@@ -87,7 +97,7 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
         }
 
         for (var updater : entityChanges.values()) {
-            if (updater.isReadOnly() || (removePredicate != null && removePredicate.test(updater.getValue()))) {
+            if (updater.isReadOnly() || updater.isTransient() || (removePredicate != null && removePredicate.test(updater.getValue()))) {
                 continue;
             }
             if (updater.isDeleted()) {
@@ -95,7 +105,7 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
                 continue;
             }
 
-            var expiration = updater.computeExpiration(session);
+            var expiration = updater.computeExpiration();
 
             if (expiration.isExpired()) {
                 stage.dependsOn(cache.removeAsync(updater.getKey()));
@@ -129,15 +139,23 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
     public T get(K key) {
         var updater = entityChanges.get(key);
         if (updater != null) {
-            return updater;
+            return updater.isDeleted() ? null : updater;
         }
-        var entity = cache.getWithMetadata(key);
+        return onEntityFromCache(key, cache.getWithMetadata(key));
-        if (entity == null) {
+    }
-            return null;
+
+    /**
+     * Nonblocking alternative of {@link #get(Object)}
+     *
+     * @param key The Infinispan cache key to fetch.
+     * @return The {@link Updater} to track further changes of the Infinispan cache value.
+     */
+    public CompletionStage<T> getAsync(K key) {
+        var updater = entityChanges.get(key);
+        if (updater != null) {
+            return updater.isDeleted() ? CompletableFutures.completedNull() : CompletableFuture.completedFuture(updater);
         }
-        updater = factory.wrapFromCache(key, entity);
+        return cache.getWithMetadataAsync(key).thenApply(e -> onEntityFromCache(key, e));
-        entityChanges.put(key, updater);
-        return updater.isDeleted() ? null : updater;
     }
 
     /**
@@ -180,6 +198,19 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
         removePredicate = removePredicate.or(predicate);
     }
 
+    public T wrap(Map.Entry<K, MetadataValue<V>> entry) {
+        return entityChanges.computeIfAbsent(entry.getKey(), k -> factory.wrapFromCache(k, entry.getValue()));
+    }
+
+    private T onEntityFromCache(K key, MetadataValue<V> entity) {
+        if (entity == null) {
+            return null;
+        }
+        var updater = factory.wrapFromCache(key, entity);
+        entityChanges.put(key, updater);
+        return updater.isDeleted() ? null : updater;
+    }
+
     private CompletionStage<V> putIfAbsent(Updater<K, V> updater, Expiration expiration) {
         return cache.withFlags(Flag.FORCE_RETURN_VALUE)
                 .putIfAbsentAsync(updater.getKey(), updater.getValue(), expiration.lifespan(), TimeUnit.MILLISECONDS, expiration.maxIdle(), TimeUnit.MILLISECONDS)
@@ -197,7 +228,7 @@ public class RemoteChangeLogTransaction<K, V, T extends Updater<K, V>> extends A
     }
 
     private CompletionStage<V> merge(Updater<K, V> updater, Expiration expiration) {
-        return cache.computeAsync(updater.getKey(), updater, expiration.lifespan(), TimeUnit.MILLISECONDS, expiration.maxIdle(), TimeUnit.MILLISECONDS);
+        return cache.computeIfPresentAsync(updater.getKey(), updater, expiration.lifespan(), TimeUnit.MILLISECONDS, expiration.maxIdle(), TimeUnit.MILLISECONDS);
     }
 
     private Completable removeKey(K key) {

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/UserSessionTransaction.java

@@ -0,0 +1,76 @@
+package org.keycloak.models.sessions.infinispan.changes.remote;
+
+import java.util.Objects;
+import java.util.UUID;
+
+import org.infinispan.commons.util.concurrent.CompletionStages;
+import org.keycloak.models.AbstractKeycloakTransaction;
+import org.keycloak.models.KeycloakTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.client.AuthenticatedClientSessionUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.UserSessionUpdater;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+
+/**
+ * A {@link KeycloakTransaction} implementation that wraps all the user and client session transactions.
+ * <p>
+ * This implementation commits all modifications asynchronously and concurrently in both user and client sessions
+ * transactions. Waits for all them to complete. This is an optimization to reduce the response time.
+ */
+public class UserSessionTransaction extends AbstractKeycloakTransaction {
+
+    private final RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> userSessions;
+    private final RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> offlineUserSessions;
+    private final RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> clientSessions;
+    private final RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> offlineClientSessions;
+
+    public UserSessionTransaction(RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> userSessions, RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> offlineUserSessions, RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> clientSessions, RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> offlineClientSessions) {
+        this.userSessions = Objects.requireNonNull(userSessions);
+        this.offlineUserSessions = Objects.requireNonNull(offlineUserSessions);
+        this.clientSessions = Objects.requireNonNull(clientSessions);
+        this.offlineClientSessions = Objects.requireNonNull(offlineClientSessions);
+    }
+
+    @Override
+    public void begin() {
+        super.begin();
+        userSessions.begin();
+        offlineUserSessions.begin();
+        clientSessions.begin();
+        offlineClientSessions.begin();
+    }
+
+    @Override
+    protected void commitImpl() {
+        var stage = CompletionStages.aggregateCompletionStage();
+        userSessions.commitAsync(stage);
+        offlineUserSessions.commitAsync(stage);
+        clientSessions.commitAsync(stage);
+        offlineClientSessions.commitAsync(stage);
+        CompletionStages.join(stage.freeze());
+    }
+
+    @Override
+    protected void rollbackImpl() {
+        userSessions.rollback();
+        offlineUserSessions.rollback();
+        clientSessions.rollback();
+        offlineClientSessions.rollback();
+    }
+
+    public RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> getClientSessions() {
+        return clientSessions;
+    }
+
+    public RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> getOfflineClientSessions() {
+        return offlineClientSessions;
+    }
+
+    public RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> getOfflineUserSessions() {
+        return offlineUserSessions;
+    }
+
+    public RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> getUserSessions() {
+        return userSessions;
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/BaseUpdater.java

@@ -16,13 +16,15 @@
  */
 package org.keycloak.models.sessions.infinispan.changes.remote.updater;
 
+import java.util.Objects;
+
 /**
  * Base functionality of an {@link Updater} implementation.
  * <p>
  * It stores the Infinispan cache key, value, version, and it states. However, it does not keep track of the changed
  * fields in the cache value, and it is the responsibility of the implementation to do that.
  * <p>
- * The method {@link #onFieldChanged()} must be invoked to track changes in the cache value.
+ * Implement the method {@link #isUnchanged()} to signal if the entity was modified or not.
  *
  * @param <K> The type of the Infinispan cache key.
  * @param <V> The type of the Infinispan cache value.
@@ -35,10 +37,10 @@ public abstract class BaseUpdater<K, V> implements Updater<K, V> {
     private UpdaterState state;
 
     protected BaseUpdater(K cacheKey, V cacheValue, long versionRead, UpdaterState state) {
-        this.cacheKey = cacheKey;
+        this.cacheKey = Objects.requireNonNull(cacheKey);
         this.cacheValue = cacheValue;
         this.versionRead = versionRead;
-        this.state = state;
+        this.state = Objects.requireNonNull(state);
     }
 
     @Override
@@ -68,7 +70,7 @@ public abstract class BaseUpdater<K, V> implements Updater<K, V> {
 
     @Override
     public final boolean isReadOnly() {
-        return state == UpdaterState.READ_ONLY;
+        return state == UpdaterState.READ && isUnchanged();
     }
 
     @Override
@@ -76,16 +78,38 @@ public abstract class BaseUpdater<K, V> implements Updater<K, V> {
         state = UpdaterState.DELETED;
     }
 
-    /**
+    @Override
-     * Must be invoked when a field change to mark this updated and modified.
+    public boolean equals(Object o) {
-     */
+        if (this == o) return true;
-    protected final void onFieldChanged() {
+        if (o == null || getClass() != o.getClass()) return false;
-        state = state.stateAfterChange();
+
+        BaseUpdater<?, ?> that = (BaseUpdater<?, ?>) o;
+        return cacheKey.equals(that.cacheKey);
     }
 
+    @Override
+    public int hashCode() {
+        return cacheKey.hashCode();
+    }
+
+    @Override
+    public String toString() {
+        return "BaseUpdater{" +
+                "cacheKey=" + cacheKey +
+                ", cacheValue=" + cacheValue +
+                ", state=" + state +
+                ", versionRead=" + versionRead +
+                '}';
+    }
+
+    /**
+     * @return {@code true} if the entity was changed after being created/read.
+     */
+    protected abstract boolean isUnchanged();
+
     protected enum UpdaterState {
         /**
-         * The cache value is created. It implies {@link #MODIFIED}.
+         * The cache value is created.
          */
         CREATED,
         /**
@@ -93,23 +117,8 @@ public abstract class BaseUpdater<K, V> implements Updater<K, V> {
          */
         DELETED,
         /**
-         * The cache value was read the Infinispan cache and was not modified.
+         * The cache value was read from the Infinispan cache.
          */
-        READ_ONLY {
+        READ,
-            @Override
-            UpdaterState stateAfterChange() {
-                return MODIFIED;
-            }
-        },
-        /**
-         * The cache value was read from the Infinispan cache and was modified. Changes will be merged into the current
-         * Infinispan cache value.
-         */
-        MODIFIED;
-
-        UpdaterState stateAfterChange() {
-            return this;
-        }
-
     }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/Updater.java

@@ -16,7 +16,6 @@
  */
 package org.keycloak.models.sessions.infinispan.changes.remote.updater;
 
-import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.sessions.infinispan.changes.remote.RemoteChangeLogTransaction;
 
 import java.util.function.BiFunction;
@@ -68,11 +67,17 @@ public interface Updater<K, V> extends BiFunction<K, V, V> {
      */
     void markDeleted();
 
+    /**
+     * @return {@code true} if the entity is transient and shouldn't be stored in the Infinispan cache.
+     */
+    default boolean isTransient() {
+        return false;
+    }
+
     /**
      * Computes the expiration data for Infinispan cache.
      *
-     * @param session The current Keycloak session.
      * @return The {@link Expiration} data.
      */
-    Expiration computeExpiration(KeycloakSession session);
+    Expiration computeExpiration();
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/client/AuthenticatedClientSessionUpdater.java

@@ -0,0 +1,254 @@
+/*
+ * 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.
+ */
+
+package org.keycloak.models.sessions.infinispan.changes.remote.updater.client;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.Consumer;
+
+import org.infinispan.client.hotrod.MetadataValue;
+import org.infinispan.client.hotrod.RemoteCache;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.changes.remote.RemoteChangeLogTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.BaseUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.Expiration;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.Updater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.UpdaterFactory;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.helper.MapUpdater;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
+
+/**
+ * An {@link Updater} implementation that keeps track of {@link AuthenticatedClientSessionModel} changes.
+ */
+public class AuthenticatedClientSessionUpdater extends BaseUpdater<UUID, AuthenticatedClientSessionEntity> implements AuthenticatedClientSessionModel {
+
+    private static final Factory ONLINE = new Factory(false);
+    private static final Factory OFFLINE = new Factory(true);
+
+    private final MapUpdater<String, String> notesUpdater;
+    private final List<Consumer<AuthenticatedClientSessionEntity>> changes;
+    private final boolean offline;
+    private UserSessionModel userSession;
+    private ClientModel client;
+    private RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> clientTransaction;
+
+    private AuthenticatedClientSessionUpdater(UUID cacheKey, AuthenticatedClientSessionEntity cacheValue, long version, boolean offline, UpdaterState initialState) {
+        super(cacheKey, cacheValue, version, initialState);
+        this.offline = offline;
+        if (cacheValue == null) {
+            assert initialState == UpdaterState.DELETED; // cannot be undone
+            notesUpdater = null;
+            changes = List.of();
+            return;
+        }
+        initNotes(cacheValue);
+        notesUpdater = new MapUpdater<>(cacheValue.getNotes());
+        changes = new ArrayList<>(4);
+    }
+
+    /**
+     * @param offline If {@code true}, it creates offline {@link AuthenticatedClientSessionModel}.
+     * @return The {@link UpdaterFactory} implementation to create instances of
+     * {@link AuthenticatedClientSessionUpdater}.
+     */
+    public static UpdaterFactory<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> factory(boolean offline) {
+        return offline ? OFFLINE : ONLINE;
+    }
+
+    @Override
+    public AuthenticatedClientSessionEntity apply(UUID uuid, AuthenticatedClientSessionEntity entity) {
+        initNotes(entity);
+        notesUpdater.applyChanges(entity.getNotes());
+        changes.forEach(change -> change.accept(entity));
+        return entity;
+    }
+
+    @Override
+    public Expiration computeExpiration() {
+        long maxIdle;
+        long lifespan;
+        if (offline) {
+            maxIdle = SessionTimeouts.getOfflineClientSessionMaxIdleMs(userSession.getRealm(), client, getValue());
+            lifespan = SessionTimeouts.getOfflineClientSessionLifespanMs(userSession.getRealm(), client, getValue());
+        } else {
+            maxIdle = SessionTimeouts.getClientSessionMaxIdleMs(userSession.getRealm(), client, getValue());
+            lifespan = SessionTimeouts.getClientSessionLifespanMs(userSession.getRealm(), client, getValue());
+        }
+        return new Expiration(maxIdle, lifespan);
+    }
+
+    @Override
+    public String getId() {
+        return getValue().getId().toString();
+    }
+
+    @Override
+    public int getTimestamp() {
+        return getValue().getTimestamp();
+    }
+
+    @Override
+    public void setTimestamp(int timestamp) {
+        addAndApplyChange(entity -> entity.setTimestamp(timestamp));
+    }
+
+    @Override
+    public void detachFromUserSession() {
+        clientTransaction.remove(getKey());
+    }
+
+    @Override
+    public UserSessionModel getUserSession() {
+        return userSession;
+    }
+
+    @Override
+    public String getNote(String name) {
+        return notesUpdater.get(name);
+    }
+
+    @Override
+    public void setNote(String name, String value) {
+        notesUpdater.put(name, value);
+    }
+
+    @Override
+    public void removeNote(String name) {
+        notesUpdater.remove(name);
+    }
+
+    @Override
+    public Map<String, String> getNotes() {
+        return notesUpdater;
+    }
+
+    @Override
+    public String getRedirectUri() {
+        return getValue().getRedirectUri();
+    }
+
+    @Override
+    public void setRedirectUri(String uri) {
+        addAndApplyChange(entity -> entity.setRedirectUri(uri));
+    }
+
+    @Override
+    public RealmModel getRealm() {
+        return userSession.getRealm();
+    }
+
+    @Override
+    public ClientModel getClient() {
+        return client;
+    }
+
+    @Override
+    public String getAction() {
+        return getValue().getAction();
+    }
+
+    @Override
+    public void setAction(String action) {
+        addAndApplyChange(entity -> entity.setAction(action));
+    }
+
+    @Override
+    public String getProtocol() {
+        return getValue().getAuthMethod();
+    }
+
+    @Override
+    public void setProtocol(String method) {
+        addAndApplyChange(entity -> entity.setAuthMethod(method));
+    }
+
+    @Override
+    public boolean isTransient() {
+        return !isDeleted() && userSession.getPersistenceState() == UserSessionModel.SessionPersistenceState.TRANSIENT;
+    }
+
+    @Override
+    protected boolean isUnchanged() {
+        return changes.isEmpty() && notesUpdater.isUnchanged();
+    }
+
+    /**
+     * Initializes this class with references to other models classes.
+     *
+     * @param userSession       The {@link UserSessionModel} associated with this client session.
+     * @param client            The {@link ClientModel} associated with this client session.
+     * @param clientTransaction The {@link RemoteChangeLogTransaction} to perform the changes in this class into the
+     *                          {@link RemoteCache}.
+     */
+    public synchronized void initialize(UserSessionModel userSession, ClientModel client, RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> clientTransaction) {
+        this.userSession = Objects.requireNonNull(userSession);
+        this.client = Objects.requireNonNull(client);
+        this.clientTransaction = Objects.requireNonNull(clientTransaction);
+    }
+
+    /**
+     * @return {@code true} if it is already initialized.
+     */
+    public synchronized boolean isInitialized() {
+        return userSession != null;
+    }
+
+    /**
+     * Keeps track of a model changes and applies it to the entity.
+     */
+    private void addAndApplyChange(Consumer<AuthenticatedClientSessionEntity> change) {
+        changes.add(change);
+        change.accept(getValue());
+    }
+
+    private static void initNotes(AuthenticatedClientSessionEntity entity) {
+        var notes = entity.getNotes();
+        if (notes == null) {
+            entity.setNotes(new HashMap<>());
+        }
+    }
+
+    private record Factory(
+            boolean offline) implements UpdaterFactory<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> {
+
+        @Override
+        public AuthenticatedClientSessionUpdater create(UUID key, AuthenticatedClientSessionEntity entity) {
+            return new AuthenticatedClientSessionUpdater(key, Objects.requireNonNull(entity), -1, offline, UpdaterState.CREATED);
+        }
+
+        @Override
+        public AuthenticatedClientSessionUpdater wrapFromCache(UUID key, MetadataValue<AuthenticatedClientSessionEntity> entity) {
+            assert entity != null;
+            return new AuthenticatedClientSessionUpdater(key, Objects.requireNonNull(entity.getValue()), entity.getVersion(), offline, UpdaterState.READ);
+        }
+
+        @Override
+        public AuthenticatedClientSessionUpdater deleted(UUID key) {
+            return new AuthenticatedClientSessionUpdater(key, null, -1, offline, UpdaterState.DELETED);
+        }
+    }
+
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/helper/MapUpdater.java

@@ -0,0 +1,99 @@
+/*
+ * 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.
+ */
+
+package org.keycloak.models.sessions.infinispan.changes.remote.updater.helper;
+
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+
+/**
+ * An {@link Map} implementation that keeps track of any modification performed in the {@link Map}.
+ * <p>
+ * The modifications can be replayed in another {@link Map} instance.
+ *
+ * @param <K> The key type.
+ * @param <V> The value type.
+ */
+public class MapUpdater<K, V> extends AbstractMap<K, V> {
+
+    private final Map<K, V> map;
+    private final List<Consumer<Map<K, V>>> changes;
+
+    public MapUpdater(Map<K, V> map) {
+        this.map = map == null ? new HashMap<>() : map;
+        changes = new ArrayList<>(4);
+    }
+
+    @Override
+    public void clear() {
+        changes.clear();
+        addChange(Map::clear);
+    }
+
+    @Override
+    public V get(Object key) {
+        return map.get(key);
+    }
+
+    @Override
+    public V put(K key, V value) {
+        addChange(kvMap -> kvMap.put(key, value));
+        return map.put(key, value);
+    }
+
+    @Override
+    public V remove(Object key) {
+        addChange(kvMap -> kvMap.remove(key));
+        return map.remove(key);
+    }
+
+    @SuppressWarnings("NullableProblems")
+    @Override
+    public Set<Entry<K, V>> entrySet() {
+        return map.entrySet();
+    }
+
+    @Override
+    public boolean containsKey(Object key) {
+        return map.containsKey(key);
+    }
+
+    private void addChange(Consumer<Map<K, V>> change) {
+        changes.add(change);
+    }
+
+    /**
+     * Apply the changes tracked into the {@code other} map.
+     *
+     * @param other The {@link Map} to modify.
+     */
+    public void applyChanges(Map<K, V> other) {
+        changes.forEach(consumer -> consumer.accept(other));
+    }
+
+    /**
+     * @return {@code true} if this {@link Map} was not modified.
+     */
+    public boolean isUnchanged() {
+        return changes.isEmpty();
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/loginfailures/LoginFailuresUpdater.java

@@ -16,8 +16,12 @@
  */
 package org.keycloak.models.sessions.infinispan.changes.remote.updater.loginfailures;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Consumer;
+
 import org.infinispan.client.hotrod.MetadataValue;
-import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.UserLoginFailureModel;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.BaseUpdater;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.Expiration;
@@ -26,11 +30,6 @@ import org.keycloak.models.sessions.infinispan.entities.LoginFailureEntity;
 import org.keycloak.models.sessions.infinispan.entities.LoginFailureKey;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.function.Consumer;
-
 /**
  * Implementation of {@link Updater} and {@link UserLoginFailureModel}.
  * <p>
@@ -42,6 +41,11 @@ public class LoginFailuresUpdater extends BaseUpdater<LoginFailureKey, LoginFail
 
     private LoginFailuresUpdater(LoginFailureKey key, LoginFailureEntity entity, long version, UpdaterState initialState) {
         super(key, entity, version, initialState);
+        if (entity == null) {
+            assert initialState == UpdaterState.DELETED;
+            changes = List.of();
+            return;
+        }
         changes = new ArrayList<>(4);
     }
 
@@ -50,7 +54,7 @@ public class LoginFailuresUpdater extends BaseUpdater<LoginFailureKey, LoginFail
     }
 
     public static LoginFailuresUpdater wrap(LoginFailureKey key, MetadataValue<LoginFailureEntity> entity) {
-        return new LoginFailuresUpdater(Objects.requireNonNull(key), Objects.requireNonNull(entity.getValue()), entity.getVersion(), UpdaterState.READ_ONLY);
+        return new LoginFailuresUpdater(Objects.requireNonNull(key), Objects.requireNonNull(entity.getValue()), entity.getVersion(), UpdaterState.READ);
     }
 
     public static LoginFailuresUpdater delete(LoginFailureKey key) {
@@ -58,11 +62,10 @@ public class LoginFailuresUpdater extends BaseUpdater<LoginFailureKey, LoginFail
     }
 
     @Override
-    public Expiration computeExpiration(KeycloakSession session) {
+    public Expiration computeExpiration() {
-        var realm = session.realms().getRealm(getValue().getRealmId());
         return new Expiration(
-                SessionTimeouts.getLoginFailuresMaxIdleMs(realm, null, getValue()),
+                SessionTimeouts.getLoginFailuresMaxIdleMs(null, null, getValue()),
-                SessionTimeouts.getLoginFailuresLifespanMs(realm, null, getValue()));
+                SessionTimeouts.getLoginFailuresLifespanMs(null, null, getValue()));
     }
 
     @Override
@@ -115,6 +118,7 @@ public class LoginFailuresUpdater extends BaseUpdater<LoginFailureKey, LoginFail
 
     @Override
     public void clearFailures() {
+        changes.clear();
         addAndApplyChange(CLEAR);
     }
 
@@ -143,15 +147,14 @@ public class LoginFailuresUpdater extends BaseUpdater<LoginFailureKey, LoginFail
         addAndApplyChange(e -> e.setLastIPFailure(ip));
     }
 
+    @Override
+    protected boolean isUnchanged() {
+        return changes.isEmpty();
+    }
+
     private void addAndApplyChange(Consumer<LoginFailureEntity> change) {
-        if (change == CLEAR) {
+        changes.add(change);
-            changes.clear();
-            changes.add(CLEAR);
-        } else {
-            changes.add(change);
-        }
         change.accept(getValue());
-        onFieldChanged();
     }
 
     private static final Consumer<LoginFailureEntity> CLEAR = LoginFailureEntity::clearFailures;

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/user/ClientSessionMappingAdapter.java

@@ -0,0 +1,144 @@
+/*
+ * 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.
+ */
+
+package org.keycloak.models.sessions.infinispan.changes.remote.updater.user;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.commons.util.concurrent.CompletionStages;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionStore;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+
+/**
+ * This class adapts and converts the {@link UserSessionEntity#getAuthenticatedClientSessions()} into
+ * {@link UserSessionModel#getAuthenticatedClientSessions()}.
+ * <p>
+ * Its implementation optimizes methods {@link #clear()}, {@link #put(String, AuthenticatedClientSessionModel)},
+ * {@link #get(Object)} and {@link #remove(Object)} by avoiding download all client sessions from the
+ * {@link RemoteCache}.
+ * <p>
+ * The remaining methods are more expensive and require downloading all client sessions. The requests are done in
+ * concurrently to reduce the overall response time.
+ * <p>
+ * This class keeps track of any modification required in {@link UserSessionEntity#getAuthenticatedClientSessions()} and
+ * those modification can be replayed.
+ */
+public class ClientSessionMappingAdapter extends AbstractMap<String, AuthenticatedClientSessionModel> {
+
+    private static final Consumer<AuthenticatedClientSessionStore> CLEAR = AuthenticatedClientSessionStore::clear;
+
+    private final AuthenticatedClientSessionStore mappings;
+    private final ClientSessionProvider clientSessionProvider;
+    private final List<Consumer<AuthenticatedClientSessionStore>> changes;
+
+    public ClientSessionMappingAdapter(AuthenticatedClientSessionStore mappings, ClientSessionProvider clientSessionProvider) {
+        this.mappings = Objects.requireNonNull(mappings);
+        this.clientSessionProvider = Objects.requireNonNull(clientSessionProvider);
+        changes = new CopyOnWriteArrayList<>();
+    }
+
+    @Override
+    public void clear() {
+        mappings.forEach((id, uuid) -> clientSessionProvider.removeClientSession(uuid));
+        changes.clear();
+        addChangeAndApply(CLEAR);
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel put(String key, AuthenticatedClientSessionModel value) {
+        addChangeAndApply(store -> store.put(key, UUID.fromString(value.getId())));
+        return clientSessionProvider.getClientSession(key, mappings.get(key));
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel remove(Object key) {
+        var clientId = String.valueOf(key);
+        var uuid = mappings.get(clientId);
+        var existing = clientSessionProvider.getClientSession(clientId, uuid);
+        onClientRemoved(clientId, uuid);
+        return existing;
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel get(Object key) {
+        var clientId = String.valueOf(key);
+        return clientSessionProvider.getClientSession(clientId, mappings.get(clientId));
+    }
+
+    @SuppressWarnings("NullableProblems")
+    @Override
+    public Set<Entry<String, AuthenticatedClientSessionModel>> entrySet() {
+        Map<String, AuthenticatedClientSessionModel> results = new ConcurrentHashMap<>(mappings.size());
+        var stage = CompletionStages.aggregateCompletionStage();
+        mappings.forEach((clientId, uuid) -> stage.dependsOn(clientSessionProvider.getClientSessionAsync(clientId, uuid)
+                .thenAccept(updater -> {
+                    if (updater == null) {
+                        onClientRemoved(clientId, uuid);
+                        return;
+                    }
+                    results.put(clientId, updater);
+                })));
+        CompletionStages.join(stage.freeze());
+        return results.entrySet();
+    }
+
+    boolean isUnchanged() {
+        return changes.isEmpty();
+    }
+
+    void removeAll(Collection<String> removedClientUUIDS) {
+        if (removedClientUUIDS == null || removedClientUUIDS.isEmpty()) {
+            return;
+        }
+        removedClientUUIDS.forEach(this::onClientRemoved);
+    }
+
+    /**
+     * Applies the modifications recorded by this class into a different {@link AuthenticatedClientSessionStore}.
+     *
+     * @param store The {@link AuthenticatedClientSessionStore} to update.
+     */
+    void applyChanges(AuthenticatedClientSessionStore store) {
+        changes.forEach(change -> change.accept(store));
+    }
+
+    private void addChangeAndApply(Consumer<AuthenticatedClientSessionStore> change) {
+        change.accept(mappings);
+        changes.add(change);
+    }
+
+    private void onClientRemoved(String clientId) {
+        onClientRemoved(clientId, mappings.get(clientId));
+    }
+
+    private void onClientRemoved(String clientId, UUID key) {
+        addChangeAndApply(store -> store.remove(clientId));
+        clientSessionProvider.removeClientSession(key);
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/user/ClientSessionProvider.java

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+
+package org.keycloak.models.sessions.infinispan.changes.remote.updater.user;
+
+import java.util.UUID;
+import java.util.concurrent.CompletionStage;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+
+/**
+ * An SPI for {@link ClientSessionMappingAdapter} to interact with the {@link RemoteCache}.
+ */
+public interface ClientSessionProvider {
+
+    /**
+     * Synchronously fetch an {@link AuthenticatedClientSessionModel} from the {@link RemoteCache}.
+     *
+     * @param clientId        The client's ID.
+     * @param clientSessionId The {@link RemoteCache} key.
+     * @return The {@link AuthenticatedClientSessionModel} instance or {@code null} if the client session does not exist
+     * or was removed.
+     */
+    AuthenticatedClientSessionModel getClientSession(String clientId, UUID clientSessionId);
+
+    /**
+     * A non-blocking alternative to {@link #getClientSession(String, UUID)}.
+     *
+     * @see #getClientSession(String, UUID)
+     */
+    CompletionStage<AuthenticatedClientSessionModel> getClientSessionAsync(String clientId, UUID clientSessionId);
+
+    /**
+     * Removes the client session associated with the {@link RemoteCache} key.
+     * <p>
+     * If {@code clientSessionId} is {@code null}, nothing is removed. The methods
+     * {@link #getClientSession(String, UUID)} and {@link #getClientSessionAsync(String, UUID)} will return {@code null}
+     * for the session after this method is completed.
+     *
+     * @param clientSessionId The {@link RemoteCache} key to remove.
+     */
+    void removeClientSession(UUID clientSessionId);
+
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/updater/user/UserSessionUpdater.java

@@ -0,0 +1,296 @@
+package org.keycloak.models.sessions.infinispan.changes.remote.updater.user;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+import org.infinispan.client.hotrod.MetadataValue;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.BaseUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.Expiration;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.Updater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.UpdaterFactory;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.helper.MapUpdater;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionStore;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
+
+/**
+ * The {@link Updater} implementation to keep track of modifications for {@link UserSessionModel}.
+ */
+public class UserSessionUpdater extends BaseUpdater<String, UserSessionEntity> implements UserSessionModel {
+
+    private static final Factory ONLINE = new Factory(false);
+    private static final Factory OFFLINE = new Factory(true);
+
+    private final MapUpdater<String, String> notesUpdater;
+    private final List<Consumer<UserSessionEntity>> changes;
+    private final boolean offline;
+    private RealmModel realm;
+    private UserModel user;
+    private ClientSessionMappingAdapter clientSessionMappingAdapter;
+    private SessionPersistenceState persistenceState = SessionPersistenceState.PERSISTENT;
+
+    private UserSessionUpdater(String cacheKey, UserSessionEntity cacheValue, long version, boolean offline, UpdaterState initialState) {
+        super(cacheKey, cacheValue, version, initialState);
+        this.offline = offline;
+        if (cacheValue == null) {
+            assert initialState == UpdaterState.DELETED;
+            // cannot undelete
+            changes = List.of();
+            notesUpdater = null;
+            return;
+        }
+        initNotes(cacheValue);
+        notesUpdater = new MapUpdater<>(cacheValue.getNotes());
+        changes = new ArrayList<>(4);
+    }
+
+    /**
+     * @param offline If {@code true}, it creates offline {@link UserSessionModel}.
+     * @return The {@link UpdaterFactory} implementation to create instances of {@link UserSessionModel}.
+     */
+    public static UpdaterFactory<String, UserSessionEntity, UserSessionUpdater> factory(boolean offline) {
+        return offline ? OFFLINE : ONLINE;
+    }
+
+    @Override
+    public UserSessionEntity apply(String ignored, UserSessionEntity userSessionEntity) {
+        initNotes(userSessionEntity);
+        initStore(userSessionEntity);
+        changes.forEach(change -> change.accept(userSessionEntity));
+        notesUpdater.applyChanges(userSessionEntity.getNotes());
+        clientSessionMappingAdapter.applyChanges(userSessionEntity.getAuthenticatedClientSessions());
+        return userSessionEntity;
+    }
+
+    @Override
+    public Expiration computeExpiration() {
+        long maxIdle;
+        long lifespan;
+        if (offline) {
+            maxIdle = SessionTimeouts.getOfflineSessionMaxIdleMs(realm, null, getValue());
+            lifespan = SessionTimeouts.getOfflineSessionLifespanMs(realm, null, getValue());
+        } else {
+            maxIdle = SessionTimeouts.getUserSessionMaxIdleMs(realm, null, getValue());
+            lifespan = SessionTimeouts.getUserSessionLifespanMs(realm, null, getValue());
+        }
+        return new Expiration(maxIdle, lifespan);
+    }
+
+    @Override
+    public String getId() {
+        return getValue().getId();
+    }
+
+    @Override
+    public RealmModel getRealm() {
+        return realm;
+    }
+
+    @Override
+    public String getBrokerSessionId() {
+        return getValue().getBrokerSessionId();
+    }
+
+    @Override
+    public String getBrokerUserId() {
+        return getValue().getBrokerUserId();
+    }
+
+    @Override
+    public UserModel getUser() {
+        return user;
+    }
+
+    @Override
+    public String getLoginUsername() {
+        return getValue().getLoginUsername();
+    }
+
+    @Override
+    public String getIpAddress() {
+        return getValue().getIpAddress();
+    }
+
+    @Override
+    public String getAuthMethod() {
+        return getValue().getAuthMethod();
+    }
+
+    @Override
+    public boolean isRememberMe() {
+        return getValue().isRememberMe();
+    }
+
+    @Override
+    public int getStarted() {
+        return getValue().getStarted();
+    }
+
+    @Override
+    public int getLastSessionRefresh() {
+        return getValue().getLastSessionRefresh();
+    }
+
+    @Override
+    public void setLastSessionRefresh(int seconds) {
+        addAndApplyChange(userSessionEntity -> userSessionEntity.setLastSessionRefresh(seconds));
+    }
+
+    @Override
+    public boolean isOffline() {
+        return offline;
+    }
+
+    @Override
+    public Map<String, AuthenticatedClientSessionModel> getAuthenticatedClientSessions() {
+        return clientSessionMappingAdapter;
+    }
+
+    @Override
+    public void removeAuthenticatedClientSessions(Collection<String> removedClientUUIDS) {
+        clientSessionMappingAdapter.removeAll(removedClientUUIDS);
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel getAuthenticatedClientSessionByClient(String clientUUID) {
+        return clientSessionMappingAdapter.get(clientUUID);
+    }
+
+    @Override
+    public String getNote(String name) {
+        return notesUpdater.get(name);
+    }
+
+    @Override
+    public void setNote(String name, String value) {
+        if (value == null) {
+            removeNote(name);
+        } else {
+            notesUpdater.put(name, value);
+        }
+    }
+
+    @Override
+    public void removeNote(String name) {
+        notesUpdater.remove(name);
+    }
+
+    @Override
+    public Map<String, String> getNotes() {
+        return notesUpdater;
+    }
+
+    @Override
+    public State getState() {
+        return getValue().getState();
+    }
+
+    @Override
+    public void setState(State state) {
+        addAndApplyChange(userSessionEntity -> userSessionEntity.setState(state));
+    }
+
+    @Override
+    public void restartSession(RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId) {
+        this.realm = realm;
+        this.user = user;
+        changes.clear();
+        notesUpdater.clear();
+        clientSessionMappingAdapter.clear();
+        addAndApplyChange(userSessionEntity -> UserSessionEntity.updateSessionEntity(userSessionEntity, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId));
+    }
+
+    @Override
+    public SessionPersistenceState getPersistenceState() {
+        return persistenceState;
+    }
+
+    @Override
+    public boolean isTransient() {
+        return !isDeleted() && persistenceState == SessionPersistenceState.TRANSIENT;
+    }
+
+    @Override
+    protected boolean isUnchanged() {
+        return changes.isEmpty() && notesUpdater.isUnchanged() && clientSessionMappingAdapter.isUnchanged();
+    }
+
+    /**
+     * Initializes this class with references to other models classes.
+     *
+     * @param persistenceState The {@link SessionPersistenceState}.
+     * @param realm            The {@link RealmModel} to where this user session belongs.
+     * @param user             The {@link UserModel} associated to this user session.
+     * @param factory          The {@link ClientSessionAdapterFactory} to create the {@link ClientSessionMappingAdapter}
+     *                         to track modifications into the client sessions.
+     */
+    public synchronized void initialize(SessionPersistenceState persistenceState, RealmModel realm, UserModel user, ClientSessionAdapterFactory factory) {
+        initStore(getValue());
+        this.realm = Objects.requireNonNull(realm);
+        this.user = Objects.requireNonNull(user);
+        this.persistenceState = Objects.requireNonNull(persistenceState);
+        clientSessionMappingAdapter = factory.create(getValue().getAuthenticatedClientSessions());
+    }
+
+    /**
+     * @return {@code true} if it is already initialized.
+     */
+    public synchronized boolean isInitialized() {
+        return realm != null;
+    }
+
+    private void addAndApplyChange(Consumer<UserSessionEntity> change) {
+        change.accept(getValue());
+        changes.add(change);
+    }
+
+    private static void initNotes(UserSessionEntity entity) {
+        var notes = entity.getNotes();
+        if (notes == null) {
+            entity.setNotes(new HashMap<>());
+        }
+    }
+
+    private static void initStore(UserSessionEntity entity) {
+        var store = entity.getAuthenticatedClientSessions();
+        if (store == null) {
+            entity.setAuthenticatedClientSessions(new AuthenticatedClientSessionStore());
+        }
+    }
+
+    /**
+     * Factory SPI to create {@link ClientSessionMappingAdapter} for the {@link AuthenticatedClientSessionStore} used by
+     * this instance.
+     */
+    public interface ClientSessionAdapterFactory {
+        ClientSessionMappingAdapter create(AuthenticatedClientSessionStore clientSessionStore);
+    }
+
+    private record Factory(boolean offline) implements UpdaterFactory<String, UserSessionEntity, UserSessionUpdater> {
+
+        @Override
+        public UserSessionUpdater create(String key, UserSessionEntity entity) {
+            return new UserSessionUpdater(key, Objects.requireNonNull(entity), -1, offline, UpdaterState.CREATED);
+        }
+
+        @Override
+        public UserSessionUpdater wrapFromCache(String key, MetadataValue<UserSessionEntity> entity) {
+            assert entity != null;
+            return new UserSessionUpdater(key, Objects.requireNonNull(entity.getValue()), entity.getVersion(), offline, UpdaterState.READ);
+        }
+
+        @Override
+        public UserSessionUpdater deleted(String key) {
+            return new UserSessionUpdater(key, null, -1, offline, UpdaterState.DELETED);
+        }
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/entities/AuthenticatedClientSessionEntity.java

@@ -17,19 +17,23 @@
 
 package org.keycloak.models.sessions.infinispan.entities;
 
-import org.infinispan.protostream.annotations.ProtoFactory;
-import org.infinispan.protostream.annotations.ProtoField;
-import org.infinispan.protostream.annotations.ProtoTypeId;
-import org.jboss.logging.Logger;
-import org.keycloak.models.AuthenticatedClientSessionModel;
-import org.keycloak.marshalling.Marshalling;
-import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
-
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.infinispan.protostream.annotations.ProtoFactory;
+import org.infinispan.protostream.annotations.ProtoField;
+import org.infinispan.protostream.annotations.ProtoTypeId;
+import org.jboss.logging.Logger;
+import org.keycloak.common.util.Time;
+import org.keycloak.marshalling.Marshalling;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
+
 /**
  *
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
@@ -196,4 +200,24 @@ public class AuthenticatedClientSessionEntity extends SessionEntity {
     public void setUserSessionId(String userSessionId) {
         this.userSessionId = userSessionId;
     }
+
+    public static AuthenticatedClientSessionEntity create(UUID clientSessionId, RealmModel realm, ClientModel client, UserSessionModel userSession) {
+        var entity = new AuthenticatedClientSessionEntity(clientSessionId);
+        entity.setRealmId(realm.getId());
+        entity.setClientId(client.getId());
+        entity.setTimestamp(Time.currentTime());
+        entity.getNotes().put(AuthenticatedClientSessionModel.STARTED_AT_NOTE, String.valueOf(entity.getTimestamp()));
+        entity.getNotes().put(AuthenticatedClientSessionModel.USER_SESSION_STARTED_AT_NOTE, String.valueOf(userSession.getStarted()));
+        if (userSession.isRememberMe()) {
+            entity.getNotes().put(AuthenticatedClientSessionModel.USER_SESSION_REMEMBER_ME_NOTE, "true");
+        }
+        return entity;
+    }
+
+    public static AuthenticatedClientSessionEntity createFromModel(AuthenticatedClientSessionModel model) {
+        var entity =  create(UUID.fromString(model.getId()), model.getRealm(), model.getClient(), model.getUserSession());
+        entity.setNotes(model.getNotes() == null ? new ConcurrentHashMap<>() : model.getNotes());
+        return entity;
+    }
+
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/entities/UserSessionEntity.java

@@ -26,7 +26,11 @@ import org.infinispan.protostream.annotations.ProtoFactory;
 import org.infinispan.protostream.annotations.ProtoField;
 import org.infinispan.protostream.annotations.ProtoTypeId;
 import org.jboss.logging.Logger;
+import org.keycloak.common.util.Time;
 import org.keycloak.marshalling.Marshalling;
+import org.keycloak.models.OfflineUserSessionModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserModel;
 import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 
@@ -248,4 +252,54 @@ public class UserSessionEntity extends SessionEntity {
         return entityWrapper;
     }
 
+    public static UserSessionEntity create(String id, RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId) {
+        UserSessionEntity entity = new UserSessionEntity(id);
+        updateSessionEntity(entity, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
+        return entity;
+    }
+
+    public static void updateSessionEntity(UserSessionEntity entity, RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId) {
+        entity.setRealmId(realm.getId());
+        entity.setUser(user.getId());
+        entity.setLoginUsername(loginUsername);
+        entity.setIpAddress(ipAddress);
+        entity.setAuthMethod(authMethod);
+        entity.setRememberMe(rememberMe);
+        entity.setBrokerSessionId(brokerSessionId);
+        entity.setBrokerUserId(brokerUserId);
+
+        int currentTime = Time.currentTime();
+
+        entity.setStarted(currentTime);
+        entity.setLastSessionRefresh(currentTime);
+    }
+
+    public static UserSessionEntity createFromModel(UserSessionModel userSession) {
+        UserSessionEntity entity = new UserSessionEntity(userSession.getId());
+        entity.setRealmId(userSession.getRealm().getId());
+
+        entity.setAuthMethod(userSession.getAuthMethod());
+        entity.setBrokerSessionId(userSession.getBrokerSessionId());
+        entity.setBrokerUserId(userSession.getBrokerUserId());
+        entity.setIpAddress(userSession.getIpAddress());
+        entity.setNotes(userSession.getNotes() == null ? new ConcurrentHashMap<>() : userSession.getNotes());
+        entity.setAuthenticatedClientSessions(new AuthenticatedClientSessionStore());
+        entity.setRememberMe(userSession.isRememberMe());
+        entity.setState(userSession.getState());
+        if (userSession instanceof OfflineUserSessionModel offline) {
+            // this is a hack so that UserModel doesn't have to be available when offline token is imported.
+            // see related JIRA - KEYCLOAK-5350 and corresponding test
+            entity.setUser(offline.getUserId());
+            // NOTE: Hack
+            // We skip calling entity.setLoginUsername(userSession.getLoginUsername())
+        } else {
+            entity.setLoginUsername(userSession.getLoginUsername());
+            entity.setUser(userSession.getUser().getId());
+        }
+
+        entity.setStarted(userSession.getStarted());
+        entity.setLastSessionRefresh(userSession.getLastSessionRefresh());
+
+        return entity;
+    }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteUserLoginFailureProviderFactory.java

@@ -46,7 +46,7 @@ public class RemoteUserLoginFailureProviderFactory implements UserLoginFailurePr
 
     @Override
     public RemoteUserLoginFailureProvider create(KeycloakSession session) {
-        var tx = new RemoteChangeLogTransaction<>(this, cache, session);
+        var tx = new RemoteChangeLogTransaction<>(this, cache);
         session.getTransactionManager().enlistAfterCompletion(tx);
         return new RemoteUserLoginFailureProvider(tx);
     }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteUserSessionProvider.java

@@ -0,0 +1,576 @@
+/*
+ * 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.
+ */
+
+package org.keycloak.models.sessions.infinispan.remote;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import io.reactivex.rxjava3.core.Flowable;
+import org.infinispan.client.hotrod.MetadataValue;
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.commons.util.concurrent.CompletableFutures;
+import org.infinispan.commons.util.concurrent.CompletionStages;
+import org.jboss.logging.Logger;
+import org.keycloak.cluster.ClusterProvider;
+import org.keycloak.common.Profile;
+import org.keycloak.common.util.Time;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakSessionFactory;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.UserSessionProvider;
+import org.keycloak.models.light.LightweightUserAdapter;
+import org.keycloak.models.session.UserSessionPersisterProvider;
+import org.keycloak.models.sessions.infinispan.changes.remote.RemoteChangeLogTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.UserSessionTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.BaseUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.client.AuthenticatedClientSessionUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.ClientSessionMappingAdapter;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.ClientSessionProvider;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.UserSessionUpdater;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionStore;
+import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+import org.keycloak.models.utils.KeycloakModelUtils;
+import org.keycloak.utils.StreamsUtil;
+
+import static org.keycloak.models.Constants.SESSION_NOTE_LIGHTWEIGHT_USER;
+
+/**
+ * An {@link UserSessionProvider} implementation that uses only {@link RemoteCache} as storage.
+ */
+public class RemoteUserSessionProvider implements UserSessionProvider {
+
+    private static final Logger log = Logger.getLogger(MethodHandles.lookup().lookupClass());
+
+    private final KeycloakSession session;
+    private final UserSessionTransaction transaction;
+    private final int batchSize;
+
+    public RemoteUserSessionProvider(KeycloakSession session, UserSessionTransaction transaction, int batchSize) {
+        this.session = session;
+        this.transaction = transaction;
+        this.batchSize = batchSize;
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel createClientSession(RealmModel realm, ClientModel client, UserSessionModel userSession) {
+        var transaction = getClientSessionTransaction(false);
+        var clientSessionId = UUID.randomUUID();
+        var entity = AuthenticatedClientSessionEntity.create(clientSessionId, realm, client, userSession);
+        var model = transaction.create(clientSessionId, entity);
+        if (!model.isInitialized()) {
+            model.initialize(userSession, client, transaction);
+        }
+        userSession.getAuthenticatedClientSessions().put(client.getId(), model);
+        return model;
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel getClientSession(UserSessionModel userSession, ClientModel client, String clientSessionId, boolean offline) {
+        if (clientSessionId == null) {
+            return null;
+        }
+        var transaction = getClientSessionTransaction(offline);
+        var updater = transaction.get(UUID.fromString(clientSessionId));
+        if (updater == null) {
+            return null;
+        }
+        if (!updater.isInitialized()) {
+            updater.initialize(userSession, client, transaction);
+        }
+        return updater;
+    }
+
+    @Override
+    public UserSessionModel createUserSession(String id, RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId, UserSessionModel.SessionPersistenceState persistenceState) {
+        if (id == null) {
+            id = KeycloakModelUtils.generateId();
+        }
+
+        var entity = UserSessionEntity.create(id, realm, user, loginUsername, ipAddress, authMethod, rememberMe, brokerSessionId, brokerUserId);
+        var updater = transaction.getUserSessions().create(id, entity);
+        return initUserSessionUpdater(updater, persistenceState, realm, user, false);
+    }
+
+    @Override
+    public UserSessionModel getUserSession(RealmModel realm, String id) {
+        return getUserSession(realm, id, false);
+    }
+
+    @Override
+    public Stream<UserSessionModel> getUserSessionsStream(RealmModel realm, UserModel user) {
+        return StreamsUtil.closing(streamUserSessions(new UserAndRealmPredicate(realm.getId(), user.getId()), realm, user, false));
+    }
+
+    @Override
+    public Stream<UserSessionModel> getUserSessionsStream(RealmModel realm, ClientModel client) {
+        return StreamsUtil.closing(streamUserSessions(new ClientAndRealmPredicate(realm.getId(), client.getId()), realm, null, false));
+    }
+
+    @Override
+    public Stream<UserSessionModel> getUserSessionsStream(RealmModel realm, ClientModel client, Integer firstResult, Integer maxResults) {
+        return StreamsUtil.paginatedStream(getUserSessionsStream(realm, client).sorted(Comparator.comparing(UserSessionModel::getLastSessionRefresh)), firstResult, maxResults);
+    }
+
+    @Override
+    public Stream<UserSessionModel> getUserSessionByBrokerUserIdStream(RealmModel realm, String brokerUserId) {
+        return StreamsUtil.closing(streamUserSessions(new BrokerUserIdAndRealmPredicate(realm.getId(), brokerUserId), realm, null, false));
+    }
+
+    @Override
+    public UserSessionModel getUserSessionByBrokerSessionId(RealmModel realm, String brokerSessionId) {
+        return StreamsUtil.closing(streamUserSessions(new BrokerSessionIdAndRealmPredicate(realm.getId(), brokerSessionId), realm, null, false))
+                .findFirst()
+                .orElse(null);
+    }
+
+    @Override
+    public UserSessionModel getUserSessionWithPredicate(RealmModel realm, String id, boolean offline, Predicate<UserSessionModel> predicate) {
+        var updater = getUserSession(realm, id, offline);
+        return updater != null && predicate.test(updater) ? updater : null;
+    }
+
+    @Override
+    public long getActiveUserSessions(RealmModel realm, ClientModel client) {
+        return StreamsUtil.closing(getUserSessionsStream(realm, client)).count();
+    }
+
+    @Override
+    public Map<String, Long> getActiveClientSessionStats(RealmModel realm, boolean offline) {
+        var userSessions = getUserSessionTransaction(offline);
+        return Flowable.fromPublisher(userSessions.getCache().publishEntriesWithMetadata(null, batchSize))
+                .filter(new RealmPredicate(realm.getId()))
+                .map(Map.Entry::getValue)
+                .map(MetadataValue::getValue)
+                .map(UserSessionEntity::getAuthenticatedClientSessions)
+                .map(AuthenticatedClientSessionStore::keySet)
+                .map(Collection::stream)
+                .flatMap(Flowable::fromStream)
+                .collect(Collectors.groupingBy(Function.identity(), Collectors.counting()))
+                .blockingGet();
+    }
+
+    @Override
+    public void removeUserSession(RealmModel realm, UserSessionModel userSession) {
+        internalRemoveUserSession(userSession, false);
+    }
+
+    @Override
+    public void removeUserSessions(RealmModel realm, UserModel user) {
+        getUserSessionsStream(realm, user).forEach(s -> removeUserSession(realm, s));
+    }
+
+    @Override
+    public void removeAllExpired() {
+        //rely on Infinispan expiration
+    }
+
+    @Override
+    public void removeExpired(RealmModel realm) {
+        //rely on Infinispan expiration
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public void removeUserSessions(RealmModel realm) {
+        Predicate<? extends SessionEntity> predicate = e -> Objects.equals(e.getRealmId(), realm.getId());
+        transaction.getUserSessions().removeIf((Predicate<UserSessionEntity>) predicate);
+        transaction.getClientSessions().removeIf((Predicate<AuthenticatedClientSessionEntity>) predicate);
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public void onRealmRemoved(RealmModel realm) {
+        Predicate<? extends SessionEntity> predicate = e -> Objects.equals(e.getRealmId(), realm.getId());
+        transaction.getUserSessions().removeIf((Predicate<UserSessionEntity>) predicate);
+        transaction.getOfflineUserSessions().removeIf((Predicate<UserSessionEntity>) predicate);
+        transaction.getClientSessions().removeIf((Predicate<AuthenticatedClientSessionEntity>) predicate);
+        transaction.getOfflineClientSessions().removeIf((Predicate<AuthenticatedClientSessionEntity>) predicate);
+        var database = session.getProvider(UserSessionPersisterProvider.class);
+        if (database != null) {
+            database.onRealmRemoved(realm);
+        }
+    }
+
+    @Override
+    public void onClientRemoved(RealmModel realm, ClientModel client) {
+        var database = session.getProvider(UserSessionPersisterProvider.class);
+        if (database != null) {
+            database.onClientRemoved(realm, client);
+        }
+    }
+
+    @Override
+    public UserSessionModel createOfflineUserSession(UserSessionModel userSession) {
+        var entity = UserSessionEntity.createFromModel(userSession);
+
+        int currentTime = Time.currentTime();
+        entity.setStarted(currentTime);
+        entity.setLastSessionRefresh(currentTime);
+
+        var updater = getUserSessionTransaction(true).create(entity.getId(), entity);
+        return initUserSessionUpdater(updater, userSession.getPersistenceState(), userSession.getRealm(), userSession.getUser(), true);
+    }
+
+    @Override
+    public UserSessionModel getOfflineUserSession(RealmModel realm, String userSessionId) {
+        return getUserSession(realm, userSessionId, true);
+    }
+
+    @Override
+    public void removeOfflineUserSession(RealmModel realm, UserSessionModel userSession) {
+        internalRemoveUserSession(userSession, true);
+    }
+
+    @Override
+    public AuthenticatedClientSessionModel createOfflineClientSession(AuthenticatedClientSessionModel clientSession, UserSessionModel offlineUserSession) {
+        var transaction = getClientSessionTransaction(true);
+        var entity = AuthenticatedClientSessionEntity.createFromModel(clientSession);
+        var model = transaction.create(entity.getId(), entity);
+        if (!model.isInitialized()) {
+            model.initialize(offlineUserSession, clientSession.getClient(), transaction);
+        }
+        offlineUserSession.getAuthenticatedClientSessions().put(clientSession.getClient().getId(), model);
+        return model;
+    }
+
+    @Override
+    public Stream<UserSessionModel> getOfflineUserSessionsStream(RealmModel realm, UserModel user) {
+        return StreamsUtil.closing(streamUserSessions(new UserAndRealmPredicate(realm.getId(), user.getId()), realm, user, true));
+    }
+
+    @Override
+    public Stream<UserSessionModel> getOfflineUserSessionByBrokerUserIdStream(RealmModel realm, String brokerUserId) {
+        return StreamsUtil.closing(streamUserSessions(new BrokerUserIdAndRealmPredicate(realm.getId(), brokerUserId), realm, null, true));
+    }
+
+    @Override
+    public long getOfflineSessionsCount(RealmModel realm, ClientModel client) {
+        return StreamsUtil.closing(streamUserSessions(new ClientAndRealmPredicate(realm.getId(), client.getId()), realm, null, true)).count();
+    }
+
+    @Override
+    public Stream<UserSessionModel> getOfflineUserSessionsStream(RealmModel realm, ClientModel client, Integer firstResult, Integer maxResults) {
+        return StreamsUtil.closing(StreamsUtil.paginatedStream(streamUserSessions(new ClientAndRealmPredicate(realm.getId(), client.getId()), realm, null, true), firstResult, maxResults));
+    }
+
+    @Override
+    public int getStartupTime(RealmModel realm) {
+        return session.getProvider(ClusterProvider.class).getClusterStartupTime();
+    }
+
+    @Override
+    public KeycloakSession getKeycloakSession() {
+        return session;
+    }
+
+    @Override
+    public void importUserSessions(Collection<UserSessionModel> persistentUserSessions, boolean offline) {
+        //no-op
+    }
+
+    @Override
+    public void close() {
+
+    }
+
+    @Override
+    public void migrate(String modelVersion) {
+        if ("25.0.0".equals(modelVersion)) {
+            migrateUserSessions(true);
+            migrateUserSessions(false);
+        }
+
+    }
+
+    private void migrateUserSessions(boolean offline) {
+        log.info("Migrate user sessions from database to the remote cache");
+
+        List<String> userSessionIds = Collections.synchronizedList(new ArrayList<>(batchSize));
+        List<Map.Entry<String, String>> clientSessionIds = Collections.synchronizedList(new ArrayList<>(batchSize));
+        boolean hasSessions;
+        do {
+            hasSessions = migrateUserSessionBatch(session.getKeycloakSessionFactory(), offline, userSessionIds, clientSessionIds);
+        } while (hasSessions);
+
+        log.info("All sessions migrated.");
+    }
+
+    private boolean migrateUserSessionBatch(KeycloakSessionFactory factory, boolean offline, List<String> userSessionBuffer, List<Map.Entry<String, String>> clientSessionBuffer) {
+        var userSessionCache = getUserSessionTransaction(offline).getCache();
+        var clientSessionCache = getClientSessionTransaction(offline).getCache();
+
+        log.infof("Migrating %s user(s) session(s) from database.", batchSize);
+
+        return KeycloakModelUtils.runJobInTransactionWithResult(factory, kcSession -> {
+            var database = kcSession.getProvider(UserSessionPersisterProvider.class);
+            var stage = CompletionStages.aggregateCompletionStage();
+            database.loadUserSessionsStream(-1, batchSize, offline, "")
+                    .forEach(userSessionModel -> {
+                        var userSessionEntity = UserSessionEntity.createFromModel(userSessionModel);
+                        stage.dependsOn(userSessionCache.putIfAbsentAsync(userSessionModel.getId(), userSessionEntity));
+                        userSessionBuffer.add(userSessionModel.getId());
+                        for (var clientSessionModel : userSessionModel.getAuthenticatedClientSessions().values()) {
+                            clientSessionBuffer.add(Map.entry(userSessionModel.getId(), clientSessionModel.getId()));
+                            var clientSessionEntity = AuthenticatedClientSessionEntity.createFromModel(clientSessionModel);
+                            stage.dependsOn(clientSessionCache.putIfAbsentAsync(clientSessionEntity.getId(), clientSessionEntity));
+                        }
+                    });
+            CompletionStages.join(stage.freeze());
+
+            if (userSessionBuffer.isEmpty() && clientSessionBuffer.isEmpty()) {
+                return false;
+            }
+
+            log.infof("%s user(s) session(s) stored in the remote cache. Removing them from database.", userSessionBuffer.size());
+
+            userSessionBuffer.forEach(s -> database.removeUserSession(s, offline));
+            userSessionBuffer.clear();
+
+            clientSessionBuffer.forEach(e -> database.removeClientSession(e.getKey(), e.getValue(), offline));
+            clientSessionBuffer.clear();
+
+            return true;
+        });
+    }
+
+    private UserSessionUpdater getUserSession(RealmModel realm, String id, boolean offline) {
+        if (id == null) {
+            return null;
+        }
+        var updater = getUserSessionTransaction(offline).get(id);
+        if (updater == null || !updater.getValue().getRealmId().equals(realm.getId())) {
+            return null;
+        }
+        if (updater.isInitialized()) {
+            return updater;
+        }
+        UserModel user = session.users().getUserById(realm, updater.getValue().getUser());
+        return initUserSessionUpdater(updater, UserSessionModel.SessionPersistenceState.PERSISTENT, realm, user, offline);
+    }
+
+    private void internalRemoveUserSession(UserSessionModel userSession, boolean offline) {
+        var clientSessionTransaction = getClientSessionTransaction(offline);
+        var userSessionTransaction = getUserSessionTransaction(offline);
+        userSession.getAuthenticatedClientSessions().values()
+                .stream()
+                .filter(Objects::nonNull) // we need to filter, it may not be a UserSessionUpdater class.
+                .map(AuthenticatedClientSessionModel::getId)
+                .filter(Objects::nonNull) // we need to filter, it may not be a AuthenticatedClientSessionUpdater class.
+                .map(UUID::fromString)
+                .forEach(clientSessionTransaction::remove);
+        userSessionTransaction.remove(userSession.getId());
+    }
+
+    private Stream<UserSessionModel> streamUserSessions(InternalUserSessionPredicate predicate, RealmModel realm, UserModel user, boolean offline) {
+        var userSessions = getUserSessionTransaction(offline);
+        return Flowable.fromPublisher(userSessions.getCache().publishEntriesWithMetadata(null, batchSize))
+                .filter(predicate)
+                .map(userSessions::wrap)
+                .map(s -> initFromStream(s, realm, user, offline))
+                .filter(Optional::isPresent)
+                .map(Optional::get)
+                .map(UserSessionModel.class::cast)
+                .blockingStream(batchSize);
+    }
+
+    private RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> getUserSessionTransaction(boolean offline) {
+        return offline ? transaction.getOfflineUserSessions() : transaction.getUserSessions();
+    }
+
+    private RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> getClientSessionTransaction(boolean offline) {
+        return offline ? transaction.getOfflineClientSessions() : transaction.getClientSessions();
+    }
+
+    private Optional<UserSessionUpdater> initFromStream(UserSessionUpdater updater, RealmModel realm, UserModel user, boolean offline) {
+        if (updater.isInitialized()) {
+            return Optional.of(updater);
+        }
+        assert realm != null;
+        if (user == null) {
+            user = session.users().getUserById(realm, updater.getValue().getUser());
+        }
+        return Optional.ofNullable(initUserSessionUpdater(updater, UserSessionModel.SessionPersistenceState.PERSISTENT, realm, user, offline));
+    }
+
+    private UserSessionUpdater initUserSessionUpdater(UserSessionUpdater updater, UserSessionModel.SessionPersistenceState persistenceState, RealmModel realm, UserModel user, boolean offline) {
+        var provider = new RemoteClientSessionAdapterProvider(getClientSessionTransaction(offline), updater);
+        if (user instanceof LightweightUserAdapter) {
+            updater.initialize(persistenceState, realm, user, provider);
+            return checkExpiration(updater);
+        }
+        // copied from org.keycloak.models.sessions.infinispan.InfinispanUserSessionProvider
+        if (Profile.isFeatureEnabled(Profile.Feature.TRANSIENT_USERS) && updater.getNotes().containsKey(SESSION_NOTE_LIGHTWEIGHT_USER)) {
+            LightweightUserAdapter lua = LightweightUserAdapter.fromString(session, realm, updater.getNotes().get(SESSION_NOTE_LIGHTWEIGHT_USER));
+            updater.initialize(persistenceState, realm, lua, provider);
+            lua.setUpdateHandler(lua1 -> {
+                if (lua == lua1) {  // Ensure there is no conflicting user model, only the latest lightweight user can be used
+                    updater.setNote(SESSION_NOTE_LIGHTWEIGHT_USER, lua1.serialize());
+                }
+            });
+            return checkExpiration(updater);
+        }
+
+        if (user == null) {
+            // remove orphaned user session from the cache
+            internalRemoveUserSession(updater, offline);
+            return null;
+        }
+        updater.initialize(persistenceState, realm, user, provider);
+        return checkExpiration(updater);
+    }
+
+    private <K, V, T extends BaseUpdater<K, V>> T checkExpiration(T updater) {
+        var expiration = updater.computeExpiration();
+        if (expiration.isExpired()) {
+            updater.markDeleted();
+            return null;
+        }
+        return updater;
+    }
+
+    private record RealmPredicate(String realmId) implements InternalUserSessionPredicate {
+
+        @Override
+        public boolean testUserSession(UserSessionEntity userSession) {
+            return Objects.equals(userSession.getRealmId(), realmId);
+        }
+    }
+
+    private interface InternalUserSessionPredicate extends io.reactivex.rxjava3.functions.Predicate<Map.Entry<String, MetadataValue<UserSessionEntity>>> {
+
+        @Override
+        default boolean test(Map.Entry<String, MetadataValue<UserSessionEntity>> e) {
+            return testUserSession(e.getValue().getValue());
+        }
+
+        boolean testUserSession(UserSessionEntity userSession);
+    }
+
+    private record UserAndRealmPredicate(String realmId, String userId) implements InternalUserSessionPredicate {
+
+        @Override
+        public boolean testUserSession(UserSessionEntity userSession) {
+            return Objects.equals(userSession.getRealmId(), realmId) && Objects.equals(userSession.getUser(), userId);
+        }
+
+    }
+
+    private record ClientAndRealmPredicate(String realmId, String clientId) implements InternalUserSessionPredicate {
+
+        @Override
+        public boolean testUserSession(UserSessionEntity userSession) {
+            return Objects.equals(userSession.getRealmId(), realmId) && userSession.getAuthenticatedClientSessions().containsKey(clientId);
+        }
+    }
+
+    private record BrokerUserIdAndRealmPredicate(String realmId, String brokerUserId) implements InternalUserSessionPredicate {
+
+        @Override
+        public boolean testUserSession(UserSessionEntity userSession) {
+            return Objects.equals(userSession.getRealmId(), realmId) && Objects.equals(userSession.getBrokerUserId(), brokerUserId);
+        }
+    }
+
+    private record BrokerSessionIdAndRealmPredicate(String realmId,
+                                                    String brokeSessionId) implements InternalUserSessionPredicate {
+
+        @Override
+        public boolean testUserSession(UserSessionEntity userSession) {
+            return Objects.equals(userSession.getRealmId(), realmId) && Objects.equals(userSession.getBrokerSessionId(), brokeSessionId);
+        }
+    }
+
+    private class RemoteClientSessionAdapterProvider implements ClientSessionProvider, UserSessionUpdater.ClientSessionAdapterFactory {
+
+        private final RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> transaction;
+        private final UserSessionUpdater userSession;
+
+        private RemoteClientSessionAdapterProvider(RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> transaction, UserSessionUpdater userSession) {
+            this.transaction = transaction;
+            this.userSession = userSession;
+        }
+
+        @Override
+        public AuthenticatedClientSessionModel getClientSession(String clientId, UUID clientSessionId) {
+            if (clientId == null || clientSessionId == null) {
+                return null;
+            }
+            var client = userSession.getRealm().getClientById(clientId);
+            if (client == null) {
+                return null;
+            }
+            return initialize(client, transaction.get(clientSessionId));
+        }
+
+        @Override
+        public CompletionStage<AuthenticatedClientSessionModel> getClientSessionAsync(String clientId, UUID clientSessionId) {
+            if (clientId == null || clientSessionId == null) {
+                return CompletableFutures.completedNull();
+            }
+            var client = userSession.getRealm().getClientById(clientId);
+            if (client == null) {
+                return CompletableFutures.completedNull();
+            }
+            return transaction.getAsync(clientSessionId).thenApply(updater -> initialize(client, updater));
+        }
+
+        @Override
+        public void removeClientSession(UUID clientSessionId) {
+            if (clientSessionId == null) {
+                return;
+            }
+            transaction.remove(clientSessionId);
+        }
+
+        private AuthenticatedClientSessionModel initialize(ClientModel client, AuthenticatedClientSessionUpdater updater) {
+            if (updater == null) {
+                return null;
+            }
+            if (updater.isInitialized()) {
+                return updater;
+            }
+            updater.initialize(userSession, client, transaction);
+            return checkExpiration(updater);
+        }
+
+        @Override
+        public ClientSessionMappingAdapter create(AuthenticatedClientSessionStore clientSessionStore) {
+            return new ClientSessionMappingAdapter(clientSessionStore, this);
+        }
+    }
+
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteUserSessionProviderFactory.java

@@ -0,0 +1,141 @@
+package org.keycloak.models.sessions.infinispan.remote;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.keycloak.Config;
+import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
+import org.keycloak.infinispan.util.InfinispanUtils;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakSessionFactory;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserSessionProvider;
+import org.keycloak.models.UserSessionProviderFactory;
+import org.keycloak.models.session.UserSessionPersisterProvider;
+import org.keycloak.models.sessions.infinispan.changes.remote.RemoteChangeLogTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.UserSessionTransaction;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.client.AuthenticatedClientSessionUpdater;
+import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.UserSessionUpdater;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+import org.keycloak.provider.EnvironmentDependentProviderFactory;
+import org.keycloak.provider.ProviderConfigProperty;
+import org.keycloak.provider.ProviderConfigurationBuilder;
+import org.keycloak.provider.ProviderEvent;
+import org.keycloak.provider.ProviderEventListener;
+
+public class RemoteUserSessionProviderFactory implements UserSessionProviderFactory<RemoteUserSessionProvider>, EnvironmentDependentProviderFactory, ProviderEventListener {
+
+    // Sessions are close to 1KB of data. Fetch 1MB per batch request (can be configured)
+    private static final int DEFAULT_BATCH_SIZE = 1024;
+    private static final String CONFIG_MAX_BATCH_SIZE = "batchSize";
+
+    private volatile RemoteCacheHolder remoteCacheHolder;
+    private volatile int batchSize = DEFAULT_BATCH_SIZE;
+
+    @Override
+    public RemoteUserSessionProvider create(KeycloakSession session) {
+        var tx = createTransaction(session);
+        session.getTransactionManager().enlistAfterCompletion(tx);
+        return new RemoteUserSessionProvider(session, tx, batchSize);
+    }
+
+    @Override
+    public void init(Config.Scope config) {
+        batchSize = config.getInt(CONFIG_MAX_BATCH_SIZE, DEFAULT_BATCH_SIZE);
+    }
+
+    @Override
+    public void postInit(KeycloakSessionFactory factory) {
+        try (var session = factory.create()) {
+            lazyInit(session);
+        }
+        factory.register(this);
+
+    }
+
+    @Override
+    public void close() {
+        remoteCacheHolder = null;
+    }
+
+    @Override
+    public String getId() {
+        return InfinispanUtils.REMOTE_PROVIDER_ID;
+    }
+
+    @Override
+    public boolean isSupported(Config.Scope config) {
+        return InfinispanUtils.isRemoteInfinispan();
+    }
+
+    @Override
+    public List<ProviderConfigProperty> getConfigMetadata() {
+        ProviderConfigurationBuilder builder = ProviderConfigurationBuilder.create();
+        builder.property()
+                .name(CONFIG_MAX_BATCH_SIZE)
+                .type("int")
+                .helpText("Batch size when streaming session from the remote cache")
+                .defaultValue(DEFAULT_BATCH_SIZE)
+                .add();
+        return builder.build();
+    }
+
+    @Override
+    public void onEvent(ProviderEvent event) {
+        if (event instanceof UserModel.UserRemovedEvent ure) {
+            onUserRemoved(ure);
+        }
+    }
+
+    private void onUserRemoved(UserModel.UserRemovedEvent event) {
+        event.getKeycloakSession().getProvider(UserSessionProvider.class, getId()).removeUserSessions(event.getRealm(), event.getUser());
+        event.getKeycloakSession().getProvider(UserSessionPersisterProvider.class).onUserRemoved(event.getRealm(), event.getUser());
+    }
+
+    private void lazyInit(KeycloakSession session) {
+        if (remoteCacheHolder != null) {
+            return;
+        }
+        InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
+        RemoteCache<String, UserSessionEntity> userSessionCache = connections.getRemoteCache(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME);
+        RemoteCache<String, UserSessionEntity> offlineUserSessionsCache = connections.getRemoteCache(InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME);
+        RemoteCache<UUID, AuthenticatedClientSessionEntity> clientSessionCache = connections.getRemoteCache(InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME);
+        RemoteCache<UUID, AuthenticatedClientSessionEntity> offlineClientSessionsCache = connections.getRemoteCache(InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME);
+        remoteCacheHolder = new RemoteCacheHolder(userSessionCache, offlineUserSessionsCache, clientSessionCache, offlineClientSessionsCache);
+    }
+
+    private UserSessionTransaction createTransaction(KeycloakSession session) {
+        lazyInit(session);
+        return new UserSessionTransaction(
+                createUserSessionTransaction(false),
+                createUserSessionTransaction(true),
+                createClientSessionTransaction(false),
+                createClientSessionTransaction(true)
+        );
+    }
+
+    private RemoteChangeLogTransaction<String, UserSessionEntity, UserSessionUpdater> createUserSessionTransaction(boolean offline) {
+        return new RemoteChangeLogTransaction<>(UserSessionUpdater.factory(offline), remoteCacheHolder.userSessionCache(offline));
+    }
+
+    private RemoteChangeLogTransaction<UUID, AuthenticatedClientSessionEntity, AuthenticatedClientSessionUpdater> createClientSessionTransaction(boolean offline) {
+        return new RemoteChangeLogTransaction<>(AuthenticatedClientSessionUpdater.factory(offline), remoteCacheHolder.clientSessionCache(offline));
+    }
+
+    private record RemoteCacheHolder(
+            RemoteCache<String, UserSessionEntity> userSession,
+            RemoteCache<String, UserSessionEntity> offlineUserSession,
+            RemoteCache<UUID, AuthenticatedClientSessionEntity> clientSession,
+            RemoteCache<UUID, AuthenticatedClientSessionEntity> offlineClientSession) {
+
+        RemoteCache<String, UserSessionEntity> userSessionCache(boolean offline) {
+            return offline ? offlineUserSession : userSession;
+        }
+
+        RemoteCache<UUID, AuthenticatedClientSessionEntity> clientSessionCache(boolean offline) {
+            return offline ? offlineClientSession : clientSession;
+        }
+    }
+}

model/infinispan/src/main/resources/META-INF/services/org.keycloak.models.UserSessionProviderFactory

@@ -16,3 +16,4 @@
 #
 
 org.keycloak.models.sessions.infinispan.InfinispanUserSessionProviderFactory
+org.keycloak.models.sessions.infinispan.remote.RemoteUserSessionProviderFactory

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

@@ -223,12 +223,7 @@ public class CacheManagerFactory {
             var builders = builder.getNamedConfigurationBuilders();
             // remove all distributed caches
             logger.debug("Removing all distributed caches.");
-            // TODO [pruivo] remove all distributed caches after all of them are converted
+            Arrays.stream(CLUSTERED_CACHE_NAMES).forEach(builders::remove);
-            //DISTRIBUTED_REPLICATED_CACHE_NAMES.forEach(builders::remove);
-            builders.remove(WORK_CACHE_NAME);
-            builders.remove(AUTHENTICATION_SESSIONS_CACHE_NAME);
-            builders.remove(ACTION_TOKEN_CACHE);
-            builders.remove(LOGIN_FAILURE_CACHE_NAME);
         }
 
         var start = isStartEagerly();
@@ -291,12 +286,6 @@ public class CacheManagerFactory {
             transportConfig.addProperty(JGroupsTransport.SOCKET_FACTORY, tls.createSocketFactory());
             Logger.getLogger(CacheManagerFactory.class).info("MTLS enabled for communications for embedded caches");
         }
-
-        //TODO [pruivo] disable JGroups after all distributed caches are converted
-//        if (isCrossSiteEnabled() && isRemoteCacheEnabled()) {
-//            logger.debug("Disabling JGroups between Keycloak nodes");
-//            builder.getGlobalConfigurationBuilder().nonClusteredDefault();
-//        }
     }
 
     private void validateTlsAvailable(GlobalConfiguration config) {

testsuite/integration-arquillian/servers/app-server/jboss/wildfly/pom.xml

@@ -16,6 +16,7 @@
 ~ limitations under the License.
 -->
 
+
 <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
     <parent>

testsuite/integration-arquillian/servers/auth-server/services/testsuite-providers/src/main/java/org/keycloak/testsuite/model/infinispan/InfinispanTestUtil.java

@@ -47,7 +47,7 @@ public class InfinispanTestUtil {
         InfinispanConnectionProvider ispnProvider = session.getProvider(InfinispanConnectionProvider.class);
         if (ispnProvider != null) {
             logger.info("Will set KeycloakIspnTimeService to the infinispan cacheManager");
-            EmbeddedCacheManager cacheManager = ispnProvider.getCache(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME).getCacheManager();
+            EmbeddedCacheManager cacheManager = ispnProvider.getCache(InfinispanConnectionProvider.USER_CACHE_NAME).getCacheManager();
             origTimeService = setTimeServiceToKeycloakTime(cacheManager);
         }
     }

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

@@ -213,12 +213,13 @@ public abstract class AbstractQuarkusDeployableContainer implements DeployableCo
 
         var features = getDefaultFeatures();
         if (features.contains("remote-cache") && features.contains("multi-site")) {
-            commands.add("--cache-remote-host=localhost");
+            commands.add("--cache-remote-host=127.0.0.1");
             commands.add("--cache-remote-username=keycloak");
             commands.add("--cache-remote-password=Password1!");
             commands.add("--cache-remote-tls-enabled=false");
             commands.add("--spi-connections-infinispan-quarkus-site-name=test");
             configuration.appendJavaOpts("-Dkc.cache-remote-create-caches=true");
+            System.setProperty("kc.cache-remote-create-caches", "true");
         }
 
         return commands;

testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/model/UserSessionProviderTest.java

@@ -23,7 +23,6 @@ import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.keycloak.common.util.Time;
-import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.models.AuthenticatedClientSessionModel;
 import org.keycloak.models.ClientModel;
 import org.keycloak.models.KeycloakSession;
@@ -328,15 +327,8 @@ public class UserSessionProviderTest extends AbstractTestRealmKeycloakTest {
         var user1 = session.users().getUserByUsername(realm, "user1");
         var user2 = session.users().getUserByUsername(realm, "user2");
 
-        // TODO! [pruivo] to be removed when the session cache is remote only
+        assertEquals(0, session.sessions().getUserSessionsStream(realm, user1).count());
-        // TODO! the Hot Rod events are async
+        assertEquals(0, session.sessions().getUserSessionsStream(realm, user2).count());
-        if (InfinispanUtils.isRemoteInfinispan()) {
-            eventuallyEquals(null, 0L, () -> session.sessions().getUserSessionsStream(realm, user1).count());
-            eventuallyEquals(null, 0L, () -> session.sessions().getUserSessionsStream(realm, user2).count());
-        } else {
-            assertEquals(0, session.sessions().getUserSessionsStream(realm, user1).count());
-            assertEquals(0, session.sessions().getUserSessionsStream(realm, user2).count());
-        }
     }
 
     @Test

testsuite/integration-arquillian/tests/base/testsuites/remote-cache-suite

@@ -0,0 +1,25 @@
+SessionTest
+KcSamlBrokerSessionNotOnOrAfterTest
+OidcClaimToUserSessionNoteMapperTest
+KcOidcBrokerTransientSessionsTest
+KcAdmSessionTest
+TransientSessionTest
+UserSessionProviderOfflineTest
+AuthenticationSessionProviderTest
+UserSessionProviderTest
+OAuthDanceClientSessionExtensionTest
+SessionNotOnOrAfterTest
+SessionTimeoutValidationTest
+KcOidcUserSessionLimitsBrokerTest
+KcSamlUserSessionLimitsBrokerTest
+AbstractUserSessionLimitsBrokerTest
+UserSessionLimitsTest
+ConcurrentLoginTest
+RefreshTokenTest
+OfflineTokenTest
+AccessTokenTest
+LogoutTest
+ClientStorageTest
+UserInfoTest
+LightWeightAccessTokenTest
+TokenIntrospectionTest

testsuite/model/src/test/java/org/keycloak/testsuite/model/infinispan/FeatureEnabledTest.java

@@ -66,19 +66,7 @@ public class FeatureEnabledTest extends KeycloakModelTest {
         assertFalse(InfinispanUtils.isEmbeddedInfinispan());
         inComittedTransaction(session -> {
             var clusterProvider = session.getProvider(InfinispanConnectionProvider.class);
-            assertEmbeddedCacheDoesNotExists(clusterProvider, WORK_CACHE_NAME);
+            Arrays.stream(CLUSTERED_CACHE_NAMES).forEach(s -> assertEmbeddedCacheDoesNotExists(clusterProvider, s));
-            assertEmbeddedCacheDoesNotExists(clusterProvider, AUTHENTICATION_SESSIONS_CACHE_NAME);
-            assertEmbeddedCacheDoesNotExists(clusterProvider, ACTION_TOKEN_CACHE);
-            assertEmbeddedCacheDoesNotExists(clusterProvider, LOGIN_FAILURE_CACHE_NAME);
-
-            // TODO [pruivo] all caches eventually won't exists in embedded
-            Arrays.stream(CLUSTERED_CACHE_NAMES)
-                    .filter(Predicate.not(Predicate.isEqual(WORK_CACHE_NAME)))
-                    .filter(Predicate.not(Predicate.isEqual(AUTHENTICATION_SESSIONS_CACHE_NAME)))
-                    .filter(Predicate.not(Predicate.isEqual(ACTION_TOKEN_CACHE)))
-                    .filter(Predicate.not(Predicate.isEqual(LOGIN_FAILURE_CACHE_NAME)))
-                    .forEach(s -> assertEmbeddedCacheExists(clusterProvider, s));
-
             Arrays.stream(CLUSTERED_CACHE_NAMES).forEach(s -> assertRemoteCacheExists(clusterProvider, s));
 
         });

testsuite/model/src/test/java/org/keycloak/testsuite/model/parameters/RemoteInfinispan.java

@@ -31,6 +31,7 @@ import org.keycloak.models.sessions.infinispan.remote.RemoteInfinispanAuthentica
 import org.keycloak.models.sessions.infinispan.remote.RemoteInfinispanSingleUseObjectProviderFactory;
 import org.keycloak.models.sessions.infinispan.remote.RemoteStickySessionEncoderProviderFactory;
 import org.keycloak.models.sessions.infinispan.remote.RemoteUserLoginFailureProviderFactory;
+import org.keycloak.models.sessions.infinispan.remote.RemoteUserSessionProviderFactory;
 import org.keycloak.provider.ProviderFactory;
 import org.keycloak.testsuite.model.Config;
 import org.keycloak.testsuite.model.HotRodServerRule;
@@ -61,6 +62,7 @@ public class RemoteInfinispan extends KeycloakModelParameters {
             .add(RemoteStickySessionEncoderProviderFactory.class)
             .add(RemoteLoadBalancerCheckProviderFactory.class)
             .add(RemoteUserLoginFailureProviderFactory.class)
+            .add(RemoteUserSessionProviderFactory.class)
             .build();
 
     @Override
@@ -105,7 +107,7 @@ public class RemoteInfinispan extends KeycloakModelParameters {
     @Override
     public <T> Stream<T> getParameters(Class<T> clazz) {
         if (HotRodServerRule.class.isAssignableFrom(clazz)) {
-            return Stream.of((T) hotRodServerRule);
+            return Stream.of(clazz.cast(hotRodServerRule));
         } else {
             return Stream.empty();
         }

testsuite/model/src/test/java/org/keycloak/testsuite/model/session/OfflineSessionPersistenceTest.java

@@ -16,23 +16,6 @@
  */
 package org.keycloak.testsuite.model.session;
 
-import org.infinispan.commons.CacheException;
-import org.keycloak.models.AuthenticatedClientSessionModel;
-import org.keycloak.models.ClientModel;
-import org.keycloak.models.Constants;
-import org.keycloak.models.KeycloakSession;
-import org.keycloak.models.RealmModel;
-import org.keycloak.models.RealmProvider;
-import org.keycloak.models.UserModel;
-import org.keycloak.models.UserProvider;
-import org.keycloak.models.UserSessionModel;
-import org.keycloak.models.UserSessionProvider;
-import org.keycloak.models.session.UserSessionPersisterProvider;
-import org.keycloak.models.sessions.infinispan.InfinispanUserSessionProvider;
-import org.keycloak.models.sessions.infinispan.PersistentUserSessionProvider;
-import org.keycloak.services.managers.RealmManager;
-import org.keycloak.testsuite.model.KeycloakModelTest;
-import org.keycloak.testsuite.model.RequireProvider;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedList;
@@ -50,8 +33,27 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+
 import org.hamcrest.Matchers;
+import org.infinispan.commons.CacheException;
 import org.junit.Test;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.Constants;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.RealmProvider;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserProvider;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.UserSessionProvider;
+import org.keycloak.models.session.UserSessionPersisterProvider;
+import org.keycloak.models.sessions.infinispan.InfinispanUserSessionProvider;
+import org.keycloak.models.sessions.infinispan.PersistentUserSessionProvider;
+import org.keycloak.models.sessions.infinispan.remote.RemoteUserSessionProvider;
+import org.keycloak.services.managers.RealmManager;
+import org.keycloak.testsuite.model.KeycloakModelTest;
+import org.keycloak.testsuite.model.RequireProvider;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
@@ -263,6 +265,8 @@ public class OfflineSessionPersistenceTest extends KeycloakModelTest {
                 ((InfinispanUserSessionProvider) provider).removeLocalUserSessions(realm.getId(), true);
             } else if (provider instanceof PersistentUserSessionProvider) {
                 ((PersistentUserSessionProvider) provider).removeLocalUserSessions(realm.getId(), true);
+            } else if (provider instanceof RemoteUserSessionProvider) {
+                //no-op, session not local
             } else {
                 throw new IllegalStateException("Unknown UserSessionProvider: " + provider);
             }

testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionInitializerTest.java

@@ -17,14 +17,19 @@
 
 package org.keycloak.testsuite.model.session;
 
-import org.hamcrest.Matchers;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
 import org.infinispan.Cache;
 import org.infinispan.client.hotrod.RemoteCache;
 import org.junit.Assert;
-import org.junit.Assume;
 import org.junit.Test;
-import org.keycloak.common.Profile;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
+import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.models.ClientModel;
 import org.keycloak.models.Constants;
 import org.keycloak.models.KeycloakSession;
@@ -36,21 +41,13 @@ import org.keycloak.models.UserProvider;
 import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.UserSessionProvider;
 import org.keycloak.models.session.UserSessionPersisterProvider;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Collectors;
-
 import org.keycloak.testsuite.model.HotRodServerRule;
 import org.keycloak.testsuite.model.KeycloakModelTest;
 import org.keycloak.testsuite.model.RequireProvider;
 
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Every.everyItem;
 import static org.hamcrest.core.Is.is;
-import static org.hamcrest.MatcherAssert.assertThat;
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
 
 /**
@@ -179,8 +176,10 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
                     // try to get the user session at other nodes and also at different sites
                     inComittedTransaction(session -> {
                         InfinispanConnectionProvider provider = session.getProvider(InfinispanConnectionProvider.class);
-                        Cache<String, Object> localSessions = provider.getCache(USER_SESSION_CACHE_NAME);
+                        if (InfinispanUtils.isEmbeddedInfinispan()) {
-                        containsSession.get().add(localSessions.containsKey(userSessionId.get()));
+                            Cache<String, Object> localSessions = provider.getCache(USER_SESSION_CACHE_NAME);
+                            containsSession.get().add(localSessions.containsKey(userSessionId.get()));
+                        }
 
                         if (hotRodServer.isPresent()) {
                             RemoteCache<String, Object> remoteSessions = provider.getRemoteCache(USER_SESSION_CACHE_NAME);
@@ -194,7 +193,7 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
         assertThat(containsSession.get(), everyItem(is(true)));
 
         // 3 nodes (first node just creates the session), with Hot Rod server we have local + remote cache, without just local cache
-        int size = hotRodServer.isPresent() ? 6 : 3;
+        int size = hotRodServer.isPresent() && InfinispanUtils.isEmbeddedInfinispan() ? 6 : 3;
         assertThat(containsSession.get().size(), is(size));
     }
 

testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionPersisterProviderTest.java

@@ -23,6 +23,7 @@ import org.junit.Test;
 import org.keycloak.OAuth2Constants;
 import org.keycloak.common.Profile;
 import org.keycloak.common.util.Time;
+import org.keycloak.infinispan.util.InfinispanUtils;
 import org.keycloak.models.AuthenticatedClientSessionModel;
 import org.keycloak.models.ClientModel;
 import org.keycloak.models.KeycloakSession;
@@ -396,7 +397,10 @@ public class UserSessionPersisterProviderTest extends KeycloakModelTest {
             RealmModel realm = session.realms().getRealm(realmId);
 
             UserSessionPersisterProvider persister = session.getProvider(UserSessionPersisterProvider.class);
-            Assert.assertEquals(1, persister.getUserSessionsCount(true));
+            if (InfinispanUtils.isEmbeddedInfinispan()) {
+                // when configured with external Infinispan only, the sessions are not persisted into the database.
+                Assert.assertEquals(1, persister.getUserSessionsCount(true));
+            }
 
             List<UserSessionModel> loadedSessions = loadPersistedSessionsPaginated(session, true, 10, 1, 1);
             UserSessionModel persistedSession = loadedSessions.get(0);

testsuite/model/src/test/java/org/keycloak/testsuite/model/session/UserSessionProviderOfflineModelTest.java

@@ -17,34 +17,6 @@
 
 package org.keycloak.testsuite.model.session;
 
-import org.hamcrest.Matchers;
-import org.infinispan.AdvancedCache;
-import org.infinispan.Cache;
-import org.infinispan.context.Flag;
-import org.junit.Assert;
-import org.junit.Assume;
-import org.junit.Test;
-import org.keycloak.common.Profile;
-import org.keycloak.common.util.Time;
-import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
-import org.keycloak.models.AuthenticatedClientSessionModel;
-import org.keycloak.models.ClientModel;
-import org.keycloak.models.Constants;
-import org.keycloak.models.KeycloakSession;
-import org.keycloak.models.RealmModel;
-import org.keycloak.models.RealmProvider;
-import org.keycloak.models.UserManager;
-import org.keycloak.models.UserModel;
-import org.keycloak.models.UserProvider;
-import org.keycloak.models.UserSessionModel;
-import org.keycloak.models.UserSessionProvider;
-import org.keycloak.models.session.UserSessionPersisterProvider;
-import org.keycloak.models.sessions.infinispan.changes.sessions.PersisterLastSessionRefreshStoreFactory;
-import org.keycloak.models.utils.ResetTimeOffsetEvent;
-import org.keycloak.services.managers.UserSessionManager;
-import org.keycloak.testsuite.model.infinispan.InfinispanTestUtil;
-import org.keycloak.timer.TimerProvider;
-
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -61,8 +33,36 @@ import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import org.hamcrest.Matchers;
+import org.infinispan.AdvancedCache;
+import org.infinispan.Cache;
+import org.infinispan.context.Flag;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+import org.keycloak.common.Profile;
+import org.keycloak.common.util.Time;
+import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
+import org.keycloak.infinispan.util.InfinispanUtils;
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.Constants;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.RealmProvider;
+import org.keycloak.models.UserManager;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserProvider;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.UserSessionProvider;
+import org.keycloak.models.session.UserSessionPersisterProvider;
+import org.keycloak.models.sessions.infinispan.changes.sessions.PersisterLastSessionRefreshStoreFactory;
+import org.keycloak.models.utils.ResetTimeOffsetEvent;
+import org.keycloak.services.managers.UserSessionManager;
 import org.keycloak.testsuite.model.KeycloakModelTest;
 import org.keycloak.testsuite.model.RequireProvider;
+import org.keycloak.testsuite.model.infinispan.InfinispanTestUtil;
+import org.keycloak.timer.TimerProvider;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 
@@ -167,8 +167,11 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
                 UserSessionModel session0 = session.sessions().getOfflineUserSession(realm, origSessions[0].getId());
                 Assert.assertNotNull(session0);
 
-                // sessions are in persister too
+                // skip for remote cache feature
-                Assert.assertEquals(3, persister.getUserSessionsCount(true));
+                if (InfinispanUtils.isEmbeddedInfinispan()) {
+                    // sessions are in persister too
+                    Assert.assertEquals(3, persister.getUserSessionsCount(true));
+                }
 
                 setTimeOffset(300);
                 log.infof("Set time offset to 300. Time is: %d", Time.currentTime());
@@ -215,7 +218,9 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
                 Assert.assertNull(session.sessions().getOfflineUserSession(realm, origSessions[1].getId()));
                 Assert.assertNull(session.sessions().getOfflineUserSession(realm, origSessions[2].getId()));
 
-                Assert.assertEquals(1, persister.getUserSessionsCount(true));
+                if (InfinispanUtils.isEmbeddedInfinispan()) {
+                    Assert.assertEquals(1, persister.getUserSessionsCount(true));
+                }
 
                 // Expire everything and assert nothing found
                 setTimeOffset(7000000);
@@ -238,7 +243,7 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
             setTimeOffset(0);
             kcSession.getKeycloakSessionFactory().publish(new ResetTimeOffsetEvent());
             // Enable periodic task again, skip for persistent user sessions as the periodic task is not used there
-            if (timer != null && !Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
+            if (timer != null && timerTaskCtx != null && !Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
                 timer.schedule(timerTaskCtx.getRunnable(), timerTaskCtx.getIntervalMillis(), PersisterLastSessionRefreshStoreFactory.DB_LSR_PERIODIC_TASK_NAME);
             }
 
@@ -276,11 +281,14 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
 
             log.info("Persisted 3 sessions to UserSessionPersisterProvider");
 
-            inComittedTransaction(session -> {
+            if (InfinispanUtils.isEmbeddedInfinispan()) {
-                persister = session.getProvider(UserSessionPersisterProvider.class);
+                // external Infinispan does not store data in UserSessionPersisterProvider
+                inComittedTransaction(session -> {
+                    persister = session.getProvider(UserSessionPersisterProvider.class);
 
-                Assert.assertEquals(3, persister.getUserSessionsCount(true));
+                    Assert.assertEquals(3, persister.getUserSessionsCount(true));
-            });
+                });
+            }
 
             inComittedTransaction(session -> {
                 RealmModel realm = session.realms().getRealm(realmId);
@@ -304,14 +312,14 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
                 session.sessions().createOfflineUserSession(userSession);
                 session.sessions().createOfflineUserSession(origSessions[0]);
 
-                if (!Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
+                if (!Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS) && InfinispanUtils.isEmbeddedInfinispan()) {
                     // This does not work with persistent user sessions because we currently have two transactions and the one that creates the offline user sessions is not committing the changes
                     // try to load user session from persister
                     Assert.assertEquals(2, persister.loadUserSessionsStream(0, 10, true, "00000000-0000-0000-0000-000000000000").count());
                 }
             });
 
-            if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
+            if (Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS) && InfinispanUtils.isEmbeddedInfinispan()) {
                 inComittedTransaction(session -> {
                     persister = session.getProvider(UserSessionPersisterProvider.class);
                     Assert.assertEquals(2, persister.loadUserSessionsStream(0, 10, true, "00000000-0000-0000-0000-000000000000").count());
@@ -323,7 +331,7 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
             kcSession.getKeycloakSessionFactory().publish(new ResetTimeOffsetEvent());
 
             // Enable periodic task again, skip for persistent user sessions as the periodic task is not used there
-            if (timer != null && !Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
+            if (timer != null && timerTaskCtx != null && !Profile.isFeatureEnabled(Profile.Feature.PERSISTENT_USER_SESSIONS)) {
                 timer.schedule(timerTaskCtx.getRunnable(), timerTaskCtx.getIntervalMillis(), PersisterLastSessionRefreshStoreFactory.DB_LSR_PERIODIC_TASK_NAME);
             }
 
@@ -367,16 +375,19 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
             }
             awaitLatch(afterFirstNodeLatch);
 
-            log.debug("Joining the cluster");
+            if (InfinispanUtils.isEmbeddedInfinispan()) {
-            inComittedTransaction(session -> {
+                log.debug("Joining the cluster");
-                InfinispanConnectionProvider provider = session.getProvider(InfinispanConnectionProvider.class);
+                inComittedTransaction(session -> {
-                Cache<String, Object> cache = provider.getCache(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME);
+                    InfinispanConnectionProvider provider = session.getProvider(InfinispanConnectionProvider.class);
-                while (! cache.getAdvancedCache().getDistributionManager().isJoinComplete()) {
+                    Cache<String, Object> cache = provider.getCache(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME);
-                    sleep(1000);
+                    while (!cache.getAdvancedCache().getDistributionManager().isJoinComplete()) {
-                }
+                        sleep(1000);
-                cache.keySet().forEach(s -> {});
+                    }
-            });
+                    cache.keySet().forEach(s -> {
-            log.debug("Cluster joined");
+                    });
+                });
+                log.debug("Cluster joined");
+            }
 
             withRealm(realmId, (session, realm) -> {
                 final UserModel user = session.users().getUserByUsername(realm, "user1");
@@ -393,6 +404,7 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
 
     @Test
     public void testOfflineClientSessionLoading() {
+        Assume.assumeTrue("Remote Infinispan feature does not store sessions in UserSessionPersisterProvider", InfinispanUtils.isEmbeddedInfinispan());
         // create online user and client sessions
         inComittedTransaction((Consumer<KeycloakSession>) session -> UserSessionPersisterProviderTest.createSessions(session, realmId));
 
@@ -429,6 +441,7 @@ public class UserSessionProviderOfflineModelTest extends KeycloakModelTest {
 
     @Test
     public void testLoadingOfflineClientSessionWhenCreatedBeforeSessionTime() {
+        Assume.assumeTrue("Remote Infinispan feature does not store sessions in UserSessionPersisterProvider", InfinispanUtils.isEmbeddedInfinispan());
         // setup idle timeout for the realm
         int idleTimeout = (int) TimeUnit.DAYS.toSeconds(1);
         withRealm(realmId, (session, realmModel) -> {