KEYCLOAK-4189 - Cross DC testing

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

@@ -154,7 +154,8 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
 
         if (clustered) {
             String nodeName = config.get("nodeName", System.getProperty(InfinispanConnectionProvider.JBOSS_NODE_NAME));
-            configureTransport(gcb, nodeName);
+            String jgroupsUdpMcastAddr = config.get("jgroupsUdpMcastAddr", System.getProperty(InfinispanConnectionProvider.JGROUPS_UDP_MCAST_ADDR));
+            configureTransport(gcb, nodeName, jgroupsUdpMcastAddr);
         }
         gcb.globalJmxStatistics().allowDuplicateDomains(allowDuplicateJMXDomains);
 
@@ -317,24 +318,40 @@ public class DefaultInfinispanConnectionProviderFactory implements InfinispanCon
         return cb.build();
     }
 
-    protected void configureTransport(GlobalConfigurationBuilder gcb, String nodeName) {
+    private static final Object CHANNEL_INIT_SYNCHRONIZER = new Object();
+
+    protected void configureTransport(GlobalConfigurationBuilder gcb, String nodeName, String jgroupsUdpMcastAddr) {
         if (nodeName == null) {
             gcb.transport().defaultTransport();
         } else {
             FileLookup fileLookup = FileLookupFactory.newInstance();
 
-            try {
-                // Compatibility with Wildfly
-                JChannel channel = new JChannel(fileLookup.lookupFileLocation("default-configs/default-jgroups-udp.xml", this.getClass().getClassLoader()));
-                channel.setName(nodeName);
-                JGroupsTransport transport = new JGroupsTransport(channel);
+            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 {
+                    // Compatibility with Wildfly
+                    JChannel channel = new JChannel(fileLookup.lookupFileLocation("default-configs/default-jgroups-udp.xml", this.getClass().getClassLoader()));
+                    channel.setName(nodeName);
+                    JGroupsTransport transport = new JGroupsTransport(channel);
 
-                gcb.transport().nodeName(nodeName);
-                gcb.transport().transport(transport);
+                    gcb.transport().nodeName(nodeName);
+                    gcb.transport().transport(transport);
 
-                logger.infof("Configured jgroups transport with the channel name: %s", nodeName);
-            } catch (Exception e) {
-                throw new RuntimeException(e);
+                    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);
+                    }
+                }
             }
         }
     }

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

@@ -53,6 +53,7 @@ public interface InfinispanConnectionProvider extends Provider {
 
     // System property used on Wildfly to identify distributedCache address and sticky session route
     String JBOSS_NODE_NAME = "jboss.node.name";
+    String JGROUPS_UDP_MCAST_ADDR = "jgroups.udp.mcast_addr";
 
 
     <K, V> Cache<K, V> getCache(String name);

model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/CacheManager.java

@@ -220,7 +220,7 @@ public abstract class CacheManager {
 
         addInvalidationsFromEvent(event, invalidations);
 
-        getLogger().debugf("Invalidating %d cache items after received event %s", invalidations.size(), event);
+        getLogger().debugf("[%s] Invalidating %d cache items after received event %s", cache.getCacheManager().getAddress(), invalidations.size(), event);
 
         for (String invalidation : invalidations) {
             invalidateObject(invalidation);

model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/events/AddInvalidatedActionTokenEvent.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.keycloak.models.cache.infinispan;
+package org.keycloak.models.cache.infinispan.events;
 
 import org.keycloak.cluster.ClusterEvent;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenReducedKey;

model/infinispan/src/main/java/org/keycloak/models/cache/infinispan/events/RemoveActionTokensSpecificEvent.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.keycloak.models.cache.infinispan;
+package org.keycloak.models.cache.infinispan.events;
 
 import org.keycloak.cluster.ClusterEvent;
 

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

@@ -19,8 +19,8 @@ package org.keycloak.models.sessions.infinispan;
 import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.models.*;
 
-import org.keycloak.models.cache.infinispan.AddInvalidatedActionTokenEvent;
-import org.keycloak.models.cache.infinispan.RemoveActionTokensSpecificEvent;
+import org.keycloak.models.cache.infinispan.events.AddInvalidatedActionTokenEvent;
+import org.keycloak.models.cache.infinispan.events.RemoveActionTokensSpecificEvent;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenReducedKey;
 import java.util.*;
@@ -58,7 +58,12 @@ public class InfinispanActionTokenStoreProvider implements ActionTokenStoreProvi
         ActionTokenValueEntity tokenValue = new ActionTokenValueEntity(notes);
 
         ClusterProvider cluster = session.getProvider(ClusterProvider.class);
-        this.tx.notify(cluster, InfinispanActionTokenStoreProviderFactory.ACTION_TOKEN_EVENTS, new AddInvalidatedActionTokenEvent(tokenKey, key.getExpiration(), tokenValue), false);
+        AddInvalidatedActionTokenEvent event = new AddInvalidatedActionTokenEvent(tokenKey, key.getExpiration(), tokenValue);
+        this.tx.notify(cluster, generateActionTokenEventId(), event, false);
+    }
+
+    private static String generateActionTokenEventId() {
+        return InfinispanActionTokenStoreProviderFactory.ACTION_TOKEN_EVENTS + "/" + UUID.randomUUID();
     }
 
     @Override
@@ -93,6 +98,6 @@ public class InfinispanActionTokenStoreProvider implements ActionTokenStoreProvi
         }
 
         ClusterProvider cluster = session.getProvider(ClusterProvider.class);
-        this.tx.notify(cluster, InfinispanActionTokenStoreProviderFactory.ACTION_TOKEN_EVENTS, new RemoveActionTokensSpecificEvent(userId, actionId), false);
+        this.tx.notify(cluster, generateActionTokenEventId(), new RemoveActionTokensSpecificEvent(userId, actionId), false);
     }
 }

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

@@ -23,14 +23,16 @@ import org.keycloak.common.util.Time;
 import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
 import org.keycloak.models.*;
 
-import org.keycloak.models.cache.infinispan.AddInvalidatedActionTokenEvent;
-import org.keycloak.models.cache.infinispan.RemoveActionTokensSpecificEvent;
+import org.keycloak.models.cache.infinispan.events.AddInvalidatedActionTokenEvent;
+import org.keycloak.models.cache.infinispan.events.RemoveActionTokensSpecificEvent;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenReducedKey;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import org.infinispan.Cache;
 import org.infinispan.context.Flag;
+import org.infinispan.remoting.transport.Address;
+import org.jboss.logging.Logger;
 
 /**
  *
@@ -38,6 +40,10 @@ import org.infinispan.context.Flag;
  */
 public class InfinispanActionTokenStoreProviderFactory implements ActionTokenStoreProviderFactory {
 
+    private static final Logger LOG = Logger.getLogger(InfinispanActionTokenStoreProviderFactory.class);
+
+    private volatile Cache<ActionTokenReducedKey, ActionTokenValueEntity> actionTokenCache;
+
     public static final String ACTION_TOKEN_EVENTS = "ACTION_TOKEN_EVENTS";
 
     /**
@@ -49,34 +55,7 @@ public class InfinispanActionTokenStoreProviderFactory implements ActionTokenSto
 
     @Override
     public ActionTokenStoreProvider create(KeycloakSession session) {
-        InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
-        Cache<ActionTokenReducedKey, ActionTokenValueEntity> actionTokenCache = connections.getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE);
-
-        ClusterProvider cluster = session.getProvider(ClusterProvider.class);
-
-        cluster.registerListener(ACTION_TOKEN_EVENTS, event -> {
-            if (event instanceof RemoveActionTokensSpecificEvent) {
-                RemoveActionTokensSpecificEvent e = (RemoveActionTokensSpecificEvent) event;
-
-                actionTokenCache
-                  .getAdvancedCache()
-                  .withFlags(Flag.CACHE_MODE_LOCAL, Flag.SKIP_CACHE_LOAD)
-                  .keySet()
-                  .stream()
-                  .filter(k -> Objects.equals(k.getUserId(), e.getUserId()) && Objects.equals(k.getActionId(), e.getActionId()))
-                  .forEach(actionTokenCache::remove);
-            } else if (event instanceof AddInvalidatedActionTokenEvent) {
-                AddInvalidatedActionTokenEvent e = (AddInvalidatedActionTokenEvent) event;
-
-                if (e.getExpirationInSecs() == DEFAULT_CACHE_EXPIRATION) {
-                    actionTokenCache.put(e.getKey(), e.getTokenValue());
-                } else {
-                    actionTokenCache.put(e.getKey(), e.getTokenValue(), e.getExpirationInSecs() - Time.currentTime(), TimeUnit.SECONDS);
-                }
-            }
-        });
-
-        return new InfinispanActionTokenStoreProvider(session, actionTokenCache);
+        return new InfinispanActionTokenStoreProvider(session, this.actionTokenCache);
     }
 
     @Override
@@ -84,8 +63,57 @@ public class InfinispanActionTokenStoreProviderFactory implements ActionTokenSto
         this.config = config;
     }
 
+    private static Cache<ActionTokenReducedKey, ActionTokenValueEntity> initActionTokenCache(KeycloakSession session) {
+        InfinispanConnectionProvider connections = session.getProvider(InfinispanConnectionProvider.class);
+        Cache<ActionTokenReducedKey, ActionTokenValueEntity> cache = connections.getCache(InfinispanConnectionProvider.ACTION_TOKEN_CACHE);
+        final Address cacheAddress = cache.getCacheManager().getAddress();
+
+        ClusterProvider cluster = session.getProvider(ClusterProvider.class);
+
+        cluster.registerListener(ClusterProvider.ALL, event -> {
+            if (event instanceof RemoveActionTokensSpecificEvent) {
+                RemoveActionTokensSpecificEvent e = (RemoveActionTokensSpecificEvent) event;
+
+                LOG.debugf("[%s] Removing token invalidation for user+action: userId=%s, actionId=%s", cacheAddress, e.getUserId(), e.getActionId());
+
+                cache
+                  .getAdvancedCache()
+                  .withFlags(Flag.CACHE_MODE_LOCAL, Flag.SKIP_CACHE_LOAD)
+                  .keySet()
+                  .stream()
+                  .filter(k -> Objects.equals(k.getUserId(), e.getUserId()) && Objects.equals(k.getActionId(), e.getActionId()))
+                  .forEach(cache::remove);
+            } else if (event instanceof AddInvalidatedActionTokenEvent) {
+                AddInvalidatedActionTokenEvent e = (AddInvalidatedActionTokenEvent) event;
+
+                LOG.debugf("[%s] Invalidating token %s", cacheAddress, e.getKey());
+                if (e.getExpirationInSecs() == DEFAULT_CACHE_EXPIRATION) {
+                    cache.put(e.getKey(), e.getTokenValue());
+                } else {
+                    cache.put(e.getKey(), e.getTokenValue(), e.getExpirationInSecs() - Time.currentTime(), TimeUnit.SECONDS);
+                }
+            }
+        });
+
+        LOG.debugf("[%s] Registered cluster listeners", cacheAddress);
+
+        return cache;
+    }
+
     @Override
     public void postInit(KeycloakSessionFactory factory) {
+        Cache<ActionTokenReducedKey, ActionTokenValueEntity> cache = this.actionTokenCache;
+
+        // It is necessary to put the cache initialization here, otherwise the cache would be initialized lazily, that
+        // means also listeners will start only after first cache initialization - that would be too late
+        if (cache == null) {
+            synchronized (this) {
+                cache = this.actionTokenCache;
+                if (cache == null) {
+                    this.actionTokenCache = initActionTokenCache(factory.create());
+                }
+            }
+        }
     }
 
     @Override

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

@@ -92,7 +92,7 @@ public class InfinispanAuthenticationSessionProviderFactory implements Authentic
                     ClusterProvider cluster = session.getProvider(ClusterProvider.class);
                     cluster.registerListener(AUTHENTICATION_SESSION_EVENTS, this::updateAuthNotes);
 
-                    log.debug("Registered cluster listeners");
+                    log.debugf("[%s] Registered cluster listeners", authSessionsCache.getCacheManager().getAddress());
                 }
             }
         }

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

@@ -81,6 +81,11 @@ public class ActionTokenReducedKey implements Serializable {
           && Objects.equals(this.actionVerificationNonce, other.getActionVerificationNonce());
     }
 
+    @Override
+    public String toString() {
+        return "userId=" + userId + ", actionId=" + actionId + ", actionVerificationNonce=" + actionVerificationNonce;
+    }
+
     public static class ExternalizerImpl implements Externalizer<ActionTokenReducedKey> {
 
         @Override

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

@@ -53,7 +53,7 @@ public class ActionTokenValueEntity implements ActionTokenValueModel {
         public void writeObject(ObjectOutput output, ActionTokenValueEntity t) throws IOException {
             output.writeByte(VERSION_1);
 
-            output.writeBoolean(! t.notes.isEmpty());
+            output.writeBoolean(t.notes.isEmpty());
             if (! t.notes.isEmpty()) {
                 output.writeObject(t.notes);
             }