Merge pull request keycloak#4640 from mposolda/crossdc

KEYCLOAK-5716 KEYCLOAK-5738 Avoid infinispan deadlock. Ensure code-to…

testsuite/integration-arquillian/tests/base/src/main/java/org/keycloak/testsuite/Retry.java → common/src/main/java/org/keycloak/common/util/Retry.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 Red Hat, Inc. and/or its affiliates
+ * Copyright 2017 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");
@@ -15,9 +15,7 @@
  * limitations under the License.
  */
 
-package org.keycloak.testsuite;
-
-import java.util.function.Supplier;
+package org.keycloak.common.util;
 
 /**
  * @author <a href="mailto:sthorger@redhat.com">Stian Thorgersen</a>
@@ -44,7 +42,9 @@ public static int execute(Runnable runnable, int attemptsCount, long intervalMil
                 executionIndex++;
                 if (attemptsCount > 0) {
                     try {
-                        Thread.sleep(intervalMillis);
+                        if (intervalMillis > 0) {
+                            Thread.sleep(intervalMillis);
+                        }
                     } catch (InterruptedException ie) {
                         ie.addSuppressed(e);
                         throw new RuntimeException(ie);
@@ -73,7 +73,9 @@ public static <T> T call(Supplier<T> supplier, int attemptsCount, long intervalM
                 attemptsCount--;
                 if (attemptsCount > 0) {
                     try {
-                        Thread.sleep(intervalMillis);
+                        if (intervalMillis > 0) {
+                            Thread.sleep(intervalMillis);
+                        }
                     } catch (InterruptedException ie) {
                         ie.addSuppressed(e);
                         throw new RuntimeException(ie);
@@ -85,4 +87,19 @@ public static <T> T call(Supplier<T> supplier, int attemptsCount, long intervalM
         }
     }
 
+
+    /**
+     * Needed here just because java.util.function.Supplier defined from Java 8
+     */
+    public interface Supplier<T> {
+
+        /**
+         * Gets a result.
+         *
+         * @return a result
+         */
+        T get();
+    }
+
+
 }

misc/CrossDataCenter.md

@@ -95,15 +95,18 @@ Infinispan Server setup
 <cache-container name="clustered" default-cache="default" statistics="true">
         ...
         <replicated-cache-configuration name="sessions-cfg" mode="SYNC" start="EAGER" batching="false">
-            <transaction mode="NON_XA" locking="PESSIMISTIC"/>
+            <transaction mode="NON_DURABLE_XA" locking="PESSIMISTIC"/>
+            <locking acquire-timeout="0" />
             <backups>
                 <backup site="site2" failure-policy="FAIL" strategy="SYNC" enabled="true"/>
             </backups>
         </replicated-cache-configuration>
 
         <replicated-cache name="work" configuration="sessions-cfg"/>
         <replicated-cache name="sessions" configuration="sessions-cfg"/>
+        <replicated-cache name="clientSessions" configuration="sessions-cfg"/>
         <replicated-cache name="offlineSessions" configuration="sessions-cfg"/>
+        <replicated-cache name="offlineClientSessions" configuration="sessions-cfg"/>
         <replicated-cache name="actionTokens" configuration="sessions-cfg"/>
         <replicated-cache name="loginFailures" configuration="sessions-cfg"/>
 

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

@@ -22,6 +22,8 @@
 import java.util.function.Supplier;
 
 import org.infinispan.commons.api.BasicCache;
+import org.jboss.logging.Logger;
+import org.keycloak.common.util.Retry;
 import org.keycloak.models.CodeToTokenStoreProvider;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.sessions.infinispan.entities.ActionTokenValueEntity;
@@ -31,6 +33,8 @@
  */
 public class InfinispanCodeToTokenStoreProvider implements CodeToTokenStoreProvider {
 
+    public static final Logger logger = Logger.getLogger(InfinispanCodeToTokenStoreProvider.class);
+
     private final Supplier<BasicCache<UUID, ActionTokenValueEntity>> codeCache;
     private final KeycloakSession session;
 
@@ -45,9 +49,24 @@ public boolean putIfAbsent(UUID codeId) {
 
         int lifespanInSeconds = session.getContext().getRealm().getAccessCodeLifespan();
 
-        BasicCache<UUID, ActionTokenValueEntity> cache = codeCache.get();
-        ActionTokenValueEntity existing = cache.putIfAbsent(codeId, tokenValue, lifespanInSeconds, TimeUnit.SECONDS);
-        return existing == null;
+        boolean codeAlreadyExists = Retry.call(() -> {
+
+            try {
+                BasicCache<UUID, ActionTokenValueEntity> cache = codeCache.get();
+                ActionTokenValueEntity existing = cache.putIfAbsent(codeId, tokenValue, lifespanInSeconds, TimeUnit.SECONDS);
+                return existing == null;
+            } catch (RuntimeException re) {
+                if (logger.isDebugEnabled()) {
+                    logger.debugf(re, "Failed when adding code %s", codeId);
+                }
+
+                // Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
+                throw re;
+            }
+
+        }, 3, 0);
+
+        return codeAlreadyExists;
     }
 
     @Override

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

@@ -255,6 +255,10 @@ private <K, V extends SessionEntity> boolean checkRemoteCache(KeycloakSession se
 
             RemoteCache<K, SessionEntityWrapper<V>> remoteCache = (RemoteCache) remoteStores.iterator().next().getRemoteCache();
 
+            if (remoteCache == null) {
+                throw new IllegalStateException("No remote cache available for the infinispan cache: " + ispnCache.getName());
+            }
+
             remoteCacheInvoker.addRemoteCache(ispnCache.getName(), remoteCache, maxIdleLoader);
 
             RemoteCacheSessionListener hotrodListener = RemoteCacheSessionListener.createListener(session, ispnCache, remoteCache);

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

@@ -17,6 +17,7 @@
 
 package org.keycloak.models.sessions.infinispan.remotestore;
 
+import org.keycloak.common.util.Retry;
 import org.keycloak.common.util.Time;
 import java.util.Collections;
 import java.util.HashMap;
@@ -32,6 +33,7 @@
 import org.keycloak.models.RealmModel;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
 import org.keycloak.models.sessions.infinispan.changes.SessionUpdateTask;
+import org.keycloak.models.sessions.infinispan.entities.LoginFailureEntity;
 import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
 import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
 
@@ -71,14 +73,28 @@ public <K, V extends SessionEntity> void runTask(KeycloakSession kcSession, Real
             return;
         }
 
-        long maxIdleTimeMs = context.maxIdleTimeLoader.getMaxIdleTimeMs(realm);
+        long loadedMaxIdleTimeMs = context.maxIdleTimeLoader.getMaxIdleTimeMs(realm);
 
         // Double the timeout to ensure that entry won't expire on remoteCache in case that write of some entities to remoteCache is postponed (eg. userSession.lastSessionRefresh)
-        maxIdleTimeMs = maxIdleTimeMs * 2;
+        final long maxIdleTimeMs = loadedMaxIdleTimeMs * 2;
 
         logger.debugf("Running task '%s' on remote cache '%s' . Key is '%s'", operation, cacheName, key);
 
-        runOnRemoteCache(context.remoteCache, maxIdleTimeMs, key, task, sessionWrapper);
+        Retry.execute(() -> {
+
+            try {
+                runOnRemoteCache(context.remoteCache, maxIdleTimeMs, key, task, sessionWrapper);
+            } catch (RuntimeException re) {
+                if (logger.isDebugEnabled()) {
+                    logger.debugf(re, "Failed running task '%s' on remote cache '%s' . Key: '%s' . Will try to retry the task",
+                            operation, cacheName, key);
+                }
+
+                // Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
+                throw re;
+            }
+
+        }, 10, 0);
     }
 
 

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

@@ -30,6 +30,7 @@
 import org.infinispan.client.hotrod.annotation.ClientListener;
 import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
 import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
+import org.infinispan.client.hotrod.exceptions.HotRodClientException;
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.infinispan.persistence.manager.PersistenceManager;
 import org.infinispan.persistence.remote.RemoteStore;
@@ -47,9 +48,13 @@ public class ConcurrencyJDGRemoteCacheTest {
 
     private static Map<String, EntryInfo> state = new HashMap<>();
 
+    private RemoteCache remoteCache1;
+    private RemoteCache remoteCache2;
+
+
     public static void main(String[] args) throws Exception {
         // Init map somehow
-        for (int i=0 ; i<30 ; i++) {
+        for (int i=0 ; i<3000 ; i++) {
             String key = "key-" + i;
             state.put(key, new EntryInfo());
         }
@@ -58,19 +63,25 @@ public static void main(String[] args) throws Exception {
         Worker worker1 = createWorker(1);
         Worker worker2 = createWorker(2);
 
+        long start = System.currentTimeMillis();
+
         // Start and join workers
         worker1.start();
         worker2.start();
 
         worker1.join();
         worker2.join();
 
+        long took = System.currentTimeMillis() - start;
+
         // Output
         for (Map.Entry<String, EntryInfo> entry : state.entrySet()) {
             System.out.println(entry.getKey() + ":::" + entry.getValue());
             worker1.cache.remove(entry.getKey());
         }
 
+        System.out.println("Took: " + took + " ms");
+
         // Finish JVM
         worker1.cache.getCacheManager().stop();
         worker2.cache.getCacheManager().stop();
@@ -127,7 +138,7 @@ public void run() {
                 String cacheKey = entry.getKey();
                 EntryInfo wrapper = state.get(cacheKey);
 
-                int val = getClusterStartupTime(this.cache, cacheKey, wrapper);
+                int val = getClusterStartupTime(this.cache, cacheKey, wrapper, myThreadId);
                 if (myThreadId == 1) {
                     wrapper.th1.set(val);
                 } else {
@@ -141,8 +152,8 @@ public void run() {
 
     }
 
-    public static int getClusterStartupTime(Cache<String, Integer> cache, String cacheKey, EntryInfo wrapper) {
-        Integer startupTime = new Random().nextInt(1024);
+    public static int getClusterStartupTime(Cache<String, Integer> cache, String cacheKey, EntryInfo wrapper, int myThreadId) {
+        Integer startupTime = myThreadId==1 ? Integer.parseInt(cacheKey.substring(4)) : Integer.parseInt(cacheKey.substring(4)) * 2;
 
         // Concurrency doesn't work correctly with this
         //Integer existingClusterStartTime = (Integer) cache.putIfAbsent(cacheKey, startupTime);
@@ -154,21 +165,25 @@ public static int getClusterStartupTime(Cache<String, Integer> cache, String cac
         for (int i=0 ; i<10 ; i++) {
             try {
                 existingClusterStartTime = (Integer) remoteCache.withFlags(Flag.FORCE_RETURN_VALUE).putIfAbsent(cacheKey, startupTime);
-            } catch (Exception ce) {
+                break;
+            } catch (HotRodClientException ce) {
                 if (i == 9) {
                     throw ce;
                     //break;
                 } else {
-                    System.err.println("EXception: i=" + i);
+                    wrapper.exceptions.incrementAndGet();
+                    System.err.println("Exception: i=" + i + " for key: " + cacheKey + " and myThreadId: " + myThreadId);
                 }
             }
         }
 
-        if (existingClusterStartTime == null || startupTime.equals(remoteCache.get(cacheKey))) {
+        if (existingClusterStartTime == null
+//                || startupTime.equals(remoteCache.get(cacheKey))
+                ) {
             wrapper.successfulInitializations.incrementAndGet();
             return startupTime;
         } else {
-            System.err.println("Not equal!!! startupTime=" + startupTime + ", existingClusterStartTime=" + existingClusterStartTime );
+            wrapper.failedInitializations.incrementAndGet();
             return existingClusterStartTime;
         }
     }
@@ -178,10 +193,13 @@ public static class EntryInfo {
         AtomicInteger successfulListenerWrites = new AtomicInteger(0);
         AtomicInteger th1 = new AtomicInteger();
         AtomicInteger th2 = new AtomicInteger();
+        AtomicInteger failedInitializations = new AtomicInteger();
+        AtomicInteger exceptions = new AtomicInteger();
 
         @Override
         public String toString() {
-            return String.format("Inits: %d, listeners: %d, th1: %d, th2: %d", successfulInitializations.get(), successfulListenerWrites.get(), th1.get(), th2.get());
+            return String.format("Inits: %d, listeners: %d, failedInits: %d, exceptions: %s, th1: %d, th2: %d", successfulInitializations.get(), successfulListenerWrites.get(),
+            failedInitializations.get(), exceptions.get(), th1.get(), th2.get());
         }
     }
 

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

@@ -121,6 +121,8 @@ public static void main(String[] args) throws Exception {
 
             logger.info("SESSIONS NOT AVAILABLE ON DC2");
 
+            long took = System.currentTimeMillis() - start;
+            logger.infof("took %d ms", took);
 
             //        // Start and join workers
 //        worker1.start();
@@ -137,8 +139,6 @@ public static void main(String[] args) throws Exception {
             cache2.getCacheManager().stop();
         }
 
-        long took = System.currentTimeMillis() - start;
-
 //        // Output
 //        for (Map.Entry<String, EntryInfo> entry : state.entrySet()) {
 //            System.out.println(entry.getKey() + ":::" + entry.getValue());

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

@@ -187,6 +187,10 @@ public static void main(String[] args) throws Exception {
                 ", successfulListenerWrites: " + successfulListenerWrites.get() + ", successfulListenerWrites2: " + successfulListenerWrites2.get() +
                 ", failedReplaceCounter: " + failedReplaceCounter.get() + ", failedReplaceCounter2: " + failedReplaceCounter2.get());
 
+
+        System.out.println("remoteCache1.notes: " + ((UserSessionEntity) remoteCache1.get("123")).getNotes().size() );
+        System.out.println("remoteCache2.notes: " + ((UserSessionEntity) remoteCache2.get("123")).getNotes().size() );
+
         System.out.println("Histogram: ");
         //histogram.dumpStats();
 
@@ -314,14 +318,26 @@ public void run() {
                 // In case it's hardcoded (eg. all the replaces are doing same change, so session is defacto not changed), then histogram may contain bigger value than 1 on some places.
                 //String noteKey = "some";
 
-                boolean replaced = false;
-                while (!replaced) {
+                ReplaceStatus replaced = ReplaceStatus.NOT_REPLACED;
+                while (replaced != ReplaceStatus.REPLACED) {
                     VersionedValue<UserSessionEntity> versioned = remoteCache.getVersioned("123");
                     UserSessionEntity oldSession = versioned.getValue();
                     //UserSessionEntity clone = DistributedCacheConcurrentWritesTest.cloneSession(oldSession);
                     UserSessionEntity clone = oldSession;
 
-                    clone.getNotes().put(noteKey, "someVal");
+                    // In case that exception was thrown (ReplaceStatus.ERROR), the remoteCache may have the note. Seems that transactions are not fully rolled-back on the JDG side
+                    // in case that backup fails
+                    if (replaced == ReplaceStatus.NOT_REPLACED) {
+                        clone.getNotes().put(noteKey, "someVal");
+                    } else if (replaced == ReplaceStatus.ERROR) {
+                        if (clone.getNotes().containsKey(noteKey)) {
+                            System.err.println("I HAVE THE KEY: " + noteKey);
+                        } else {
+                            System.err.println("I DON'T HAVE THE KEY: " + noteKey);
+                            clone.getNotes().put(noteKey, "someVal");
+                        }
+                    }
+
                     //cache.replace("123", clone);
                     replaced = cacheReplace(versioned, clone);
                 }
@@ -336,7 +352,7 @@ public void run() {
 
         }
 
-        private boolean cacheReplace(VersionedValue<UserSessionEntity> oldSession, UserSessionEntity newSession) {
+        private ReplaceStatus cacheReplace(VersionedValue<UserSessionEntity> oldSession, UserSessionEntity newSession) {
             try {
                 boolean replaced = remoteCache.replaceWithVersion("123", newSession, oldSession.getVersion());
                 //boolean replaced = true;
@@ -348,15 +364,19 @@ private boolean cacheReplace(VersionedValue<UserSessionEntity> oldSession, UserS
                 } else {
                     histogram.increaseSuccessOpsCount(oldSession.getVersion());
                 }
-                return replaced;
+                return replaced ? ReplaceStatus.REPLACED : ReplaceStatus.NOT_REPLACED;
             } catch (Exception re) {
                 failedReplaceCounter2.incrementAndGet();
-                return false;
+                return ReplaceStatus.ERROR;
             }
             //return replaced;
         }
 
     }
+
+    private enum ReplaceStatus {
+        REPLACED, NOT_REPLACED, ERROR
+    }
 /*
     // Worker, which operates on "classic" cache and rely on operations delegated to the second cache
     private static class CacheWorker extends Thread {

testsuite/integration-arquillian/servers/cache-server/jboss/common/add-keycloak-caches.xsl

@@ -36,7 +36,8 @@
             <xsl:apply-templates select="@* | node()" />
 
             <replicated-cache-configuration name="sessions-cfg" mode="SYNC" start="EAGER" batching="false">
-                <transaction mode="NON_XA" locking="PESSIMISTIC"/>
+                <transaction mode="NON_DURABLE_XA" locking="PESSIMISTIC"/>
+                <locking acquire-timeout="0" />
                 <backups>
                     <backup site="{$remote.site}" failure-policy="FAIL" strategy="SYNC" enabled="true"/>
                 </backups>