KEYCLOAK-18445 Add support for cross-site model tests

model/infinispan/pom.xml

@@ -75,5 +75,10 @@
             <artifactId>microprofile-metrics-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+        </dependency>
     </dependencies>
 </project>

model/infinispan/src/main/java/org/keycloak/cluster/infinispan/InfinispanClusterProviderFactory.java

@@ -36,7 +36,7 @@ import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.connections.infinispan.TopologyInfo;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 import java.io.Serializable;
 import java.util.Collection;

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

@@ -17,39 +17,29 @@
 
 package org.keycloak.connections.infinispan;
 
-import java.util.Iterator;
-import java.util.ServiceLoader;
-import java.util.concurrent.TimeUnit;
-
 import org.infinispan.client.hotrod.ProtocolVersion;
 import org.infinispan.commons.dataconversion.MediaType;
-import org.infinispan.commons.util.FileLookup;
-import org.infinispan.commons.util.FileLookupFactory;
 import org.infinispan.configuration.cache.CacheMode;
 import org.infinispan.configuration.cache.Configuration;
 import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.configuration.global.GlobalConfigurationBuilder;
-import org.infinispan.configuration.global.TransportConfigurationBuilder;
 import org.infinispan.eviction.EvictionStrategy;
 import org.infinispan.eviction.EvictionType;
 import org.infinispan.jboss.marshalling.core.JBossUserMarshaller;
 import org.infinispan.manager.DefaultCacheManager;
 import org.infinispan.manager.EmbeddedCacheManager;
-import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
+import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
 import org.infinispan.transaction.LockingMode;
 import org.infinispan.transaction.TransactionMode;
 import org.infinispan.transaction.lookup.EmbeddedTransactionManagerLookup;
 import org.jboss.logging.Logger;
-import org.jgroups.JChannel;
 import org.keycloak.Config;
 import org.keycloak.cluster.ClusterEvent;
 import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.cluster.ManagedCacheManagerProvider;
 import org.keycloak.cluster.infinispan.KeycloakHotRodMarshallerFactory;
-import org.keycloak.common.util.Time;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
-
 import org.keycloak.models.cache.infinispan.ClearCacheEvent;
 import org.keycloak.models.cache.infinispan.events.RealmRemovedEvent;
 import org.keycloak.models.cache.infinispan.events.RealmUpdatedEvent;
@@ -57,16 +47,17 @@ import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;
 import org.keycloak.provider.InvalidationHandler.ObjectType;
 import org.keycloak.provider.ProviderEvent;
-import java.time.Instant;
+
-import java.util.concurrent.atomic.AtomicReference;
+import java.util.Iterator;
-import org.infinispan.commons.time.TimeService;
+import java.util.ServiceLoader;
-import org.infinispan.factories.GlobalComponentRegistry;
+import java.util.concurrent.TimeUnit;
-import org.infinispan.factories.impl.BasicComponentRegistry;
+
-import org.infinispan.factories.impl.ComponentRef;
-import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
-import org.infinispan.util.EmbeddedTimeService;
 import static org.keycloak.models.cache.infinispan.InfinispanCacheRealmProviderFactory.REALM_CLEAR_CACHE_EVENTS;
 import static org.keycloak.models.cache.infinispan.InfinispanCacheRealmProviderFactory.REALM_INVALIDATION_EVENTS;
+import static org.keycloak.connections.infinispan.InfinispanUtil.configureTransport;
+import static org.keycloak.connections.infinispan.InfinispanUtil.createCacheConfigurationBuilder;
+import static org.keycloak.connections.infinispan.InfinispanUtil.getActionTokenCacheConfig;
+import static org.keycloak.connections.infinispan.InfinispanUtil.setTimeServiceToKeycloakTime;
 
 /**
  * @author <a href="mailto:sthorger@redhat.com">Stian Thorgersen</a>
@@ -205,7 +196,8 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
 
         if (clustered) {
             String jgroupsUdpMcastAddr = config.get("jgroupsUdpMcastAddr", System.getProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR));
-            configureTransport(gcb, topologyInfo.getMyNodeName(), topologyInfo.getMySiteName(), jgroupsUdpMcastAddr);
+            configureTransport(gcb, topologyInfo.getMyNodeName(), topologyInfo.getMySiteName(), jgroupsUdpMcastAddr,
+                    "default-configs/default-keycloak-jgroups-udp.xml");
             gcb.jmx()
               .domain(InfinispanConnectionProvider.JMX_DOMAIN + "-" + topologyInfo.getMyNodeName()).enable();
         } else {
@@ -266,7 +258,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         if (jdgEnabled) {
             sessionConfigBuilder = createCacheConfigurationBuilder();
             sessionConfigBuilder.read(sessionCacheConfigurationBase);
-            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.USER_SESSION_CACHE_NAME, true);
+            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.USER_SESSION_CACHE_NAME);
         }
         Configuration sessionCacheConfiguration = sessionConfigBuilder.build();
         cacheManager.defineConfiguration(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME, sessionCacheConfiguration);
@@ -274,7 +266,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         if (jdgEnabled) {
             sessionConfigBuilder = createCacheConfigurationBuilder();
             sessionConfigBuilder.read(sessionCacheConfigurationBase);
-            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME, true);
+            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME);
         }
         sessionCacheConfiguration = sessionConfigBuilder.build();
         cacheManager.defineConfiguration(InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME, sessionCacheConfiguration);
@@ -282,7 +274,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         if (jdgEnabled) {
             sessionConfigBuilder = createCacheConfigurationBuilder();
             sessionConfigBuilder.read(sessionCacheConfigurationBase);
-            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME, true);
+            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME);
         }
         sessionCacheConfiguration = sessionConfigBuilder.build();
         cacheManager.defineConfiguration(InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME, sessionCacheConfiguration);
@@ -290,7 +282,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         if (jdgEnabled) {
             sessionConfigBuilder = createCacheConfigurationBuilder();
             sessionConfigBuilder.read(sessionCacheConfigurationBase);
-            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME, true);
+            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME);
         }
         sessionCacheConfiguration = sessionConfigBuilder.build();
         cacheManager.defineConfiguration(InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME, sessionCacheConfiguration);
@@ -298,7 +290,7 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         if (jdgEnabled) {
             sessionConfigBuilder = createCacheConfigurationBuilder();
             sessionConfigBuilder.read(sessionCacheConfigurationBase);
-            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.LOGIN_FAILURE_CACHE_NAME, true);
+            configureRemoteCacheStore(sessionConfigBuilder, async, InfinispanConnectionProvider.LOGIN_FAILURE_CACHE_NAME);
         }
         sessionCacheConfiguration = sessionConfigBuilder.build();
         cacheManager.defineConfiguration(InfinispanConnectionProvider.LOGIN_FAILURE_CACHE_NAME, sessionCacheConfiguration);
@@ -319,12 +311,13 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         }
 
         if (jdgEnabled) {
-            configureRemoteCacheStore(replicationConfigBuilder, async, InfinispanConnectionProvider.WORK_CACHE_NAME, false);
+            configureRemoteCacheStore(replicationConfigBuilder, async, InfinispanConnectionProvider.WORK_CACHE_NAME);
         }
 
         Configuration replicationEvictionCacheConfiguration = replicationConfigBuilder
-          .expiration().enableReaper().wakeUpInterval(15, TimeUnit.SECONDS)
+            .expiration().enableReaper().wakeUpInterval(15, TimeUnit.SECONDS)
-          .build();
+            .build();
+
         cacheManager.defineConfiguration(InfinispanConnectionProvider.WORK_CACHE_NAME, replicationEvictionCacheConfiguration);
         cacheManager.getCache(InfinispanConnectionProvider.WORK_CACHE_NAME, true);
 
@@ -366,26 +359,6 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         cacheManager.getCache(InfinispanConnectionProvider.AUTHORIZATION_REVISIONS_CACHE_NAME, true);
     }
 
-    /**
-     * Replaces the {@link TimeService} in infinispan with the one that respects Keycloak {@link Time}.
-     * @param cacheManager
-     * @return Runnable to revert replacement of the infinispan time service
-     */
-    public static Runnable setTimeServiceToKeycloakTime(EmbeddedCacheManager cacheManager) {
-        TimeService previousTimeService = replaceComponent(cacheManager, TimeService.class, KEYCLOAK_TIME_SERVICE, true);
-        AtomicReference<TimeService> ref = new AtomicReference<>(previousTimeService);
-        return () -> {
-            if (ref.get() == null) {
-                logger.warn("Calling revert of the TimeService when testing TimeService was already reverted");
-                return;
-            }
-
-            logger.info("Revert set KeycloakIspnTimeService to the infinispan cacheManager");
-
-            replaceComponent(cacheManager, TimeService.class, ref.getAndSet(null), true);
-        };
-    }
-
     private Configuration getRevisionCacheConfig(long maxEntries) {
         ConfigurationBuilder cb = createCacheConfigurationBuilder();
         cb.invocationBatching().enable().transaction().transactionMode(TransactionMode.TRANSACTIONAL);
@@ -406,42 +379,29 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         return cb.build();
     }
 
-    private ConfigurationBuilder createCacheConfigurationBuilder() {
-        ConfigurationBuilder builder = new ConfigurationBuilder();
-
-        // need to force the encoding to application/x-java-object to avoid unnecessary conversion of keys/values. See WFLY-14356.
-        builder.encoding().mediaType(MediaType.APPLICATION_OBJECT_TYPE);
-
-        return builder;
-    }
-
     // Used for cross-data centers scenario. Usually integration with external JDG server, which itself handles communication between DCs.
-    private void configureRemoteCacheStore(ConfigurationBuilder builder, boolean async, String cacheName, boolean sessionCache) {
+    private void configureRemoteCacheStore(ConfigurationBuilder builder, boolean async, String cacheName) {
         String jdgServer = config.get("remoteStoreHost", "localhost");
         Integer jdgPort = config.getInt("remoteStorePort", 11222);
 
         builder.persistence()
                 .passivation(false)
                 .addStore(RemoteStoreConfigurationBuilder.class)
-                    .fetchPersistentState(false)
+                .fetchPersistentState(false)
-                    .ignoreModifications(false)
+                .ignoreModifications(false)
-                    .purgeOnStartup(false)
+                .purgeOnStartup(false)
-                    .preload(false)
+                .preload(false)
-                    .shared(true)
+                .shared(true)
-                    .remoteCacheName(cacheName)
+                .remoteCacheName(cacheName)
-                    .rawValues(true)
+                .rawValues(true)
-                    .forceReturnValues(false)
+                .forceReturnValues(false)
-                    .marshaller(KeycloakHotRodMarshallerFactory.class.getName())
+                .marshaller(KeycloakHotRodMarshallerFactory.class.getName())
-                    .protocolVersion(getHotrodVersion())
+                .protocolVersion(getHotrodVersion())
-                    .addServer()
+                .addServer()
-                        .host(jdgServer)
+                .host(jdgServer)
-                        .port(jdgPort)
+                .port(jdgPort)
-//                  .connectionPool()
+                .async()
-//                      .maxActive(100)
+                .enabled(async);
-//                      .exhaustedAction(ExhaustedAction.CREATE_NEW)
-                    .async()
-                        .enabled(async);
-
     }
 
     private void configureRemoteActionTokenCacheStore(ConfigurationBuilder builder, boolean async) {
@@ -494,69 +454,6 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         return cb.build();
     }
 
-    private ConfigurationBuilder getActionTokenCacheConfig() {
-        ConfigurationBuilder cb = createCacheConfigurationBuilder();
-
-        cb.memory()
-                .evictionStrategy(EvictionStrategy.NONE)
-                .evictionType(EvictionType.COUNT)
-                .size(InfinispanConnectionProvider.ACTION_TOKEN_CACHE_DEFAULT_MAX);
-        cb.expiration()
-                .maxIdle(InfinispanConnectionProvider.ACTION_TOKEN_MAX_IDLE_SECONDS, TimeUnit.SECONDS)
-                .wakeUpInterval(InfinispanConnectionProvider.ACTION_TOKEN_WAKE_UP_INTERVAL_SECONDS, TimeUnit.SECONDS);
-
-        return cb;
-    }
-
-    private static final Object CHANNEL_INIT_SYNCHRONIZER = new Object();
-
-    protected void configureTransport(GlobalConfigurationBuilder gcb, String nodeName, String siteName, String jgroupsUdpMcastAddr) {
-        if (nodeName == null) {
-            gcb.transport().defaultTransport();
-        } else {
-            FileLookup fileLookup = FileLookupFactory.newInstance();
-
-            synchronized (CHANNEL_INIT_SYNCHRONIZER) {
-                String originalMcastAddr = System.getProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
-                if (jgroupsUdpMcastAddr == null) {
-                    System.getProperties().remove(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
-                } else {
-                    System.setProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR, jgroupsUdpMcastAddr);
-                }
-                try {
-                    JChannel channel = new JChannel(fileLookup.lookupFileLocation("default-configs/default-keycloak-jgroups-udp.xml", this.getClass().getClassLoader()).openStream());
-                    channel.setName(nodeName);
-                    JGroupsTransport transport = new JGroupsTransport(channel);
-
-                    TransportConfigurationBuilder transportBuilder = gcb.transport()
-                      .nodeName(nodeName)
-                      .siteId(siteName)
-                      .transport(transport);
-
-                    // Use the cluster corresponding to current site. This is needed as the nodes in different DCs should not share same cluster
-                    if (siteName != null) {
-                        transportBuilder.clusterName(siteName);
-                    }
-
-
-                    transportBuilder.jmx()
-                        .domain(InfinispanConnectionProvider.JMX_DOMAIN + "-" + nodeName)
-                        .enable();
-
-                    logger.infof("Configured jgroups transport with the channel name: %s", nodeName);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                } finally {
-                    if (originalMcastAddr == null) {
-                        System.getProperties().remove(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
-                    } else {
-                        System.setProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR, originalMcastAddr);
-                    }
-                }
-            }
-        }
-    }
-
     private void registerSystemWideListeners(KeycloakSession session) {
         KeycloakSessionFactory sessionFactory = session.getKeycloakSessionFactory();
         ClusterProvider cluster = session.getProvider(ClusterProvider.class);
@@ -575,51 +472,4 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
             }
         });
     }
-
-
-    /**
-     * Forked from org.infinispan.test.TestingUtil class
-     *
-     * Replaces a component in a running cache manager (global component registry).
-     *
-     * @param cacheMgr       cache in which to replace component
-     * @param componentType        component type of which to replace
-     * @param replacementComponent new instance
-     * @param rewire               if true, ComponentRegistry.rewire() is called after replacing.
-     *
-     * @return the original component that was replaced
-     */
-    private static <T> T replaceComponent(EmbeddedCacheManager cacheMgr, Class<T> componentType, T replacementComponent, boolean rewire) {
-        GlobalComponentRegistry cr = cacheMgr.getGlobalComponentRegistry();
-        BasicComponentRegistry bcr = cr.getComponent(BasicComponentRegistry.class);
-        ComponentRef<T> old = bcr.getComponent(componentType);
-        bcr.replaceComponent(componentType.getName(), replacementComponent, true);
-        if (rewire) {
-            cr.rewire();
-            cr.rewireNamedRegistries();
-        }
-        return old != null ? old.wired() : null;
-    }
-
-    public static final TimeService KEYCLOAK_TIME_SERVICE = new EmbeddedTimeService() {
-
-        private long getCurrentTimeMillis() {
-            return Time.currentTimeMillis();
-        }
-
-        @Override
-        public long wallClockTime() {
-            return getCurrentTimeMillis();
-        }
-
-        @Override
-        public long time() {
-            return TimeUnit.MILLISECONDS.toNanos(getCurrentTimeMillis());
-        }
-
-        @Override
-        public Instant instant() {
-            return Instant.ofEpochMilli(getCurrentTimeMillis());
-        }
-    };
 }

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

@@ -0,0 +1,252 @@
+/*
+ * Copyright 2016 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.connections.infinispan;
+
+import java.time.Instant;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.infinispan.Cache;
+import org.infinispan.client.hotrod.ProtocolVersion;
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.commons.api.BasicCache;
+import org.infinispan.commons.dataconversion.MediaType;
+import org.infinispan.commons.time.TimeService;
+import org.infinispan.commons.util.FileLookup;
+import org.infinispan.commons.util.FileLookupFactory;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.configuration.global.TransportConfigurationBuilder;
+import org.infinispan.eviction.EvictionStrategy;
+import org.infinispan.eviction.EvictionType;
+import org.infinispan.factories.GlobalComponentRegistry;
+import org.infinispan.factories.impl.BasicComponentRegistry;
+import org.infinispan.factories.impl.ComponentRef;
+import org.infinispan.manager.EmbeddedCacheManager;
+import org.infinispan.persistence.manager.PersistenceManager;
+import org.infinispan.persistence.remote.RemoteStore;
+import org.infinispan.remoting.transport.Transport;
+import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
+import org.infinispan.util.EmbeddedTimeService;
+import org.jboss.logging.Logger;
+import org.jgroups.JChannel;
+import org.keycloak.common.util.Time;
+import org.keycloak.models.KeycloakSession;
+
+/**
+ * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
+ */
+public class InfinispanUtil {
+
+    protected static final Logger logger = Logger.getLogger(InfinispanUtil.class);
+
+    public static final int MAXIMUM_REPLACE_RETRIES = 25;
+
+    // See if we have RemoteStore (external JDG) configured for cross-Data-Center scenario
+    public static Set<RemoteStore> getRemoteStores(Cache ispnCache) {
+        return ispnCache.getAdvancedCache().getComponentRegistry().getComponent(PersistenceManager.class).getStores(RemoteStore.class);
+    }
+
+
+    public static RemoteCache getRemoteCache(Cache ispnCache) {
+        Set<RemoteStore> remoteStores = getRemoteStores(ispnCache);
+        if (remoteStores.isEmpty()) {
+            return null;
+        } else {
+            return remoteStores.iterator().next().getRemoteCache();
+        }
+    }
+
+
+    public static TopologyInfo getTopologyInfo(KeycloakSession session) {
+        return session.getProvider(InfinispanConnectionProvider.class).getTopologyInfo();
+    }
+
+
+    /**
+     *
+     * @param cache
+     * @return true if cluster coordinator OR if it's local cache
+     */
+    public static boolean isCoordinator(Cache cache) {
+        Transport transport = cache.getCacheManager().getTransport();
+        return transport == null || transport.isCoordinator();
+    }
+
+    /**
+     * Convert the given value to the proper value, which can be used when calling operations for the infinispan remoteCache.
+     *
+     * Infinispan HotRod protocol of versions older than 3.0 uses the "lifespan" or "maxIdle" as the normal expiration time when the value is 30 days or less.
+     * However for the bigger values, it assumes that the value is unix timestamp.
+     *
+     * @param ispnCache
+     * @param lifespanOrigMs
+     * @return
+     */
+    public static long toHotrodTimeMs(BasicCache ispnCache, long lifespanOrigMs) {
+        if (ispnCache instanceof RemoteCache && lifespanOrigMs > 2592000000L) {
+            RemoteCache remoteCache = (RemoteCache) ispnCache;
+            ProtocolVersion protocolVersion = remoteCache.getRemoteCacheManager().getConfiguration().version();
+            if (ProtocolVersion.PROTOCOL_VERSION_30.compareTo(protocolVersion) > 0) {
+                return Time.currentTimeMillis() + lifespanOrigMs;
+            }
+        }
+
+        return lifespanOrigMs;
+    }
+
+    private static final Object CHANNEL_INIT_SYNCHRONIZER = new Object();
+
+    public static void configureTransport(GlobalConfigurationBuilder gcb, String nodeName, String siteName, String jgroupsUdpMcastAddr,
+                                      String jgroupsConfigPath) {
+        if (nodeName == null) {
+            gcb.transport().defaultTransport();
+        } else {
+            FileLookup fileLookup = FileLookupFactory.newInstance();
+
+            synchronized (CHANNEL_INIT_SYNCHRONIZER) {
+                String originalMcastAddr = System.getProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
+                if (jgroupsUdpMcastAddr == null) {
+                    System.getProperties().remove(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
+                } else {
+                    System.setProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR, jgroupsUdpMcastAddr);
+                }
+                try {
+                    JChannel channel = new JChannel(fileLookup.lookupFileLocation(jgroupsConfigPath, InfinispanUtil.class.getClassLoader()).openStream());
+                    channel.setName(nodeName);
+                    JGroupsTransport transport = new JGroupsTransport(channel);
+
+                    TransportConfigurationBuilder transportBuilder = gcb.transport()
+                            .nodeName(nodeName)
+                            .siteId(siteName)
+                            .transport(transport);
+
+                    // Use the cluster corresponding to current site. This is needed as the nodes in different DCs should not share same cluster
+                    if (siteName != null) {
+                        transportBuilder.clusterName(siteName);
+                    }
+
+
+                    transportBuilder.jmx()
+                            .domain(InfinispanConnectionProvider.JMX_DOMAIN + "-" + nodeName)
+                            .enable();
+
+                    logger.infof("Configured jgroups transport with the channel name: %s", nodeName);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                } finally {
+                    if (originalMcastAddr == null) {
+                        System.getProperties().remove(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR);
+                    } else {
+                        System.setProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR, originalMcastAddr);
+                    }
+                }
+            }
+        }
+    }
+
+    public static ConfigurationBuilder createCacheConfigurationBuilder() {
+        ConfigurationBuilder builder = new ConfigurationBuilder();
+
+        // need to force the encoding to application/x-java-object to avoid unnecessary conversion of keys/values. See WFLY-14356.
+        builder.encoding().mediaType(MediaType.APPLICATION_OBJECT_TYPE);
+
+        return builder;
+    }
+
+    public static ConfigurationBuilder getActionTokenCacheConfig() {
+        ConfigurationBuilder cb = createCacheConfigurationBuilder();
+
+        cb.memory()
+                .evictionStrategy(EvictionStrategy.NONE)
+                .evictionType(EvictionType.COUNT)
+                .size(InfinispanConnectionProvider.ACTION_TOKEN_CACHE_DEFAULT_MAX);
+        cb.expiration()
+                .maxIdle(InfinispanConnectionProvider.ACTION_TOKEN_MAX_IDLE_SECONDS, TimeUnit.SECONDS)
+                .wakeUpInterval(InfinispanConnectionProvider.ACTION_TOKEN_WAKE_UP_INTERVAL_SECONDS, TimeUnit.SECONDS);
+
+        return cb;
+    }
+
+    /**
+     * Replaces the {@link TimeService} in infinispan with the one that respects Keycloak {@link Time}.
+     * @param cacheManager
+     * @return Runnable to revert replacement of the infinispan time service
+     */
+    public static Runnable setTimeServiceToKeycloakTime(EmbeddedCacheManager cacheManager) {
+        TimeService previousTimeService = replaceComponent(cacheManager, TimeService.class, KEYCLOAK_TIME_SERVICE, true);
+        AtomicReference<TimeService> ref = new AtomicReference<>(previousTimeService);
+        return () -> {
+            if (ref.get() == null) {
+                logger.warn("Calling revert of the TimeService when testing TimeService was already reverted");
+                return;
+            }
+
+            logger.info("Revert set KeycloakIspnTimeService to the infinispan cacheManager");
+
+            replaceComponent(cacheManager, TimeService.class, ref.getAndSet(null), true);
+        };
+    }
+
+    /**
+     * Forked from org.infinispan.test.TestingUtil class
+     *
+     * Replaces a component in a running cache manager (global component registry).
+     *
+     * @param cacheMgr       cache in which to replace component
+     * @param componentType        component type of which to replace
+     * @param replacementComponent new instance
+     * @param rewire               if true, ComponentRegistry.rewire() is called after replacing.
+     *
+     * @return the original component that was replaced
+     */
+    private static <T> T replaceComponent(EmbeddedCacheManager cacheMgr, Class<T> componentType, T replacementComponent, boolean rewire) {
+        GlobalComponentRegistry cr = cacheMgr.getGlobalComponentRegistry();
+        BasicComponentRegistry bcr = cr.getComponent(BasicComponentRegistry.class);
+        ComponentRef<T> old = bcr.getComponent(componentType);
+        bcr.replaceComponent(componentType.getName(), replacementComponent, true);
+        if (rewire) {
+            cr.rewire();
+            cr.rewireNamedRegistries();
+        }
+        return old != null ? old.wired() : null;
+    }
+
+    public static final TimeService KEYCLOAK_TIME_SERVICE = new EmbeddedTimeService() {
+
+        private long getCurrentTimeMillis() {
+            return Time.currentTimeMillis();
+        }
+
+        @Override
+        public long wallClockTime() {
+            return getCurrentTimeMillis();
+        }
+
+        @Override
+        public long time() {
+            return TimeUnit.MILLISECONDS.toNanos(getCurrentTimeMillis());
+        }
+
+        @Override
+        public Instant instant() {
+            return Instant.ofEpochMilli(getCurrentTimeMillis());
+        }
+    };
+}

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

@@ -38,7 +38,7 @@ import org.infinispan.manager.EmbeddedCacheManager;
 import org.jboss.logging.Logger;
 import org.keycloak.Config;
 import org.keycloak.common.util.reflections.Reflections;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+
 import java.util.stream.Collectors;
 import org.infinispan.client.hotrod.exceptions.HotRodClientException;
 

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

@@ -29,7 +29,7 @@ import org.keycloak.common.util.Time;
 import org.keycloak.models.CodeToTokenStoreProvider;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>

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

@@ -24,7 +24,7 @@ import org.keycloak.common.util.Time;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.PushedAuthzRequestStoreProvider;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 import java.util.Map;
 import java.util.UUID;

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

@@ -24,10 +24,9 @@ import org.keycloak.common.util.Time;
 import org.keycloak.models.AuthenticatedClientSessionModel;
 import org.keycloak.models.SamlArtifactSessionMappingModel;
 import org.keycloak.models.SamlArtifactSessionMappingStoreProvider;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 import java.nio.charset.StandardCharsets;
-import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;

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

@@ -27,7 +27,7 @@ import org.keycloak.common.util.Time;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.SingleUseTokenStoreProvider;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * TODO: Check if Boolean can be used as single-use cache argument instead of ActionTokenValueEntity. With respect to other single-use cache usecases like "Revoke Refresh Token" .

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

@@ -30,7 +30,7 @@ import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
 import org.keycloak.models.SingleUseTokenStoreProviderFactory;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import static org.keycloak.models.sessions.infinispan.InfinispanAuthenticationSessionProviderFactory.PROVIDER_PRIORITY;
 
 /**

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

@@ -20,7 +20,7 @@ package org.keycloak.models.sessions.infinispan;
 import org.infinispan.Cache;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.KeycloakSession;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.sessions.StickySessionEncoderProvider;
 
 /**

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

@@ -30,7 +30,7 @@ import org.keycloak.common.util.Time;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.TokenRevocationStoreProvider;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>

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

@@ -42,7 +42,7 @@ import org.keycloak.models.sessions.infinispan.initializer.InfinispanCacheInitia
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheInvoker;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionListener;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionsLoader;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;

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

@@ -56,7 +56,7 @@ import org.keycloak.models.sessions.infinispan.stream.SessionPredicate;
 import org.keycloak.models.sessions.infinispan.stream.UserSessionPredicate;
 import org.keycloak.models.sessions.infinispan.util.FuturesHelper;
 import org.keycloak.models.sessions.infinispan.util.InfinispanKeyGenerator;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 
 import java.io.Serializable;

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

@@ -53,7 +53,7 @@ import org.keycloak.models.sessions.infinispan.initializer.OfflinePersistentUser
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionListener;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionsLoader;
 import org.keycloak.models.sessions.infinispan.util.InfinispanKeyGenerator;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.PostMigrationEvent;

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

@@ -32,7 +32,7 @@ import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.sessions.infinispan.CacheDecorators;
 import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheInvoker;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>

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

@@ -30,7 +30,7 @@ import org.keycloak.models.RealmModel;
 import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.models.utils.KeycloakModelUtils;
 
 /**

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/events/AbstractUserSessionClusterListener.java

@@ -24,7 +24,7 @@ import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.connections.infinispan.TopologyInfo;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.provider.Provider;
 

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/events/SessionClusterEvent.java

@@ -20,7 +20,7 @@ package org.keycloak.models.sessions.infinispan.events;
 import org.keycloak.cluster.ClusterEvent;
 import org.keycloak.connections.infinispan.TopologyInfo;
 import org.keycloak.models.KeycloakSession;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/RemoteCacheInvoker.java

@@ -36,7 +36,7 @@ import org.keycloak.models.sessions.infinispan.changes.MergedUpdate;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.changes.SessionUpdateTask;
 import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/RemoteCacheSessionListener.java

@@ -36,12 +36,11 @@ import org.keycloak.models.KeycloakSessionFactory;
 import org.keycloak.models.RealmModel;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 
 import org.infinispan.client.hotrod.VersionedValue;

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/util/InfinispanUtil.java

@@ -1,95 +0,0 @@
-/*
- * Copyright 2016 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.util;
-
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
-import org.infinispan.Cache;
-import org.infinispan.client.hotrod.ProtocolVersion;
-import org.infinispan.client.hotrod.RemoteCache;
-import org.infinispan.commons.api.BasicCache;
-import org.infinispan.persistence.manager.PersistenceManager;
-import org.infinispan.persistence.remote.RemoteStore;
-import org.infinispan.remoting.transport.Transport;
-import org.keycloak.common.util.Time;
-import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
-import org.keycloak.connections.infinispan.TopologyInfo;
-import org.keycloak.models.KeycloakSession;
-
-/**
- * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
- */
-public class InfinispanUtil {
-
-    public static final int MAXIMUM_REPLACE_RETRIES = 25;
-
-    // See if we have RemoteStore (external JDG) configured for cross-Data-Center scenario
-    public static Set<RemoteStore> getRemoteStores(Cache ispnCache) {
-        return ispnCache.getAdvancedCache().getComponentRegistry().getComponent(PersistenceManager.class).getStores(RemoteStore.class);
-    }
-
-
-    public static RemoteCache getRemoteCache(Cache ispnCache) {
-        Set<RemoteStore> remoteStores = getRemoteStores(ispnCache);
-        if (remoteStores.isEmpty()) {
-            return null;
-        } else {
-            return remoteStores.iterator().next().getRemoteCache();
-        }
-    }
-
-
-    public static TopologyInfo getTopologyInfo(KeycloakSession session) {
-        return session.getProvider(InfinispanConnectionProvider.class).getTopologyInfo();
-    }
-
-
-    /**
-     *
-     * @param cache
-     * @return true if cluster coordinator OR if it's local cache
-     */
-    public static boolean isCoordinator(Cache cache) {
-        Transport transport = cache.getCacheManager().getTransport();
-        return transport == null || transport.isCoordinator();
-    }
-
-    /**
-     * Convert the given value to the proper value, which can be used when calling operations for the infinispan remoteCache.
-     *
-     * Infinispan HotRod protocol of versions older than 3.0 uses the "lifespan" or "maxIdle" as the normal expiration time when the value is 30 days or less.
-     * However for the bigger values, it assumes that the value is unix timestamp.
-     *
-     * @param ispnCache
-     * @param lifespanOrigMs
-     * @return
-     */
-    public static long toHotrodTimeMs(BasicCache ispnCache, long lifespanOrigMs) {
-        if (ispnCache instanceof RemoteCache && lifespanOrigMs > 2592000000L) {
-            RemoteCache remoteCache = (RemoteCache) ispnCache;
-            ProtocolVersion protocolVersion = remoteCache.getRemoteCacheManager().getConfiguration().version();
-            if (ProtocolVersion.PROTOCOL_VERSION_30.compareTo(protocolVersion) > 0) {
-                return Time.currentTimeMillis() + lifespanOrigMs;
-            }
-        }
-
-        return lifespanOrigMs;
-    }
-
-}

model/infinispan/src/test/java/org/keycloak/cluster/infinispan/ConcurrencyJDGCacheReplaceTest.java

@@ -41,7 +41,7 @@ import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
 import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
 import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import java.util.UUID;
 import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
 

model/infinispan/src/test/java/org/keycloak/cluster/infinispan/ConcurrencyJDGRemoteCacheClientListenersTest.java

@@ -37,7 +37,7 @@ import org.infinispan.persistence.remote.RemoteStore;
 import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
 import org.junit.Assert;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * Test that hotrod ClientListeners are correctly executed as expected

model/infinispan/src/test/java/org/keycloak/cluster/infinispan/ConcurrencyJDGRemoveSessionTest.java

@@ -17,9 +17,7 @@
 
 package org.keycloak.cluster.infinispan;
 
-import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -44,7 +42,7 @@ import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
 import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import java.util.UUID;
 
 /**

model/infinispan/src/test/java/org/keycloak/cluster/infinispan/JDGPutTest.java

@@ -18,7 +18,6 @@
 
 package org.keycloak.cluster.infinispan;
 
-import java.awt.print.Book;
 import java.net.SocketAddress;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -40,7 +39,7 @@ import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionsLoaderContext;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 import static org.infinispan.client.hotrod.impl.Util.await;
 

model/infinispan/src/test/java/org/keycloak/cluster/infinispan/RemoteCacheSessionsLoaderTest.java

@@ -37,7 +37,7 @@ import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
 import org.keycloak.models.sessions.infinispan.initializer.SessionLoader;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionsLoader;
 import org.keycloak.models.sessions.infinispan.remotestore.RemoteCacheSessionsLoaderContext;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>

model/map/src/main/java/org/keycloak/models/map/userSession/MapUserSessionAdapter.java

@@ -111,22 +111,22 @@ public abstract class MapUserSessionAdapter extends AbstractUserSessionModel {
         Map<String, AuthenticatedClientSessionModel> result = new HashMap<>();
         List<String> removedClientUUIDS = new LinkedList<>();
 
-        entity.getAuthenticatedClientSessions().entrySet()
+        // to avoid concurrentModificationException
-                .stream()
+        Map<String, String> authenticatedClientSessions = new HashMap<>(entity.getAuthenticatedClientSessions());
-                .forEach(entry -> {
-                    String clientUUID = entry.getKey();
-                    ClientModel client = realm.getClientById(clientUUID);
 
-                    if (client != null) {
+        authenticatedClientSessions.forEach((clientUUID, clientSessionId) -> {
-                        AuthenticatedClientSessionModel clientSession = session.sessions()
+            ClientModel client = realm.getClientById(clientUUID);
-                                .getClientSession(this, client, entry.getValue(), isOffline());
+
-                        if (clientSession != null) {
+            if (client != null) {
-                            result.put(clientUUID, clientSession);
+                AuthenticatedClientSessionModel clientSession = session.sessions()
-                        }
+                        .getClientSession(this, client, clientSessionId, isOffline());
-                    } else {
+                if (clientSession != null) {
-                        removedClientUUIDS.add(clientUUID);
+                    result.put(clientUUID, clientSession);
-                    }
+                }
-                });
+            } else {
+                removedClientUUIDS.add(clientUUID);
+            }
+        });
 
         removeAuthenticatedClientSessions(removedClientUUIDS);
 

quarkus/pom.xml

@@ -85,6 +85,11 @@
                 <artifactId>infinispan-commons</artifactId>
                 <version>${infinispan.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.infinispan</groupId>
+                <artifactId>infinispan-server-hotrod</artifactId>
+                <version>${infinispan.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.infinispan</groupId>
                 <artifactId>infinispan-client-hotrod</artifactId>

quarkus/runtime/pom.xml

@@ -341,6 +341,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-client-hotrod</artifactId>

services/src/main/java/org/keycloak/executors/DefaultExecutorsProviderFactory.java

@@ -19,10 +19,10 @@ package org.keycloak.executors;
 
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -189,7 +189,7 @@ public class DefaultExecutorsProviderFactory implements ExecutorsProviderFactory
             // Same like Executors.newCachedThreadPool. Besides that "min" and "max" are configurable
             return new ThreadPoolExecutor(min, max,
                     60L, TimeUnit.SECONDS,
-                    new SynchronousQueue<Runnable>(),
+                    new ArrayBlockingQueue<>(1024),
                     threadFactory);
         }
     }

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

@@ -18,12 +18,13 @@
 
 package org.keycloak.testsuite.model.infinispan;
 
-import org.keycloak.connections.infinispan.DefaultInfinispanConnectionProviderFactory;
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.jboss.logging.Logger;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.KeycloakSession;
 
+import static org.keycloak.connections.infinispan.InfinispanUtil.setTimeServiceToKeycloakTime;
+
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
  */
@@ -47,7 +48,7 @@ public class InfinispanTestUtil {
 
         InfinispanConnectionProvider ispnProvider = session.getProvider(InfinispanConnectionProvider.class);
         EmbeddedCacheManager cacheManager = ispnProvider.getCache(InfinispanConnectionProvider.USER_SESSION_CACHE_NAME).getCacheManager();
-        origTimeService = DefaultInfinispanConnectionProviderFactory.setTimeServiceToKeycloakTime(cacheManager);
+        origTimeService = setTimeServiceToKeycloakTime(cacheManager);
     }
 
     public static void revertTimeService() {

testsuite/integration-arquillian/servers/auth-server/services/testsuite-providers/src/main/java/org/keycloak/testsuite/rest/resource/TestCacheResource.java

@@ -30,7 +30,6 @@ import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
-import javax.ws.rs.QueryParam;
 
 import org.infinispan.Cache;
 import org.infinispan.client.hotrod.RemoteCache;
@@ -40,7 +39,7 @@ import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.testsuite.rest.representation.JGroupsStats;
 import org.keycloak.utils.MediaType;
 import org.infinispan.stream.CacheCollectors;

testsuite/integration-arquillian/tests/base/pom.xml

@@ -143,6 +143,11 @@
             <version>${systemrules.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+        </dependency>
     </dependencies>
     
     <build>

testsuite/integration-arquillian/tests/base/src/test/java/org/keycloak/testsuite/cluster/AuthenticationSessionClusterTest.java

@@ -26,7 +26,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.sessions.infinispan.InfinispanStickySessionEncoderProviderFactory;
-import org.keycloak.models.sessions.infinispan.util.InfinispanUtil;
+import org.keycloak.connections.infinispan.InfinispanUtil;
 import org.keycloak.representations.idm.RealmRepresentation;
 import org.keycloak.sessions.StickySessionEncoderProvider;
 import org.keycloak.testsuite.pages.AppPage;

testsuite/model/pom.xml

@@ -180,6 +180,13 @@
             </properties>
         </profile>
 
+        <profile>
+            <id>jpa+cross-dc-infinispan</id>
+            <properties>
+                <keycloak.model.parameters>CrossDCInfinispan,Jpa</keycloak.model.parameters>
+            </properties>
+        </profile>
+
         <profile>
             <id>jpa+infinispan-sessions-preloading-disabled</id>
             <properties>

testsuite/model/src/main/java/org/keycloak/testsuite/model/HotRodServerRule.java

@@ -0,0 +1,135 @@
+package org.keycloak.testsuite.model;
+
+import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.commons.dataconversion.MediaType;
+import org.infinispan.configuration.cache.BackupConfiguration;
+import org.infinispan.configuration.cache.BackupFailurePolicy;
+import org.infinispan.configuration.cache.CacheMode;
+import org.infinispan.configuration.cache.Configuration;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.jboss.marshalling.commons.GenericJBossMarshaller;
+import org.infinispan.manager.DefaultCacheManager;
+import org.infinispan.server.hotrod.HotRodServer;
+import org.infinispan.server.hotrod.configuration.HotRodServerConfiguration;
+import org.infinispan.server.hotrod.configuration.HotRodServerConfigurationBuilder;
+import org.junit.rules.ExternalResource;
+import org.keycloak.Config;
+
+import java.io.IOException;
+
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.ACTION_TOKEN_CACHE;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.LOGIN_FAILURE_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_CLIENT_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.OFFLINE_USER_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.USER_SESSION_CACHE_NAME;
+import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.WORK_CACHE_NAME;
+
+public class HotRodServerRule extends ExternalResource {
+
+    protected HotRodServer hotRodServer;
+
+    protected HotRodServer hotRodServer2;
+
+    protected RemoteCacheManager remoteCacheManager;
+
+    protected DefaultCacheManager hotRodCacheManager;
+
+    protected DefaultCacheManager hotRodCacheManager2;
+
+    @Override
+    protected void after() {
+        remoteCacheManager.stop();
+    }
+
+    public void createEmbeddedHotRodServer(Config.Scope config) {
+        try {
+            hotRodCacheManager = new DefaultCacheManager("hotrod/hotrod1.xml");
+            hotRodCacheManager2 = new DefaultCacheManager("hotrod/hotrod2.xml");
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+        HotRodServerConfiguration build = new HotRodServerConfigurationBuilder().build();
+        hotRodServer = new HotRodServer();
+        hotRodServer.start(build, hotRodCacheManager);
+
+        HotRodServerConfiguration build2 = new HotRodServerConfigurationBuilder().port(11333).build();
+        hotRodServer2 = new HotRodServer();
+        hotRodServer2.start(build2, hotRodCacheManager2);
+
+        // Create a Hot Rod client
+        org.infinispan.client.hotrod.configuration.ConfigurationBuilder remoteBuilder = new org.infinispan.client.hotrod.configuration.ConfigurationBuilder();
+        remoteBuilder.marshaller(new GenericJBossMarshaller());
+        org.infinispan.client.hotrod.configuration.Configuration cfg = remoteBuilder
+                .addServers(hotRodServer.getHost() + ":" + hotRodServer.getPort() + ";"
+                        + hotRodServer2.getHost() + ":" + hotRodServer2.getPort()).build();
+        remoteCacheManager = new RemoteCacheManager(cfg);
+
+        boolean async = config.getBoolean("async", false);
+
+        // create remote keycloak caches
+        createKeycloakCaches(async, USER_SESSION_CACHE_NAME, OFFLINE_USER_SESSION_CACHE_NAME, CLIENT_SESSION_CACHE_NAME,
+                OFFLINE_CLIENT_SESSION_CACHE_NAME, LOGIN_FAILURE_CACHE_NAME, WORK_CACHE_NAME, ACTION_TOKEN_CACHE);
+
+        getCaches(USER_SESSION_CACHE_NAME, OFFLINE_USER_SESSION_CACHE_NAME, CLIENT_SESSION_CACHE_NAME, OFFLINE_CLIENT_SESSION_CACHE_NAME,
+                LOGIN_FAILURE_CACHE_NAME, WORK_CACHE_NAME, ACTION_TOKEN_CACHE);
+    }
+
+    private void getCaches(String... cache) {
+        for (String c: cache) {
+            hotRodCacheManager.getCache(c, true);
+            hotRodCacheManager2.getCache(c, true);
+        }
+    }
+
+    private void createKeycloakCaches(boolean async, String... cache) {
+        ConfigurationBuilder sessionConfigBuilder1 = createCacheConfigurationBuilder();
+        ConfigurationBuilder sessionConfigBuilder2 = createCacheConfigurationBuilder();
+        sessionConfigBuilder1.clustering().cacheMode(async ? CacheMode.REPL_ASYNC: CacheMode.REPL_SYNC);
+        sessionConfigBuilder2.clustering().cacheMode(async ? CacheMode.REPL_ASYNC: CacheMode.REPL_SYNC);
+
+        sessionConfigBuilder1.sites().addBackup()
+                .site("site-2").backupFailurePolicy(BackupFailurePolicy.IGNORE).strategy(BackupConfiguration.BackupStrategy.SYNC)
+                .replicationTimeout(15000).enabled(true);
+        sessionConfigBuilder2.sites().addBackup()
+                .site("site-1").backupFailurePolicy(BackupFailurePolicy.IGNORE).strategy(BackupConfiguration.BackupStrategy.SYNC)
+                .replicationTimeout(15000).enabled(true);
+
+        Configuration sessionCacheConfiguration1 = sessionConfigBuilder1.build();
+        Configuration sessionCacheConfiguration2 = sessionConfigBuilder2.build();
+        for (String c: cache) {
+            hotRodCacheManager.defineConfiguration(c, sessionCacheConfiguration1);
+            hotRodCacheManager2.defineConfiguration(c, sessionCacheConfiguration2);
+        }
+    }
+
+    public static ConfigurationBuilder createCacheConfigurationBuilder() {
+        ConfigurationBuilder builder = new ConfigurationBuilder();
+
+        // need to force the encoding to application/x-jboss-marshalling to avoid unnecessary conversion of keys/values. See WFLY-14356.
+        builder.encoding().mediaType(MediaType.APPLICATION_JBOSS_MARSHALLING_TYPE);
+
+        return builder;
+    }
+
+    public RemoteCacheManager getRemoteCacheManager() {
+        return remoteCacheManager;
+    }
+
+    public HotRodServer getHotRodServer() {
+        return hotRodServer;
+    }
+
+    public HotRodServer getHotRodServer2() {
+        return hotRodServer2;
+    }
+
+    public DefaultCacheManager getHotRodCacheManager() {
+        return hotRodCacheManager;
+    }
+
+    public DefaultCacheManager getHotRodCacheManager2() {
+        return hotRodCacheManager2;
+    }
+}

testsuite/model/src/main/java/org/keycloak/testsuite/model/KeycloakModelParameters.java

@@ -67,4 +67,7 @@ public class KeycloakModelParameters {
         return base;
     }
 
+    public void beforeSuite(Config cf) {
+
+    }
 }

testsuite/model/src/main/resources/hotrod/hotrod1.xml

@@ -0,0 +1,43 @@
+<infinispan>
+    <jgroups>
+        <!-- Extends the default UDP stack. -->
+        <stack name="xsite" extends="udp">
+            <UDP bind_addr="${jgroups.bind.address,jgroups.udp.address:127.0.0.1}"
+                 bind_port="${jgroups.bind.port,jgroups.udp.port:0}"
+                 mcast_addr="${jgroups.udp.mcast_addr,jgroups.mcast_addr:228.6.7.10}"
+                 mcast_port="${jgroups.udp.mcast_port,jgroups.mcast_port:46655}"
+                 tos="0"
+                 ucast_send_buf_size="1m"
+                 mcast_send_buf_size="1m"
+                 ucast_recv_buf_size="20m"
+                 mcast_recv_buf_size="25m"
+                 ip_ttl="${jgroups.ip_ttl:2}"
+                 thread_naming_pattern="pl"
+                 enable_diagnostics="false"
+                 bundler_type="no-bundler"
+                 max_bundle_size="8500"
+
+                 thread_pool.min_threads="${jgroups.thread_pool.min_threads:0}"
+                 thread_pool.max_threads="${jgroups.thread_pool.max_threads:200}"
+                 thread_pool.keep_alive_time="60000"
+            />
+
+            <!-- Adds RELAY2 for cross-site replication. -->
+            <!-- Names the local site as site-1. -->
+            <!-- Specifies 1000 nodes as the maximum number of site masters. -->
+            <relay.RELAY2 site="site-1" xmlns="urn:org:jgroups" max_site_masters="1000"/>
+            <!-- Uses the default UDP stack for inter-cluster communication. -->
+            <!-- Names all sites that act as backup locations. -->
+            <remote-sites default-stack="udp">
+                <remote-site name="site-1"/>
+                <remote-site name="site-2"/>
+            </remote-sites>
+        </stack>
+    </jgroups>
+    <cache-container name="site-1" statistics="true">
+        <!-- Use the "xsite" stack for cluster transport. -->
+        <transport cluster="site-1" stack="xsite"/>
+
+        <serialization marshaller="org.infinispan.jboss.marshalling.commons.GenericJBossMarshaller"/>
+    </cache-container>
+</infinispan>

testsuite/model/src/main/resources/hotrod/hotrod2.xml

@@ -0,0 +1,42 @@
+<infinispan>
+    <jgroups>
+        <!-- Extends the default UDP stack. -->
+        <stack name="xsite" extends="udp">
+            <UDP bind_addr="${jgroups.bind.address,jgroups.udp.address:127.0.0.1}"
+                 bind_port="${jgroups.bind.port,jgroups.udp.port:0}"
+                 mcast_addr="${jgroups.udp.mcast_addr,jgroups.mcast_addr:228.6.7.11}"
+                 mcast_port="${jgroups.udp.mcast_port,jgroups.mcast_port:46655}"
+                 tos="0"
+                 ucast_send_buf_size="1m"
+                 mcast_send_buf_size="1m"
+                 ucast_recv_buf_size="20m"
+                 mcast_recv_buf_size="25m"
+                 ip_ttl="${jgroups.ip_ttl:2}"
+                 thread_naming_pattern="pl"
+                 enable_diagnostics="false"
+                 bundler_type="no-bundler"
+                 max_bundle_size="8500"
+
+                 thread_pool.min_threads="${jgroups.thread_pool.min_threads:0}"
+                 thread_pool.max_threads="${jgroups.thread_pool.max_threads:200}"
+                 thread_pool.keep_alive_time="60000"
+            />
+            <!-- Adds RELAY2 for cross-site replication. -->
+            <!-- Names the local site as site-2. -->
+            <!-- Specifies 1000 nodes as the maximum number of site masters. -->
+            <relay.RELAY2 site="site-2" xmlns="urn:org:jgroups" max_site_masters="1000"/>
+            <!-- Uses the default UDP stack for inter-cluster communication. -->
+            <!-- Names all sites that act as backup locations. -->
+            <remote-sites default-stack="udp">
+                <remote-site name="site-1"/>
+                <remote-site name="site-2"/>
+            </remote-sites>
+        </stack>
+    </jgroups>
+    <cache-container name="site-2" statistics="true">
+        <!-- Use the "xsite" stack for cluster transport. -->
+        <transport cluster="site-2" stack="xsite"/>
+
+        <serialization marshaller="org.infinispan.jboss.marshalling.commons.GenericJBossMarshaller"/>
+    </cache-container>
+</infinispan>

testsuite/model/src/test/java/org/keycloak/testsuite/model/KeycloakModelTest.java

@@ -242,6 +242,12 @@ public abstract class KeycloakModelTest {
           )
           .collect(Collectors.toList());
 
+
+        for (KeycloakModelParameters kmp : KeycloakModelTest.MODEL_PARAMETERS) {
+            kmp.beforeSuite(CONFIG);
+        }
+
+        // TODO move to a class rule
         reinitializeKeycloakSessionFactory();
         DEFAULT_FACTORY = getFactory();
     }

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

@@ -16,10 +16,15 @@
  */
 package org.keycloak.testsuite.model.infinispan;
 
+import org.hamcrest.Matchers;
+import org.infinispan.Cache;
+import org.junit.Assume;
+import org.junit.Test;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.cache.infinispan.events.AuthenticationSessionAuthNoteUpdateEvent;
 import org.keycloak.testsuite.model.KeycloakModelTest;
 import org.keycloak.testsuite.model.RequireProvider;
+
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -31,11 +36,9 @@ import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import org.hamcrest.Matchers;
+
-import org.infinispan.Cache;
-import org.junit.Assume;
-import org.junit.Test;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.Assume.assumeThat;
@@ -65,7 +68,9 @@ public class CacheExpirationTest extends KeycloakModelTest {
 
         assumeThat("jmap output format unsupported", getNumberOfInstancesOfClass(AuthenticationSessionAuthNoteUpdateEvent.class), notNullValue());
 
-        assertThat(getNumberOfInstancesOfClass(AuthenticationSessionAuthNoteUpdateEvent.class), is(2));
+        // Infinispan server is decoding the client request before processing the request at the cache level,
+        // therefore there are sometimes three instances of AuthenticationSessionAuthNoteUpdateEvent class in the memory
+        assertThat(getNumberOfInstancesOfClass(AuthenticationSessionAuthNoteUpdateEvent.class), greaterThanOrEqualTo(2));
 
         AtomicInteger maxCountOfInstances = new AtomicInteger();
         AtomicInteger minCountOfInstances = new AtomicInteger(100);
@@ -80,6 +85,17 @@ public class CacheExpirationTest extends KeycloakModelTest {
                 cache.keySet().forEach(s -> {});
             });
             log.debug("Cluster joined");
+
+            // access the items in the local cache in the different site (site-2) in order to fetch them from the remote cache
+            String site = CONFIG.scope("connectionsInfinispan", "default").get("siteName");
+            if ("site-2".equals(site)) {
+                inComittedTransaction(session -> {
+                    InfinispanConnectionProvider provider = session.getProvider(InfinispanConnectionProvider.class);
+                    Cache<String, Object> cache = provider.getCache(InfinispanConnectionProvider.WORK_CACHE_NAME);
+                    cache.get("1-2");
+                    cache.get("1-2-3");
+                });
+            }
             int c = getNumberOfInstancesOfClass(AuthenticationSessionAuthNoteUpdateEvent.class);
             maxCountOfInstances.getAndAccumulate(c, Integer::max);
             assumeThat("Seems we're running on a way too slow a computer", System.currentTimeMillis() - putTime.get(), Matchers.lessThan(20000L));

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

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2021 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.testsuite.model.parameters;
+
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+import org.keycloak.testsuite.model.Config;
+import org.keycloak.testsuite.model.KeycloakModelParameters;
+import org.keycloak.testsuite.model.HotRodServerRule;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+/**
+ * @author <a href="mailto:mkanis@redhat.com">Martin Kanis</a>
+ */
+public class CrossDCInfinispan extends KeycloakModelParameters {
+
+    private final HotRodServerRule hotRodServerRule = new HotRodServerRule();
+
+    private static final AtomicInteger NODE_COUNTER = new AtomicInteger();
+
+    private static final String SITE_1_MCAST_ADDR = "228.5.6.7";
+
+    private static final String SITE_2_MCAST_ADDR = "228.6.7.8";
+
+    private final Object lock = new Object();
+
+    @Override
+    public void updateConfig(Config cf) {
+        synchronized (lock) {
+            NODE_COUNTER.incrementAndGet();
+            cf.spi("connectionsInfinispan")
+                    .provider("default")
+                    .config("embedded", "true")
+                    .config("clustered", "true")
+                    .config("remoteStoreEnabled", "true")
+                    .config("useKeycloakTimeService", "true")
+                    .config("remoteStoreSecurityEnabled", "false")
+                    .config("nodeName", "node-" + NODE_COUNTER.get())
+                    .config("siteName", siteName(NODE_COUNTER.get()))
+                    .config("remoteStorePort", siteName(NODE_COUNTER.get()).equals("site-2") ? "11333" : "11222")
+                    .config("jgroupsUdpMcastAddr", mcastAddr(NODE_COUNTER.get()));
+        }
+    }
+
+    public CrossDCInfinispan() {
+        super(Infinispan.ALLOWED_SPIS, Infinispan.ALLOWED_FACTORIES);
+    }
+
+    @Override
+    public void beforeSuite(Config cf) {
+        hotRodServerRule.createEmbeddedHotRodServer(cf.scope("connectionsInfinispan", "default"));
+    }
+
+    private static String siteName(int node) {
+        return "site-" + (node % 2 == 0 ? 2 : 1);
+    }
+
+    private static String mcastAddr(int node) {
+        return (node % 2 == 0) ? SITE_2_MCAST_ADDR : SITE_1_MCAST_ADDR;
+    }
+
+    @Override
+    public <T> Stream<T> getParameters(Class<T> clazz) {
+        if (HotRodServerRule.class.isAssignableFrom(clazz)) {
+            return Stream.of((T) hotRodServerRule);
+        } else {
+            return Stream.empty();
+        }
+    }
+
+    @Override
+    public Statement classRule(Statement base, Description description) {
+        return hotRodServerRule.apply(base, description);
+    }
+}

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

@@ -45,6 +45,7 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.hamcrest.Matchers;
 import org.junit.Test;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 

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

@@ -17,10 +17,11 @@
 
 package org.keycloak.testsuite.model.session;
 
+import org.infinispan.Cache;
+import org.infinispan.client.hotrod.RemoteCache;
 import org.junit.Assert;
 import org.junit.Test;
-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;
@@ -32,16 +33,23 @@ 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.InfinispanUserSessionProviderFactory;
-import org.keycloak.services.managers.UserSessionManager;
 
+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.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;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
@@ -92,10 +100,8 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
 
     @Test
     public void testUserSessionInitializer() {
-        String[] origSessionIds = createSessionsInPersisterOnly();
+        UserSessionModel[] origSessionIds = createSessionsInPersisterOnly();
-        int started = Time.currentTime();
+        int started = origSessionIds[0].getStarted();
-
-        reinitializeKeycloakSessionFactory();
 
         inComittedTransaction(session -> {
             RealmModel realm = session.realms().getRealm(realmId);
@@ -109,18 +115,17 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
 
             List<UserSessionModel> loadedSessions = session.sessions().getOfflineUserSessionsStream(realm, testApp, 0, 10)
                     .collect(Collectors.toList());
-            UserSessionPersisterProviderTest.assertSessions(loadedSessions, origSessionIds);
 
-            assertSessionLoaded(loadedSessions, origSessionIds[0], session.users().getUserByUsername(realm, "user1"), "127.0.0.1", started, started, "test-app", "third-party");
+            assertSessionLoaded(loadedSessions, origSessionIds[0].getId(), session.users().getUserByUsername(realm, "user1"), "127.0.0.1", started, started, "test-app", "third-party");
-            assertSessionLoaded(loadedSessions, origSessionIds[1], session.users().getUserByUsername(realm, "user1"), "127.0.0.2", started, started, "test-app");
+            assertSessionLoaded(loadedSessions, origSessionIds[1].getId(), session.users().getUserByUsername(realm, "user1"), "127.0.0.2", started, started, "test-app");
-            assertSessionLoaded(loadedSessions, origSessionIds[2], session.users().getUserByUsername(realm, "user2"), "127.0.0.3", started, started, "test-app");
+            assertSessionLoaded(loadedSessions, origSessionIds[2].getId(), session.users().getUserByUsername(realm, "user2"), "127.0.0.3", started, started, "test-app");
         });
     }
 
     @Test
     public void testUserSessionInitializerWithDeletingClient() {
-        String[] origSessionIds = createSessionsInPersisterOnly();
+        UserSessionModel[] origSessionIds = createSessionsInPersisterOnly();
-        int started = Time.currentTime();
+        int started = origSessionIds[0].getStarted();
 
         inComittedTransaction(session -> {
             RealmModel realm = session.realms().getRealm(realmId);
@@ -130,8 +135,6 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
             realm.removeClient(testApp.getId());
         });
 
-        reinitializeKeycloakSessionFactory();
-
         inComittedTransaction(session -> {
             RealmModel realm = session.realms().getRealm(realmId);
 
@@ -143,7 +146,7 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
                     .collect(Collectors.toList());
 
             assertThat("Size of loaded Sessions", loadedSessions.size(), is(1));
-            assertSessionLoaded(loadedSessions, origSessionIds[0], session.users().getUserByUsername(realm, "user1"), "127.0.0.1", started, started, "third-party");
+            assertSessionLoaded(loadedSessions, origSessionIds[0].getId(), session.users().getUserByUsername(realm, "user1"), "127.0.0.1", started, started, "third-party");
 
             // Revert client
             realm.addClient("test-app");
@@ -151,37 +154,68 @@ public class UserSessionInitializerTest extends KeycloakModelTest {
 
     }
 
-    // Create sessions in persister + infinispan, but then delete them from infinispan cache. This is to allow later testing of initializer. Return the list of "origSessions"
+    @Test
-    private String[] createSessionsInPersisterOnly() {
+    @RequireProvider(value = UserSessionProvider.class, only = InfinispanUserSessionProviderFactory.PROVIDER_ID)
+    public void testUserSessionPropagationBetweenSites() throws InterruptedException {
+        AtomicInteger index = new AtomicInteger();
+        AtomicReference<String> userSessionId = new AtomicReference<>();
+        AtomicReference<List<Boolean>> containsSession = new AtomicReference<>(new LinkedList<>());
+
+        Object lock = new Object();
+
+        Optional<HotRodServerRule> hotRodServer = getParameters(HotRodServerRule.class).findFirst();
+
+        inIndependentFactories(4, 300, () -> {
+            synchronized (lock) {
+                if (index.incrementAndGet() == 1) {
+                    // create a user session in the first node
+                    UserSessionModel userSessionModel = withRealm(realmId, (session, realm) -> {
+                        final UserModel user = session.users().getUserByUsername(realm, "user1");
+                        return session.sessions().createUserSession(realm, user, "un1", "ip1", "auth", false, null, null);
+                    });
+                    userSessionId.set(userSessionModel.getId());
+                } else {
+                    // 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);
+                        containsSession.get().add(localSessions.containsKey(userSessionId.get()));
+
+                        if (hotRodServer.isPresent()) {
+                            RemoteCache<String, Object> remoteSessions = provider.getRemoteCache(USER_SESSION_CACHE_NAME);
+                            containsSession.get().add(remoteSessions.containsKey(userSessionId.get()));
+                        }
+                    });
+                }
+            }
+        });
+
+        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;
+        assertThat(containsSession.get().size(), is(size));
+    }
+
+    // Create sessions in persister + infinispan, but then delete them from infinispan cache by reinitializing keycloak session factory
+    private UserSessionModel[] createSessionsInPersisterOnly() {
         UserSessionModel[] origSessions = inComittedTransaction(session -> { return UserSessionPersisterProviderTest.createSessions(session, realmId); });
-        String[] res = new String[origSessions.length];
+        UserSessionModel[] res = new UserSessionModel[origSessions.length];
-
-        inComittedTransaction(session -> {
-            RealmModel realm = session.realms().getRealm(realmId);
-            UserSessionManager sessionManager = new UserSessionManager(session);
 
+        withRealm(realmId, (session, realm) -> {
             int i = 0;
             for (UserSessionModel origSession : origSessions) {
                 UserSessionModel userSession = session.sessions().getUserSession(realm, origSession.getId());
-                for (AuthenticatedClientSessionModel clientSession : userSession.getAuthenticatedClientSessions().values()) {
+                UserSessionModel offlineUserSession = session.sessions().createOfflineUserSession(userSession);
-                    sessionManager.createOrUpdateOfflineSession(clientSession, userSession);
+                userSession.getAuthenticatedClientSessions().values().forEach(clientSession ->
-                }
+                        session.sessions().createOfflineClientSession(clientSession, offlineUserSession));
-                String cs = userSession.getNote(UserSessionModel.CORRESPONDING_SESSION_ID);
+
-                res[i] = cs == null ? userSession.getId() : cs;
+                res[i++] = offlineUserSession;
-                i++;
             }
+            return null;
         });
 
-        inComittedTransaction(session -> {
+        reinitializeKeycloakSessionFactory();
-            RealmModel realm = session.realms().getRealm(realmId);
-
-            // Delete local user cache (persisted sessions are still kept)
-            UserSessionProvider provider = session.getProvider(UserSessionProvider.class);
-            if (provider instanceof InfinispanUserSessionProvider) {
-                // Remove in-memory representation of the offline sessions
-                ((InfinispanUserSessionProvider) provider).removeLocalUserSessions(realm.getId(), true);
-            }
-        });
 
         return res;
     }

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

@@ -29,20 +29,19 @@ import org.keycloak.models.UserModel;
 import org.keycloak.models.UserProvider;
 import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.UserSessionProvider;
+import org.keycloak.models.map.userSession.MapUserSessionProvider;
 import org.keycloak.models.session.UserSessionPersisterProvider;
 import org.keycloak.models.sessions.infinispan.changes.sessions.PersisterLastSessionRefreshStoreFactory;
 import org.keycloak.models.utils.ResetTimeOffsetEvent;
+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 java.util.Collection;
-import java.util.LinkedList;
 import java.util.List;
-import java.util.UUID;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
-import org.keycloak.testsuite.model.KeycloakModelTest;
-import org.keycloak.testsuite.model.RequireProvider;
 import static org.keycloak.testsuite.model.session.UserSessionPersisterProviderTest.createClients;
 import static org.keycloak.testsuite.model.session.UserSessionPersisterProviderTest.createSessions;
 
@@ -65,6 +64,7 @@ public class UserSessionProviderModelTest extends KeycloakModelTest {
         realm.setDefaultRole(s.roles().addRealmRole(realm, Constants.DEFAULT_ROLES_ROLE_PREFIX + "-" + realm.getName()));
         realm.setSsoSessionIdleTimeout(1800);
         realm.setSsoSessionMaxLifespan(36000);
+        realm.setClientSessionIdleTimeout(500);
         this.realmId = realm.getId();
         this.kcSession = s;
 
@@ -137,14 +137,15 @@ public class UserSessionProviderModelTest extends KeycloakModelTest {
             InfinispanTestUtil.setTestingTimeService(kcSession);
         }
 
-        AtomicReference<List<String>> clientSessionIds = new AtomicReference<>();
-
         try {
             UserSessionModel[] origSessions = inComittedTransaction(session -> {
                 // create some user and client sessions
                 return createSessions(session, realmId);
             });
 
+            AtomicReference<List<String>> clientSessionIds = new AtomicReference<>();
+            clientSessionIds.set(origSessions[0].getAuthenticatedClientSessions().values().stream().map(AuthenticatedClientSessionModel::getId).collect(Collectors.toList()));
+
             inComittedTransaction(session -> {
                 RealmModel realm = session.realms().getRealm(realmId);
 
@@ -160,19 +161,22 @@ public class UserSessionProviderModelTest extends KeycloakModelTest {
                 Assert.assertEquals(origSessions[1], userSession);
             });
 
+
+            // not possible to expire client session without expiring user sessions with time offset in map storage because
+            // expiration in map storage takes min of (clientSessionIdleExpiration, ssoSessionIdleTimeout)
             inComittedTransaction(session -> {
-                RealmModel realm = session.realms().getRealm(realmId);
+                if (session.getProvider(UserSessionProvider.class) instanceof MapUserSessionProvider) {
+                    RealmModel realm = session.realms().getRealm(realmId);
 
-                UserSessionModel userSession = session.sessions().getUserSession(realm, origSessions[0].getId());
+                    UserSessionModel userSession = session.sessions().getUserSession(realm, origSessions[0].getId());
 
-                Collection<AuthenticatedClientSessionModel> values = userSession.getAuthenticatedClientSessions().values();
+                    userSession.getAuthenticatedClientSessions().values().stream().forEach(clientSession -> {
-                List<String> clientSessions = new LinkedList<>();
+                        // expire client sessions
-                values.stream().forEach(clientSession -> {
+                        clientSession.setTimestamp(1);
-                    // expire client sessions
+                    });
-                    clientSession.setTimestamp(1);
+                } else {
-                    clientSessions.add(clientSession.getId());
+                    Time.setOffset(1000);
-                });
+                }
-                clientSessionIds.set(clientSessions);
             });
 
             inComittedTransaction(session -> {
@@ -183,8 +187,10 @@ public class UserSessionProviderModelTest extends KeycloakModelTest {
                 Assert.assertEquals(origSessions[0], userSession);
 
                 // assert the client sessions are expired
-                clientSessionIds.get().forEach(clientSessionId ->
+                clientSessionIds.get().forEach(clientSessionId -> {
-                        Assert.assertNull(session.sessions().getClientSession(userSession, realm.getClientByClientId("test-app"), UUID.fromString(clientSessionId), false)));
+                    Assert.assertNull(session.sessions().getClientSession(userSession, realm.getClientByClientId("test-app"), clientSessionId, false));
+                    Assert.assertNull(session.sessions().getClientSession(userSession, realm.getClientByClientId("third-party"), clientSessionId, false));
+                });
             });
         } finally {
             Time.setOffset(0);

testsuite/model/src/test/resources/log4j.properties

@@ -48,6 +48,18 @@ log4j.logger.org.keycloak.connections.jpa.DefaultJpaConnectionProviderFactory=de
 #log4j.logger.org.keycloak.STACK_TRACE=trace
 
 #log4j.logger.org.keycloak.models.sessions.infinispan=trace
+keycloak.infinispan.logging.level=info
+log4j.logger.org.keycloak.cluster.infinispan=${keycloak.infinispan.logging.level}
+log4j.logger.org.keycloak.connections.infinispan=${keycloak.infinispan.logging.level}
+log4j.logger.org.keycloak.keys.infinispan=${keycloak.infinispan.logging.level}
+log4j.logger.org.keycloak.models.cache.infinispan=${keycloak.infinispan.logging.level}
+log4j.logger.org.keycloak.models.sessions.infinispan=${keycloak.infinispan.logging.level}
+
+log4j.logger.org.infinispan.server.hotrod=info
+log4j.logger.org.infinispan.client.hotrod.impl=info
+log4j.logger.org.infinispan.client.hotrod.event.impl=info
+
+log4j.logger.org.keycloak.executors=info
 
 #log4j.logger.org.infinispan.expiration.impl.ClusterExpirationManager=trace
 #