libre.sh/keycloak-scim
2
0
Fork 0
Code Issues 14 Pull requests Releases Packages Activity Actions 6

Execute persistent sessions tests in CI and fix deadlock (#29236)

Browse source 
* Execute persistent sessions tests in CI and fix deadlock in UserSessionConcurrencyTest 
The deadlock was caused by a switch to reactive handling of embedded Infinispan updates introduced here #28862
Closes #29235

Signed-off-by: Alexander Schwartz <aschwart@redhat.com>
This commit is contained in:
Alexander Schwartz 2024-05-03 10:42:34 +02:00 committed by GitHub
parent 8c3f7cc6e9
commit 05b6f897ce
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 49 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.github/workflows/ci.yml vendored
View file

@ -324,7 +324,7 @@ jobs:
run: |
TESTS=`testsuite/integration-arquillian/tests/base/testsuites/suite.sh persistent-sessions`
echo "Tests: $TESTS"
./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Dauth.server.features=${{ matrix.variant }} -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
./mvnw test ${{ env.SUREFIRE_RETRY }} -Pauth-server-quarkus -Dauth.server.feature=${{ matrix.variant }} -Dtest=$TESTS -pl testsuite/integration-arquillian/tests/base 2>&1 | misc/log/trimmer.sh
- name: Upload JVM Heapdumps
if: always()

121
model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/EmbeddedCachesChangesPerformer.java
View file

@ -17,7 +17,6 @@
package org.keycloak.models.sessions.infinispan.changes;
import org.infinispan.AdvancedCache;
import org.infinispan.Cache;
import org.infinispan.context.Flag;
import org.jboss.logging.Logger;
@ -25,25 +24,22 @@ import org.keycloak.connections.infinispan.InfinispanUtil;
import org.keycloak.models.sessions.infinispan.CacheDecorators;
import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
public class EmbeddedCachesChangesPerformer<K, V extends SessionEntity> implements SessionChangesPerformer<K, V> {
private static final Logger LOG = Logger.getLogger(EmbeddedCachesChangesPerformer.class);
private final Cache<K, SessionEntityWrapper<V>> cache;
private final List<Supplier<CompletableFuture<?>>> changes = new LinkedList<>();
private final List<Runnable> changes = new LinkedList<>();
public EmbeddedCachesChangesPerformer(Cache<K, SessionEntityWrapper<V>> cache) {
this.cache = cache;
}
private CompletableFuture<?> runOperationInCluster(K key, MergedUpdate<V> task, SessionEntityWrapper<V> sessionWrapper) {
private void runOperationInCluster(K key, MergedUpdate<V> task, SessionEntityWrapper<V> sessionWrapper) {
V session = sessionWrapper.getEntity();
SessionUpdateTask.CacheOperation operation = task.getOperation(session);
@ -53,76 +49,67 @@ public class EmbeddedCachesChangesPerformer<K, V extends SessionEntity> implemen
switch (operation) {
case REMOVE:
// Just remove it
return CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache)
CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache)
.withFlags(Flag.IGNORE_RETURN_VALUES)
.removeAsyncEntry(key);
.remove(key);
break;
case ADD:
return CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache)
CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache)
.withFlags(Flag.IGNORE_RETURN_VALUES)
.putAsync(key, sessionWrapper, task.getLifespanMs(), TimeUnit.MILLISECONDS, task.getMaxIdleTimeMs(), TimeUnit.MILLISECONDS)
.thenAcceptAsync(v -> LOG.tracef("Added entity '%s' to the cache '%s' . Lifespan: %d ms, MaxIdle: %d ms", key, cache.getName(), task.getLifespanMs(), task.getMaxIdleTimeMs()));
.put(key, sessionWrapper, task.getLifespanMs(), TimeUnit.MILLISECONDS, task.getMaxIdleTimeMs(), TimeUnit.MILLISECONDS);
LOG.tracef("Added entity '%s' to the cache '%s' . Lifespan: %d ms, MaxIdle: %d ms", key, cache.getName(), task.getLifespanMs(), task.getMaxIdleTimeMs());
break;
case ADD_IF_ABSENT:
return CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache).putIfAbsentAsync(key, sessionWrapper, task.getLifespanMs(), TimeUnit.MILLISECONDS, task.getMaxIdleTimeMs(), TimeUnit.MILLISECONDS)
.thenAccept(existing -> {
if (existing != null) {
LOG.debugf("Existing entity in cache for key: %s . Will update it", key);
SessionEntityWrapper<V> existing = CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache).putIfAbsent(key, sessionWrapper, task.getLifespanMs(), TimeUnit.MILLISECONDS, task.getMaxIdleTimeMs(), TimeUnit.MILLISECONDS);
if (existing != null) {
LOG.debugf("Existing entity in cache for key: %s . Will update it", key);
// Apply updates on the existing entity and replace it
task.runUpdate(existing.getEntity());
// Apply updates on the existing entity and replace it
task.runUpdate(existing.getEntity());
replace(key, task, existing, task.getLifespanMs(), task.getMaxIdleTimeMs()).join();
} else {
LOG.tracef("Add_if_absent successfully called for entity '%s' to the cache '%s' . Lifespan: %d ms, MaxIdle: %d ms", key, cache.getName(), task.getLifespanMs(), task.getMaxIdleTimeMs());
}
});
replace(key, task, existing, task.getLifespanMs(), task.getMaxIdleTimeMs());
} else {
LOG.tracef("Add_if_absent successfully called for entity '%s' to the cache '%s' . Lifespan: %d ms, MaxIdle: %d ms", key, cache.getName(), task.getLifespanMs(), task.getMaxIdleTimeMs());
}
break;
case REPLACE:
return replace(key, task, sessionWrapper, task.getLifespanMs(), task.getMaxIdleTimeMs());
replace(key, task, sessionWrapper, task.getLifespanMs(), task.getMaxIdleTimeMs());
break;
default:
throw new IllegalStateException("Unsupported state " + operation);
}
}
private CompletableFuture<?> replace(K key, MergedUpdate<V> task, SessionEntityWrapper<V> oldVersionEntityFirst, long lifespanMs, long maxIdleTimeMs) {
AdvancedCache<K, SessionEntityWrapper<V>> writeCache = CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache);
// make one async attempt
SessionEntityWrapper<V> newVersionEntityFirst = generateNewVersionAndWrapEntity(oldVersionEntityFirst.getEntity(), oldVersionEntityFirst.getLocalMetadata());
return writeCache.computeIfPresentAsync(key, new ReplaceFunction<>(oldVersionEntityFirst.getVersion(), newVersionEntityFirst), lifespanMs, TimeUnit.MILLISECONDS, maxIdleTimeMs, TimeUnit.MILLISECONDS)
.thenAccept(returnValue -> {
int iteration = 0;
SessionEntityWrapper<V> newVersionEntity = newVersionEntityFirst;
SessionEntityWrapper<V> oldVersion = oldVersionEntityFirst;
private void replace(K key, MergedUpdate<V> task, SessionEntityWrapper<V> oldVersionEntity, long lifespanMs, long maxIdleTimeMs) {
SessionEntityWrapper<V> oldVersion = oldVersionEntity;
SessionEntityWrapper<V> returnValue = null;
int iteration = 0;
V session = oldVersion.getEntity();
var writeCache = CacheDecorators.skipCacheStoreIfRemoteCacheIsEnabled(cache);
while (iteration++ < InfinispanUtil.MAXIMUM_REPLACE_RETRIES) {
SessionEntityWrapper<V> newVersionEntity = generateNewVersionAndWrapEntity(session, oldVersion.getLocalMetadata());
returnValue = writeCache.computeIfPresent(key, new ReplaceFunction<>(oldVersion.getVersion(), newVersionEntity), lifespanMs, TimeUnit.MILLISECONDS, maxIdleTimeMs, TimeUnit.MILLISECONDS);
while (true) {
if (returnValue == null) {
LOG.debugf("Entity %s not found. Maybe removed in the meantime. Replace task will be ignored", key);
return;
}
if (returnValue == null) {
LOG.debugf("Entity %s not found. Maybe removed in the meantime. Replace task will be ignored", key);
return;
}
if (returnValue.getVersion().equals(newVersionEntity.getVersion())){
if (LOG.isTraceEnabled()) {
LOG.tracef("Replace SUCCESS for entity: %s . old version: %s, new version: %s, Lifespan: %d ms, MaxIdle: %d ms", key, oldVersion.getVersion(), newVersionEntity.getVersion(), task.getLifespanMs(), task.getMaxIdleTimeMs());
}
if (returnValue.getVersion().equals(newVersionEntity.getVersion())){
if (LOG.isTraceEnabled()) {
LOG.tracef("Replace SUCCESS for entity: %s . old version: %s, new version: %s, Lifespan: %d ms, MaxIdle: %d ms", key, oldVersion.getVersion(), newVersionEntity.getVersion(), task.getLifespanMs(), task.getMaxIdleTimeMs());
}
return;
}
return;
} else {
if (LOG.isTraceEnabled()) {
LOG.tracef("Replace failed for entity: %s, old version %s, new version %s. Will try again", key, oldVersion.getVersion(), newVersionEntity.getVersion());
}
}
oldVersion = returnValue;
session = oldVersion.getEntity();
task.runUpdate(session);
}
if (++iteration >= InfinispanUtil.MAXIMUM_REPLACE_RETRIES) {
LOG.warnf("Failed to replace entity '%s' in cache '%s'. Expected: %s, Current: %s", key, cache.getName(), oldVersion, returnValue);
return;
}
oldVersion = returnValue;
V session = oldVersion.getEntity();
task.runUpdate(session);
newVersionEntity = generateNewVersionAndWrapEntity(session, oldVersion.getLocalMetadata());
returnValue = writeCache.computeIfPresent(key, new ReplaceFunction<>(oldVersion.getVersion(), newVersionEntity), lifespanMs, TimeUnit.MILLISECONDS, maxIdleTimeMs, TimeUnit.MILLISECONDS);
}
});
LOG.warnf("Failed to replace entity '%s' in cache '%s'. Expected: %s, Current: %s", key, cache.getName(), oldVersion, returnValue);
}
private SessionEntityWrapper<V> generateNewVersionAndWrapEntity(V entity, Map<String, String> localMetadata) {
@ -136,18 +123,6 @@ public class EmbeddedCachesChangesPerformer<K, V extends SessionEntity> implemen
@Override
public void applyChanges() {
if (!changes.isEmpty()) {
List<Throwable> exceptions = new ArrayList<>();
CompletableFuture.allOf(changes.stream().map(s -> s.get().exceptionally(throwable -> {
exceptions.add(throwable);
return null;
})).toArray(CompletableFuture[]::new)).join();
// If any of those futures has failed, add the exceptions as suppressed exceptions to our runtime exception
if (!exceptions.isEmpty()) {
RuntimeException ex = new RuntimeException("unable to complete the session updates");
exceptions.forEach(ex::addSuppressed);
throw ex;
}
}
changes.forEach(Runnable::run);
}
}