mariusz.pacek/keycloak
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

KEYCLOAK-5371 More stable cross-dc tests

Browse Source
This commit is contained in:
mposolda committed 2017-11-08 10:03:04 +01:00
1 parent 7cf52044b2
commit 701b7acd80
6 files changed
+125 -42
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

model/infinispan/src/main/java/org/keycloak/cluster/infinispan/InfinispanNotificationsManager.java
+2 -5
View File
@@ -222,11 +222,8 @@ public class InfinispanNotificationsManager {
});
} catch (RejectedExecutionException ree) {
logger.warnf("Rejected submitting of the event for key: %s. Probably server going to shutdown", key);
if (logger.isDebugEnabled()) {
logger.debug(ree.getMessage(), ree);
}
logger.errorf("Rejected submitting of the event for key: %s. Value: %s, Server going to shutdown or pool exhausted. Pool: %s", key, workCache.get(key), listenersExecutor.toString());
throw ree;
}
}
model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/InfinispanCodeToTokenStoreProvider.java
+12 -15
View File
@@ -21,6 +21,7 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import org.infinispan.client.hotrod.exceptions.HotRodClientException;
import org.infinispan.commons.api.BasicCache;
import org.jboss.logging.Logger;
import org.keycloak.common.util.Retry;
@@ -49,24 +50,20 @@ public class InfinispanCodeToTokenStoreProvider implements CodeToTokenStoreProvi
int lifespanInSeconds = session.getContext().getRealm().getAccessCodeLifespan();
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;
try {
BasicCache<UUID, ActionTokenValueEntity> cache = codeCache.get();
ActionTokenValueEntity existing = cache.putIfAbsent(codeId, tokenValue, lifespanInSeconds, TimeUnit.SECONDS);
return existing == null;
} catch (HotRodClientException re) {
// No need to retry. The hotrod (remoteCache) has some retries in itself in case of some random network error happened.
// In case of lock conflict, we don't want to retry anyway as there was likely an attempt to use the code from different place.
if (logger.isDebugEnabled()) {
logger.debugf(re, "Failed when adding code %s", codeId);
}
}, 3, 0);
return false;
}
return codeAlreadyExists;
}
@Override
model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/ClientListenerExecutorDecorator.java
+21 -3
View File
@@ -21,6 +21,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import org.infinispan.client.hotrod.event.ClientCacheEntryCreatedEvent;
import org.infinispan.client.hotrod.event.ClientCacheEntryModifiedEvent;
@@ -28,6 +29,7 @@ import org.infinispan.client.hotrod.event.ClientCacheEntryRemovedEvent;
import org.infinispan.client.hotrod.event.ClientEvent;
import org.jboss.logging.Logger;
import org.keycloak.common.util.MultivaluedHashMap;
import org.keycloak.common.util.Time;
import static org.infinispan.client.hotrod.event.ClientEvent.Type.CLIENT_CACHE_ENTRY_CREATED;
import static org.infinispan.client.hotrod.event.ClientEvent.Type.CLIENT_CACHE_ENTRY_REMOVED;
@@ -94,24 +96,40 @@ public class ClientListenerExecutorDecorator<K> {
// Assume it's called from the synchronized block
private void submitImpl(K key, MyClientEvent event, Runnable r) {
logger.debugf("Submitting event to the executor: %s", event.toString());
logger.debugf("Submitting event to the executor: %s . eventsInProgress size: %d, eventsQueue size: %d", event.toString(), eventsInProgress.size(), eventsQueue.size());
eventsInProgress.put(key, event);
Runnable decoratedRunnable = () -> {
Long start = null;
try {
if (logger.isDebugEnabled()) {
start = Time.currentTimeMillis();
}
r.run();
} finally {
synchronized (lock) {
logger.debugf("Finished processing event by the executor: %s", event.toString());
eventsInProgress.remove(key);
if (logger.isDebugEnabled()) {
long took = Time.currentTimeMillis() - start;
logger.debugf("Finished processing event by the executor: %s, took: %d ms. EventsInProgress size: %d", event.toString(), took, eventsInProgress.size());
}
pollQueue(key);
}
}
};
decorated.submit(decoratedRunnable);
try {
decorated.submit(decoratedRunnable);
} catch (RejectedExecutionException ree) {
eventsInProgress.remove(key);
logger.errorf("Rejected execution of task for the event '%s' . Try to increase the pool size. Pool is '%s'", event.toString(), decorated.toString());
throw ree;
}
}
model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remotestore/RemoteCacheInvoker.java
+17 -10
View File
@@ -17,6 +17,7 @@
package org.keycloak.models.sessions.infinispan.remotestore;
import org.infinispan.client.hotrod.exceptions.HotRodClientException;
import org.keycloak.common.util.Retry;
import org.keycloak.common.util.Time;
import java.util.Collections;
@@ -69,7 +70,9 @@ public class RemoteCacheInvoker {
SessionUpdateTask.CrossDCMessageStatus status = task.getCrossDCMessageStatus(sessionWrapper);
if (status == SessionUpdateTask.CrossDCMessageStatus.NOT_NEEDED) {
logger.debugf("Skip writing to remoteCache for entity '%s' of cache '%s' and operation '%s'", key, cacheName, operation);
if (logger.isTraceEnabled()) {
logger.tracef("Skip writing to remoteCache for entity '%s' of cache '%s' and operation '%s'", key, cacheName, operation);
}
return;
}
@@ -78,23 +81,25 @@ public class RemoteCacheInvoker {
// 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)
final long maxIdleTimeMs = loadedMaxIdleTimeMs * 2;
logger.debugf("Running task '%s' on remote cache '%s' . Key is '%s'", operation, cacheName, key);
if (logger.isTraceEnabled()) {
logger.tracef("Running task '%s' on remote cache '%s' . Key is '%s'", operation, cacheName, key);
}
Retry.execute(() -> {
Retry.executeWithBackoff((int iteration) -> {
try {
runOnRemoteCache(context.remoteCache, maxIdleTimeMs, key, task, sessionWrapper);
} catch (RuntimeException re) {
} catch (HotRodClientException 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);
logger.debugf(re, "Failed running task '%s' on remote cache '%s' . Key: '%s', iteration '%s'. Will try to retry the task",
operation, cacheName, key, iteration);
}
// Rethrow the exception. Retry will take care of handle the exception and eventually retry the operation.
throw re;
}
}, 10, 0);
}, 10, 10);
}
@@ -146,15 +151,17 @@ public class RemoteCacheInvoker {
// Run task on the remote session
task.runUpdate(session);
logger.debugf("Before replaceWithVersion. Entity to write version %d: %s", versioned.getVersion(), session);
if (logger.isTraceEnabled()) {
logger.tracef("Before replaceWithVersion. Entity to write version %d: %s", versioned.getVersion(), session);
}
replaced = remoteCache.replaceWithVersion(key, SessionEntityWrapper.forTransport(session), versioned.getVersion(), lifespanMs, TimeUnit.MILLISECONDS, maxIdleMs, TimeUnit.MILLISECONDS);
if (!replaced) {
logger.debugf("Failed to replace entity '%s' version %d. Will retry again", key, versioned.getVersion());
} else {
if (logger.isDebugEnabled()) {
logger.debugf("Replaced entity version %d in remote cache: %s", versioned.getVersion(), session);
if (logger.isTraceEnabled()) {
logger.tracef("Replaced entity version %d in remote cache: %s", versioned.getVersion(), session);
}
}
}