Enable ProtoStream encoding for External Infinispan feature

The ProtoStream schema is automatically uploaded to the Infinispan
server during startup.
When the schema is updated, the indexes are updated and re-created.
Use the delete statement to delete entities when a realm is removed.

Fixes #30931

Signed-off-by: Pedro Ruivo <pruivo@redhat.com>

model/infinispan/pom.xml

@@ -52,6 +52,10 @@
             <artifactId>keycloak-server-spi-private</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-api</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-core</artifactId>
@@ -60,6 +64,16 @@
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-cachestore-remote</artifactId>
         </dependency>
+        <!-- required for query/search in the external Infinispan server -->
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-remote-query-client</artifactId>
+        </dependency>
+        <!-- to be removed after https://issues.redhat.com/browse/ISPN-16220 -->
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-query-dsl</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-component-annotations</artifactId>

model/infinispan/src/main/java/org/keycloak/marshalling/KeycloakIndexSchemaUtil.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2024 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.marshalling;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.infinispan.api.annotations.indexing.model.Values;
+import org.infinispan.protostream.config.Configuration;
+import org.infinispan.protostream.descriptors.AnnotationElement;
+import org.infinispan.protostream.descriptors.Descriptor;
+import org.infinispan.protostream.descriptors.FieldDescriptor;
+import org.infinispan.protostream.impl.AnnotatedDescriptorImpl;
+
+public class KeycloakIndexSchemaUtil {
+
+    // Basic annotation data
+    private static final String BASIC_ANNOTATION = "Basic";
+    private static final String NAME_ATTRIBUTE = "name";
+    private static final String SEARCHABLE_ATTRIBUTE = "searchable";
+    private static final String PROJECTABLE_ATTRIBUTE = "projectable";
+    private static final String AGGREGABLE_ATTRIBUTE = "aggregable";
+    private static final String SORTABLE_ATTRIBUTE = "sortable";
+    private static final String INDEX_NULL_AS_ATTRIBUTE = "indexNullAs";
+
+    // we only use Basic annotation, we may need to add others in the future.
+    private static final List<String> INDEX_ANNOTATION = List.of(BASIC_ANNOTATION);
+
+    /**
+     * Adds the annotations to the ProtoStream parser.
+     */
+    public static void configureAnnotationProcessor(Configuration.Builder builder) {
+        //TODO remove in the future?
+        builder.annotationsConfig()
+                .annotation(BASIC_ANNOTATION, AnnotationElement.AnnotationTarget.FIELD)
+                .attribute(NAME_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.STRING)
+                .defaultValue("")
+                .attribute(SEARCHABLE_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.BOOLEAN)
+                .defaultValue(true)
+                .attribute(PROJECTABLE_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.BOOLEAN)
+                .defaultValue(false)
+                .attribute(AGGREGABLE_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.BOOLEAN)
+                .defaultValue(false)
+                .attribute(SORTABLE_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.BOOLEAN)
+                .defaultValue(false)
+                .attribute(INDEX_NULL_AS_ATTRIBUTE)
+                .type(AnnotationElement.AttributeType.STRING)
+                .defaultValue(Values.DO_NOT_INDEX_NULL);
+    }
+
+    /**
+     * Compares two entities and returns {@code true} if any indexing related annotation were changed, added or removed.
+     */
+    public static boolean isIndexSchemaChanged(Descriptor oldDescriptor, Descriptor newDescriptor) {
+        var allFields = Stream.concat(
+                oldDescriptor.getFields().stream().map(AnnotatedDescriptorImpl::getName),
+                newDescriptor.getFields().stream().map(AnnotatedDescriptorImpl::getName)
+        ).collect(Collectors.toSet());
+        for (var fieldName : allFields) {
+            var oldField = oldDescriptor.findFieldByName(fieldName);
+            var newField = newDescriptor.findFieldByName(fieldName);
+            if (isNewFieldAdded(oldField, newField)) {
+                if (isFieldIndexed(newField)) {
+                    // a new field is added and is indexed
+                    return true;
+                }
+                continue;
+            }
+            if (isNewFieldRemoved(oldField, newField)) {
+                if (isFieldIndexed(oldField)) {
+                    // an old field is indexed and has been removed
+                    return true;
+                }
+                continue;
+            }
+            if (isFieldIndexed(oldField) != isFieldIndexed(newField)) {
+                // some annotation added or removed
+                return true;
+            }
+            if (!isFieldIndexed(oldField) && !isFieldIndexed(newField)) {
+                // nothing changes
+                continue;
+            }
+            if (isAnnotationChanged(oldField, newField)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean isNewFieldAdded(FieldDescriptor oldField, FieldDescriptor newField) {
+        return oldField == null && newField != null;
+    }
+
+    private static boolean isNewFieldRemoved(FieldDescriptor oldField, FieldDescriptor newField) {
+        return oldField != null && newField == null;
+    }
+
+    private static boolean isFieldIndexed(FieldDescriptor descriptor) {
+        var annotations = descriptor.getAnnotations();
+        return INDEX_ANNOTATION.stream().anyMatch(annotations::containsKey);
+    }
+
+    private static boolean isAnnotationChanged(FieldDescriptor oldField, FieldDescriptor newField) {
+        return INDEX_ANNOTATION.stream().anyMatch(s -> {
+            var oldAnnot = oldField.getAnnotations().get(s);
+            var newAnnot = newField.getAnnotations().get(s);
+            return isAnnotatedDifferent(oldAnnot, newAnnot);
+        });
+    }
+
+    private static boolean isAnnotatedDifferent(AnnotationElement.Annotation oldAnnot, AnnotationElement.Annotation newAnnot) {
+        if (oldAnnot == null && newAnnot == null) {
+            // annotation not present in both field
+            return false;
+        }
+        if (oldAnnot != null && newAnnot == null) {
+            // annotation present *only* in old field
+            return true;
+        }
+        if (oldAnnot == null) {
+            // annotation present *only* in new field
+            return true;
+        }
+        // check if the attributes didn't change
+        return !Objects.equals(getAnnotationValues(oldAnnot), getAnnotationValues(newAnnot));
+
+    }
+
+    private static Map<String, Object> getAnnotationValues(AnnotationElement.Annotation annotation) {
+        return annotation.getAttributes()
+                .entrySet()
+                .stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue().getValue()));
+    }
+
+}

model/infinispan/src/main/java/org/keycloak/marshalling/KeycloakModelSchema.java

@@ -17,9 +17,17 @@
 
 package org.keycloak.marshalling;
 
+import java.util.Objects;
+import java.util.Optional;
+
+import org.infinispan.protostream.FileDescriptorSource;
 import org.infinispan.protostream.GeneratedSchema;
 import org.infinispan.protostream.annotations.ProtoSchema;
 import org.infinispan.protostream.annotations.ProtoSyntax;
+import org.infinispan.protostream.config.Configuration;
+import org.infinispan.protostream.descriptors.Descriptor;
+import org.infinispan.protostream.descriptors.FileDescriptor;
+import org.infinispan.protostream.impl.parser.ProtostreamProtoParser;
 import org.infinispan.protostream.types.java.CommonTypes;
 import org.keycloak.cluster.infinispan.LockEntry;
 import org.keycloak.cluster.infinispan.LockEntryPredicate;
@@ -40,6 +48,7 @@ import org.keycloak.models.cache.infinispan.authorization.stream.InResourcePredi
 import org.keycloak.models.cache.infinispan.authorization.stream.InResourceServerPredicate;
 import org.keycloak.models.cache.infinispan.authorization.stream.InScopePredicate;
 import org.keycloak.models.cache.infinispan.events.AuthenticationSessionAuthNoteUpdateEvent;
+import org.keycloak.models.cache.infinispan.events.CacheKeyInvalidatedEvent;
 import org.keycloak.models.cache.infinispan.events.ClientAddedEvent;
 import org.keycloak.models.cache.infinispan.events.ClientRemovedEvent;
 import org.keycloak.models.cache.infinispan.events.ClientScopeAddedEvent;
@@ -49,7 +58,6 @@ import org.keycloak.models.cache.infinispan.events.GroupAddedEvent;
 import org.keycloak.models.cache.infinispan.events.GroupMovedEvent;
 import org.keycloak.models.cache.infinispan.events.GroupRemovedEvent;
 import org.keycloak.models.cache.infinispan.events.GroupUpdatedEvent;
-import org.keycloak.models.cache.infinispan.events.CacheKeyInvalidatedEvent;
 import org.keycloak.models.cache.infinispan.events.RealmRemovedEvent;
 import org.keycloak.models.cache.infinispan.events.RealmUpdatedEvent;
 import org.keycloak.models.cache.infinispan.events.RoleAddedEvent;
@@ -92,7 +100,7 @@ import org.keycloak.storage.managers.UserStorageSyncManager;
 
 @ProtoSchema(
         syntax = ProtoSyntax.PROTO3,
-        schemaPackageName = "keycloak",
+        schemaPackageName = Marshalling.PROTO_SCHEMA_PACKAGE,
         schemaFilePath = "proto/generated",
         allowNullFields = true,
 
@@ -203,5 +211,23 @@ public interface KeycloakModelSchema extends GeneratedSchema {
 
     KeycloakModelSchema INSTANCE = new KeycloakModelSchemaImpl();
 
+    /**
+     * Parses a Google Protocol Buffers schema file.
+     */
+    static FileDescriptor parseProtoSchema(String fileContent) {
+        var files = FileDescriptorSource.fromString("a", fileContent);
+        var builder = Configuration.builder();
+        KeycloakIndexSchemaUtil.configureAnnotationProcessor(builder);
+        var parser = new ProtostreamProtoParser(builder.build());
+        return parser.parse(files).get("a");
+    }
 
+    /**
+     * Finds an entity in a Google Protocol Buffers schema file
+     */
+    static Optional<Descriptor> findEntity(FileDescriptor fileDescriptor, String entity) {
+        return fileDescriptor.getMessageTypes().stream()
+                .filter(descriptor -> Objects.equals(entity, descriptor.getFullName()))
+                .findFirst();
+    }
 }

model/infinispan/src/main/java/org/keycloak/marshalling/Marshalling.java

@@ -43,6 +43,8 @@ import org.infinispan.configuration.global.GlobalConfigurationBuilder;
  */
 public final class Marshalling {
 
+    public static final String PROTO_SCHEMA_PACKAGE = "keycloak";
+
     private Marshalling() {
     }
 
@@ -155,4 +157,8 @@ public final class Marshalling {
     public static void configure(ConfigurationBuilder builder) {
         builder.addContextInitializer(KeycloakModelSchema.INSTANCE);
     }
+
+    public static String protoEntity(Class<?> clazz) {
+        return PROTO_SCHEMA_PACKAGE + "." + clazz.getSimpleName();
+    }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/remover/query/ByRealmIdQueryConditionalRemover.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2024 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.changes.remote.remover.query;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.ConditionalRemover;
+import org.keycloak.models.sessions.infinispan.entities.SessionEntity;
+
+/**
+ * A {@link ConditionalRemover} implementation to delete {@link SessionEntity} based on the {@code realmId} value.
+ * <p>
+ * This implementation uses Infinispan Ickle Queries to delete all entries belonging to the realm.
+ *
+ * @param <K> The key's type stored in the {@link RemoteCache}.
+ * @param <V> The value's type stored in the {@link RemoteCache}.
+ */
+public class ByRealmIdQueryConditionalRemover<K, V extends SessionEntity> extends QueryBasedConditionalRemover<K, V> {
+
+    private static final String CONDITION_FMT = "realmId IN (%s)";
+
+    private final String entity;
+    private final List<String> realms;
+
+    public ByRealmIdQueryConditionalRemover(String entity, Executor executor) {
+        super(executor);
+        this.entity = entity;
+        this.realms = new ArrayList<>();
+    }
+
+    private static String parameter(int index) {
+        return "p" + index;
+    }
+
+    public void removeByRealmId(String realmId) {
+        realms.add(realmId);
+    }
+
+    @Override
+    String getEntity() {
+        return entity;
+    }
+
+    @Override
+    String getQueryConditions() {
+        assert !isEmpty();
+        var condition = IntStream.range(0, realms.size())
+                .mapToObj(value -> ":" + parameter(value))
+                .collect(Collectors.joining(", "));
+        return CONDITION_FMT.formatted(condition);
+    }
+
+    @Override
+    Map<String, Object> getQueryParameters() {
+        assert !isEmpty();
+        Map<String, Object> params = new HashMap<>();
+        int paramIdx = 0;
+        for (var realmId : realms) {
+            params.put(parameter(paramIdx++), realmId);
+        }
+        return params;
+    }
+
+    @Override
+    boolean isEmpty() {
+        return realms.isEmpty();
+    }
+
+    @Override
+    public boolean willRemove(K key, V value) {
+        return value != null && realms.contains(value.getRealmId());
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/changes/remote/remover/query/QueryBasedConditionalRemover.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2024 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.changes.remote.remover.query;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.infinispan.commons.util.concurrent.AggregateCompletionStage;
+import org.jboss.logging.Logger;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.ConditionalRemover;
+
+/**
+ * An implementation of {@link ConditionalRemover} that uses the delete statement to remove entries from a
+ * {@link RemoteCache}.
+ * <p>
+ * This class is generic and requires the concrete implementation to provide the entity, the condition clause and the
+ * parameters.
+ *
+ * @param <K> The key's type stored in the {@link RemoteCache}.
+ * @param <V> The value's type stored in the {@link RemoteCache}.
+ */
+abstract class QueryBasedConditionalRemover<K, V> implements ConditionalRemover<K, V> {
+
+    private final static Logger logger = Logger.getLogger(MethodHandles.lookup().lookupClass());
+
+    private static final String QUERY_FMT = "DELETE FROM %s WHERE %s";
+
+    private final Executor executor;
+
+    QueryBasedConditionalRemover(Executor executor) {
+        this.executor = Objects.requireNonNull(executor);
+    }
+
+    @Override
+    public void executeRemovals(RemoteCache<K, V> cache, AggregateCompletionStage<Void> stage) {
+        if (isEmpty()) {
+            return;
+        }
+        // TODO replace with async method: https://issues.redhat.com/browse/ISPN-16279
+        stage.dependsOn(CompletableFuture.runAsync(() -> executeDeleteStatement(cache), executor));
+    }
+
+    private void executeDeleteStatement(RemoteCache<K, V> cache) {
+        var deleteStatement = QUERY_FMT.formatted(getEntity(), getQueryConditions());
+        if (logger.isTraceEnabled()) {
+            logger.tracef("About to execute delete statement in cache '%s': %s", cache.getName(), deleteStatement);
+        }
+        var removed = cache.query(deleteStatement)
+                .setParameters(getQueryParameters())
+                .executeStatement();
+        logger.debugf("Delete Statement removed %d entries from cache '%s'", removed, cache.getName());
+    }
+
+    /**
+     * @return The Infinispan ProtoStream entity.
+     */
+    abstract String getEntity();
+
+    /**
+     * @return The remove condition clause to test.
+     */
+    abstract String getQueryConditions();
+
+    /**
+     * @return The {@link Map} with the parameter name and its value. If the condition does not have any parameter, it
+     * should return an empty map.
+     */
+    abstract Map<String, Object> getQueryParameters();
+
+    /**
+     * @return {@code true} if the concrete implement won't remove anything. This is an optimization to avoid creating
+     * and sending the delete statement.
+     */
+    abstract boolean isEmpty();
+}

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

@@ -19,6 +19,7 @@ package org.keycloak.models.sessions.infinispan.entities;
 
 import java.util.Objects;
 
+import org.infinispan.api.annotations.indexing.Indexed;
 import org.infinispan.protostream.annotations.ProtoFactory;
 import org.infinispan.protostream.annotations.ProtoField;
 import org.infinispan.protostream.annotations.ProtoTypeId;
@@ -28,6 +29,7 @@ import org.keycloak.marshalling.Marshalling;
  * @author <a href="mailto:sthorger@redhat.com">Stian Thorgersen</a>
  */
 @ProtoTypeId(Marshalling.LOGIN_FAILURE_ENTITY)
+@Indexed
 public class LoginFailureEntity extends SessionEntity {
 
     private final String userId;

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

@@ -17,6 +17,7 @@
 
 package org.keycloak.models.sessions.infinispan.entities;
 
+import org.infinispan.api.annotations.indexing.Indexed;
 import org.infinispan.protostream.annotations.ProtoFactory;
 import org.infinispan.protostream.annotations.ProtoField;
 import org.infinispan.protostream.annotations.ProtoTypeId;
@@ -30,6 +31,7 @@ import java.util.concurrent.ConcurrentHashMap;
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
  */
 @ProtoTypeId(Marshalling.ROOT_AUTHENTICATION_SESSION_ENTITY)
+@Indexed
 public class RootAuthenticationSessionEntity extends SessionEntity {
 
     private final String id;

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

@@ -17,6 +17,7 @@
 
 package org.keycloak.models.sessions.infinispan.entities;
 
+import org.infinispan.api.annotations.indexing.Basic;
 import org.infinispan.protostream.annotations.ProtoField;
 import org.keycloak.common.Profile;
 import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
@@ -39,6 +40,7 @@ public abstract class SessionEntity {
      * @return
      */
     @ProtoField(1)
+    @Basic
     public String getRealmId() {
         return realmId;
     }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteInfinispanAuthenticationSessionProviderFactory.java

@@ -20,30 +20,32 @@ package org.keycloak.models.sessions.infinispan.remote;
 import java.lang.invoke.MethodHandles;
 import java.util.List;
 
-import org.infinispan.client.hotrod.RemoteCache;
 import org.jboss.logging.Logger;
 import org.keycloak.Config;
 import org.keycloak.infinispan.util.InfinispanUtils;
+import org.keycloak.marshalling.Marshalling;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
 import org.keycloak.models.sessions.infinispan.InfinispanAuthenticationSessionProviderFactory;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.query.ByRealmIdQueryConditionalRemover;
 import org.keycloak.models.sessions.infinispan.entities.RootAuthenticationSessionEntity;
 import org.keycloak.models.sessions.infinispan.remote.transaction.AuthenticationSessionTransaction;
+import org.keycloak.models.sessions.infinispan.remote.transaction.RemoteCacheAndExecutor;
 import org.keycloak.provider.EnvironmentDependentProviderFactory;
 import org.keycloak.provider.ProviderConfigProperty;
 import org.keycloak.provider.ProviderConfigurationBuilder;
 import org.keycloak.sessions.AuthenticationSessionProviderFactory;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.AUTHENTICATION_SESSIONS_CACHE_NAME;
-import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.getRemoteCache;
 import static org.keycloak.models.sessions.infinispan.InfinispanAuthenticationSessionProviderFactory.DEFAULT_AUTH_SESSIONS_LIMIT;
 
 public class RemoteInfinispanAuthenticationSessionProviderFactory implements AuthenticationSessionProviderFactory<RemoteInfinispanAuthenticationSessionProvider>, EnvironmentDependentProviderFactory {
 
     private final static Logger logger = Logger.getLogger(MethodHandles.lookup().lookupClass());
+    private static final String PROTO_ENTITY = Marshalling.protoEntity(RootAuthenticationSessionEntity.class);
 
     private int authSessionsLimit;
-    private volatile RemoteCache<String, RootAuthenticationSessionEntity> cache;
+    private volatile RemoteCacheAndExecutor<String, RootAuthenticationSessionEntity> cacheHolder;
 
     @Override
     public boolean isSupported(Config.Scope config) {
@@ -62,13 +64,13 @@ public class RemoteInfinispanAuthenticationSessionProviderFactory implements Aut
 
     @Override
     public void postInit(KeycloakSessionFactory factory) {
-        cache = getRemoteCache(factory, AUTHENTICATION_SESSIONS_CACHE_NAME);
+        cacheHolder = RemoteCacheAndExecutor.create(factory, AUTHENTICATION_SESSIONS_CACHE_NAME);
         logger.debugf("Provided initialized. session limit=%s", authSessionsLimit);
     }
 
     @Override
     public void close() {
-        cache = null;
+        cacheHolder = null;
     }
 
     @Override
@@ -94,8 +96,8 @@ public class RemoteInfinispanAuthenticationSessionProviderFactory implements Aut
     }
 
     private AuthenticationSessionTransaction createAndEnlistTransaction(KeycloakSession session) {
-        var tx = new AuthenticationSessionTransaction(cache);
+        var tx = new AuthenticationSessionTransaction(cacheHolder.cache(), new ByRealmIdQueryConditionalRemover<>(PROTO_ENTITY, cacheHolder.executor()));
-        session.getTransactionManager().enlist(tx);
+        session.getTransactionManager().enlistAfterCompletion(tx);
         return tx;
     }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteUserLoginFailureProviderFactory.java

@@ -19,30 +19,32 @@ package org.keycloak.models.sessions.infinispan.remote;
 import java.lang.invoke.MethodHandles;
 
 import org.infinispan.client.hotrod.MetadataValue;
-import org.infinispan.client.hotrod.RemoteCache;
 import org.jboss.logging.Logger;
 import org.keycloak.Config;
 import org.keycloak.infinispan.util.InfinispanUtils;
+import org.keycloak.marshalling.Marshalling;
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
 import org.keycloak.models.UserLoginFailureProvider;
 import org.keycloak.models.UserLoginFailureProviderFactory;
 import org.keycloak.models.UserModel;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.query.ByRealmIdQueryConditionalRemover;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.UpdaterFactory;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.loginfailures.LoginFailuresUpdater;
 import org.keycloak.models.sessions.infinispan.entities.LoginFailureEntity;
 import org.keycloak.models.sessions.infinispan.entities.LoginFailureKey;
 import org.keycloak.models.sessions.infinispan.remote.transaction.LoginFailureChangeLogTransaction;
+import org.keycloak.models.sessions.infinispan.remote.transaction.RemoteCacheAndExecutor;
 import org.keycloak.provider.EnvironmentDependentProviderFactory;
 
 import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.LOGIN_FAILURE_CACHE_NAME;
-import static org.keycloak.connections.infinispan.InfinispanConnectionProvider.getRemoteCache;
 
 public class RemoteUserLoginFailureProviderFactory implements UserLoginFailureProviderFactory<RemoteUserLoginFailureProvider>, UpdaterFactory<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater>, EnvironmentDependentProviderFactory {
 
     private static final Logger log = Logger.getLogger(MethodHandles.lookup().lookupClass());
+    private static final String PROTO_ENTITY = Marshalling.protoEntity(LoginFailureEntity.class);
 
-    private volatile RemoteCache<LoginFailureKey, LoginFailureEntity> cache;
+    private volatile RemoteCacheAndExecutor<LoginFailureKey, LoginFailureEntity> cacheHolder;
 
     @Override
     public RemoteUserLoginFailureProvider create(KeycloakSession session) {
@@ -55,19 +57,19 @@ public class RemoteUserLoginFailureProviderFactory implements UserLoginFailurePr
 
     @Override
     public void postInit(final KeycloakSessionFactory factory) {
-        cache = getRemoteCache(factory, LOGIN_FAILURE_CACHE_NAME);
+        cacheHolder = RemoteCacheAndExecutor.create(factory, LOGIN_FAILURE_CACHE_NAME);
         factory.register(event -> {
             if (event instanceof UserModel.UserRemovedEvent userRemovedEvent) {
                 UserLoginFailureProvider provider = userRemovedEvent.getKeycloakSession().getProvider(UserLoginFailureProvider.class, getId());
                 provider.removeUserLoginFailure(userRemovedEvent.getRealm(), userRemovedEvent.getUser().getId());
             }
         });
-        log.debugf("Post Init. Cache=%s", cache.getName());
+        log.debugf("Post Init. Cache=%s", cacheHolder.cache().getName());
     }
 
     @Override
     public void close() {
-        cache = null;
+        cacheHolder = null;
     }
 
     @Override
@@ -102,7 +104,7 @@ public class RemoteUserLoginFailureProviderFactory implements UserLoginFailurePr
     }
 
     private LoginFailureChangeLogTransaction createAndEnlistTransaction(KeycloakSession session) {
-        var tx = new LoginFailureChangeLogTransaction(this, cache);
+        var tx = new LoginFailureChangeLogTransaction(this, cacheHolder.cache(), new ByRealmIdQueryConditionalRemover<>(PROTO_ENTITY, cacheHolder.executor()));
         session.getTransactionManager().enlistAfterCompletion(tx);
         return tx;
     }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/transaction/AuthenticationSessionTransaction.java

@@ -18,18 +18,18 @@
 package org.keycloak.models.sessions.infinispan.remote.transaction;
 
 import org.infinispan.client.hotrod.RemoteCache;
-import org.keycloak.models.sessions.infinispan.changes.remote.remover.iteration.ByRealmIdConditionalRemover;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.query.ByRealmIdQueryConditionalRemover;
 import org.keycloak.models.sessions.infinispan.entities.RootAuthenticationSessionEntity;
 
 /**
  * Syntactic sugar for
  * {@code RemoteInfinispanKeycloakTransaction<String, RootAuthenticationSessionEntity,
- * ByRealmIdConditionalRemover<String, RootAuthenticationSessionEntity>>}
+ * ByRealmIdQueryConditionalRemover<String, RootAuthenticationSessionEntity>>
  */
-public class AuthenticationSessionTransaction extends RemoteInfinispanKeycloakTransaction<String, RootAuthenticationSessionEntity, ByRealmIdConditionalRemover<String, RootAuthenticationSessionEntity>> {
+public class AuthenticationSessionTransaction extends RemoteInfinispanKeycloakTransaction<String, RootAuthenticationSessionEntity, ByRealmIdQueryConditionalRemover<String, RootAuthenticationSessionEntity>> {
 
-    public AuthenticationSessionTransaction(RemoteCache<String, RootAuthenticationSessionEntity> cache) {
+    public AuthenticationSessionTransaction(RemoteCache<String, RootAuthenticationSessionEntity> cache, ByRealmIdQueryConditionalRemover<String, RootAuthenticationSessionEntity> conditionalRemover) {
-        super(cache, new ByRealmIdConditionalRemover<>());
+        super(cache, conditionalRemover);
     }
 
     public void removeByRealmId(String realmId) {

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/transaction/LoginFailureChangeLogTransaction.java

@@ -18,7 +18,7 @@
 package org.keycloak.models.sessions.infinispan.remote.transaction;
 
 import org.infinispan.client.hotrod.RemoteCache;
-import org.keycloak.models.sessions.infinispan.changes.remote.remover.iteration.ByRealmIdConditionalRemover;
+import org.keycloak.models.sessions.infinispan.changes.remote.remover.query.ByRealmIdQueryConditionalRemover;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.UpdaterFactory;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.loginfailures.LoginFailuresUpdater;
 import org.keycloak.models.sessions.infinispan.entities.LoginFailureEntity;
@@ -29,10 +29,10 @@ import org.keycloak.models.sessions.infinispan.entities.LoginFailureKey;
  * {@code RemoteChangeLogTransaction<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater,
  * ByRealmIdConditionalRemover<LoginFailureKey, LoginFailureEntity>>}
  */
-public class LoginFailureChangeLogTransaction extends RemoteChangeLogTransaction<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater, ByRealmIdConditionalRemover<LoginFailureKey, LoginFailureEntity>> {
+public class LoginFailureChangeLogTransaction extends RemoteChangeLogTransaction<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater, ByRealmIdQueryConditionalRemover<LoginFailureKey, LoginFailureEntity>> {
 
-    public LoginFailureChangeLogTransaction(UpdaterFactory<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater> factory, RemoteCache<LoginFailureKey, LoginFailureEntity> cache) {
+    public LoginFailureChangeLogTransaction(UpdaterFactory<LoginFailureKey, LoginFailureEntity, LoginFailuresUpdater> factory, RemoteCache<LoginFailureKey, LoginFailureEntity> cache, ByRealmIdQueryConditionalRemover<LoginFailureKey, LoginFailureEntity> conditionalRemover) {
-        super(factory, cache, new ByRealmIdConditionalRemover<>());
+        super(factory, cache, conditionalRemover);
     }
 
     public void removeByRealmId(String realmId) {

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/transaction/RemoteCacheAndExecutor.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2024 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.remote.transaction;
+
+import java.util.concurrent.Executor;
+
+import org.infinispan.client.hotrod.RemoteCache;
+import org.keycloak.connections.infinispan.InfinispanConnectionProvider;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.KeycloakSessionFactory;
+
+public record RemoteCacheAndExecutor<K, V>(RemoteCache<K, V> cache, Executor executor) {
+
+    public static <K1, V1> RemoteCacheAndExecutor<K1, V1> create(KeycloakSession session, String cacheName) {
+        var connection = session.getProvider(InfinispanConnectionProvider.class);
+        return new RemoteCacheAndExecutor<>(connection.getRemoteCache(cacheName), connection.getExecutor(cacheName + "-query-delete"));
+    }
+
+    public static <K1, V1> RemoteCacheAndExecutor<K1, V1> create(KeycloakSessionFactory factory, String cacheName) {
+        try (var session = factory.create()) {
+            return create(session, cacheName);
+        }
+    }
+}

model/infinispan/src/test/java/org/keycloak/marshalling/IndexSchemaChangeTest.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2024 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.marshalling;
+
+import org.infinispan.protostream.descriptors.FileDescriptor;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class IndexSchemaChangeTest {
+
+    private static final FileDescriptor V1 = KeycloakModelSchema.parseProtoSchema(TestModelV1.INSTANCE.getProtoFile());
+    private static final FileDescriptor V2 = KeycloakModelSchema.parseProtoSchema(TestModelV2.INSTANCE.getProtoFile());
+
+    @Test
+    public void testNothingChanged() {
+        doTest("keycloak.test.NothingChangedClass", false);
+    }
+
+    @Test
+    public void testNothingChangedIndexed() {
+        doTest("keycloak.test.NothingChangedIndexClass", false);
+    }
+
+    @Test
+    public void testAddIndexedField() {
+        doTest("keycloak.test.AddIndexedFieldClass", true);
+    }
+
+    @Test
+    public void testRemoveIndexedField() {
+        doTest("keycloak.test.RemoveIndexedFieldClass", true);
+    }
+
+    @Test
+    public void testChangedIndexedFieldAttribute() {
+        doTest("keycloak.test.ChangedIndexedFieldAttributeClass", true);
+    }
+
+    @Test
+    public void testChangedIndexedField() {
+        doTest("keycloak.test.ChangedIndexedFieldClass", true);
+    }
+
+    private static void doTest(String entity, boolean expectChanged) {
+        var v1 = KeycloakModelSchema.findEntity(V1, entity);
+        var v2 = KeycloakModelSchema.findEntity(V2, entity);
+        Assert.assertTrue(v1.isPresent());
+        Assert.assertTrue(v2.isPresent());
+        Assert.assertEquals(expectChanged, KeycloakIndexSchemaUtil.isIndexSchemaChanged(v1.get(), v2.get()));
+    }
+}

model/infinispan/src/test/java/org/keycloak/marshalling/TestModelV1.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2024 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.marshalling;
+
+import org.infinispan.api.annotations.indexing.Basic;
+import org.infinispan.api.annotations.indexing.Indexed;
+import org.infinispan.protostream.GeneratedSchema;
+import org.infinispan.protostream.annotations.ProtoField;
+import org.infinispan.protostream.annotations.ProtoSchema;
+import org.infinispan.protostream.annotations.ProtoSyntax;
+
+/**
+ * For {@link IndexSchemaChangeTest}, represents the initial version of the entities.
+ */
+@ProtoSchema(
+        syntax = ProtoSyntax.PROTO3,
+        schemaPackageName = "keycloak.test",
+        schemaFilePath = "proto/generated",
+
+        includeClasses = {
+                TestModelV1.AddIndexedFieldClass.class,
+                TestModelV1.RemoveIndexedFieldClass.class,
+                TestModelV1.ChangedIndexedFieldAttributeClass.class,
+                TestModelV1.ChangedIndexedFieldClass.class,
+                TestModelV1.NothingChangedIndexClass.class,
+                TestModelV1.NothingChangedClass.class
+        },
+
+        service = false
+)
+public interface TestModelV1 extends GeneratedSchema {
+
+    GeneratedSchema INSTANCE = new TestModelV1Impl();
+
+    class AddIndexedFieldClass {
+
+        @ProtoField(1)
+        public String field1;
+
+    }
+
+    @Indexed
+    class RemoveIndexedFieldClass {
+
+        @ProtoField(1)
+        public String field1;
+
+        @ProtoField(2)
+        @Basic
+        public String field2;
+
+    }
+
+    @Indexed
+    class ChangedIndexedFieldAttributeClass {
+
+        @ProtoField(1)
+        @Basic
+        public String field1;
+
+    }
+
+    class ChangedIndexedFieldClass {
+
+        @ProtoField(1)
+        public String field1;
+
+    }
+
+    @Indexed
+    class NothingChangedIndexClass {
+
+        @ProtoField(1)
+        @Basic
+        public String field1;
+
+    }
+
+    class NothingChangedClass {
+
+        @ProtoField(1)
+        public String field1;
+
+    }
+
+}

model/infinispan/src/test/java/org/keycloak/marshalling/TestModelV2.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2024 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.marshalling;
+
+import org.infinispan.api.annotations.indexing.Basic;
+import org.infinispan.api.annotations.indexing.Indexed;
+import org.infinispan.protostream.GeneratedSchema;
+import org.infinispan.protostream.annotations.ProtoField;
+import org.infinispan.protostream.annotations.ProtoSchema;
+import org.infinispan.protostream.annotations.ProtoSyntax;
+
+/**
+ * For {@link IndexSchemaChangeTest}, represents the final version of the entities.
+ */
+@ProtoSchema(
+        syntax = ProtoSyntax.PROTO3,
+        schemaPackageName = "keycloak.test",
+        schemaFilePath = "proto/generated",
+
+        includeClasses = {
+                TestModelV2.AddIndexedFieldClass.class,
+                TestModelV2.RemoveIndexedFieldClass.class,
+                TestModelV2.ChangedIndexedFieldAttributeClass.class,
+                TestModelV2.ChangedIndexedFieldClass.class,
+                TestModelV2.NothingChangedIndexClass.class,
+                TestModelV2.NothingChangedClass.class
+        },
+
+        service = false
+)
+public interface TestModelV2 extends GeneratedSchema {
+
+    GeneratedSchema INSTANCE = new TestModelV2Impl();
+
+    @Indexed
+    class AddIndexedFieldClass {
+
+        @ProtoField(1)
+        public String field1;
+
+        @ProtoField(2)
+        @Basic
+        public String field2;
+
+    }
+
+    class RemoveIndexedFieldClass {
+
+        @ProtoField(1)
+        public String field1;
+
+        @ProtoField(2)
+        public String field2;
+
+    }
+
+    @Indexed
+    class ChangedIndexedFieldAttributeClass {
+
+        @ProtoField(1)
+        @Basic(projectable = true)
+        public String field1;
+
+    }
+
+    @Indexed
+    class ChangedIndexedFieldClass {
+
+        @ProtoField(1)
+        @Basic
+        public String field1;
+
+    }
+
+    @Indexed
+    class NothingChangedIndexClass {
+
+        @ProtoField(1)
+        @Basic
+        public String field1;
+
+    }
+
+    class NothingChangedClass {
+
+        @ProtoField(1)
+        public String field1;
+
+        @ProtoField(2)
+        public String field2;
+
+    }
+
+}

quarkus/runtime/pom.xml

@@ -487,10 +487,6 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.infinispan</groupId>
-            <artifactId>infinispan-cachestore-remote</artifactId>
-        </dependency>
         <dependency>
             <groupId>com.github.ua-parser</groupId>
             <artifactId>uap-java</artifactId>
@@ -584,6 +580,24 @@
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-cachestore-remote</artifactId>
+        </dependency>
+        <!-- required for query/search in the external Infinispan server -->
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-remote-query-client</artifactId>
+        </dependency>
+        <!-- to be removed after https://issues.redhat.com/browse/ISPN-16220 -->
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-query-dsl</artifactId>
+        </dependency>
         <dependency>
             <groupId>jakarta.xml.bind</groupId>
             <artifactId>jakarta.xml.bind-api</artifactId>

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/legacy/infinispan/CacheManagerFactory.java

@@ -20,18 +20,25 @@ package org.keycloak.quarkus.runtime.storage.legacy.infinispan;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Stream;
 
 import io.micrometer.core.instrument.Metrics;
-import org.infinispan.client.hotrod.DefaultTemplate;
+import org.infinispan.client.hotrod.RemoteCache;
 import org.infinispan.client.hotrod.RemoteCacheManager;
+import org.infinispan.client.hotrod.RemoteCacheManagerAdmin;
 import org.infinispan.client.hotrod.impl.ConfigurationProperties;
 import org.infinispan.commons.api.Lifecycle;
+import org.infinispan.commons.dataconversion.MediaType;
+import org.infinispan.commons.internal.InternalCacheNames;
 import org.infinispan.commons.util.concurrent.CompletableFutures;
+import org.infinispan.configuration.cache.CacheMode;
 import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.configuration.cache.HashConfiguration;
 import org.infinispan.configuration.cache.PersistenceConfigurationBuilder;
@@ -42,6 +49,8 @@ import org.infinispan.manager.DefaultCacheManager;
 import org.infinispan.metrics.config.MicrometerMeterRegisterConfigurationBuilder;
 import org.infinispan.persistence.remote.configuration.ExhaustedAction;
 import org.infinispan.persistence.remote.configuration.RemoteStoreConfigurationBuilder;
+import org.infinispan.protostream.descriptors.FileDescriptor;
+import org.infinispan.query.remote.client.ProtobufMetadataManagerConstants;
 import org.infinispan.remoting.transport.jgroups.JGroupsTransport;
 import org.jboss.logging.Logger;
 import org.jgroups.protocols.TCP_NIO2;
@@ -52,7 +61,12 @@ import org.keycloak.common.Profile;
 import org.keycloak.config.CachingOptions;
 import org.keycloak.config.MetricsOptions;
 import org.keycloak.infinispan.util.InfinispanUtils;
+import org.keycloak.marshalling.KeycloakIndexSchemaUtil;
+import org.keycloak.marshalling.KeycloakModelSchema;
 import org.keycloak.marshalling.Marshalling;
+import org.keycloak.models.sessions.infinispan.RootAuthenticationSessionAdapter;
+import org.keycloak.models.sessions.infinispan.entities.LoginFailureEntity;
+import org.keycloak.models.sessions.infinispan.entities.RootAuthenticationSessionEntity;
 import org.keycloak.quarkus.runtime.configuration.Configuration;
 
 import javax.net.ssl.SSLContext;
@@ -156,15 +170,14 @@ public class CacheManagerFactory {
 
         Marshalling.configure(builder);
 
-        if (createRemoteCaches()) {
+        if (shouldCreateRemoteCaches()) {
-            // fall back for distributed caches if not defined
+            createRemoteCaches(builder);
-            logger.warn("Creating remote cache in external Infinispan server. It should not be used in production!");
-            for (String name : CLUSTERED_CACHE_NAMES) {
-                builder.remoteCache(name).templateName(DefaultTemplate.DIST_SYNC);
-            }
         }
 
-        RemoteCacheManager remoteCacheManager = new RemoteCacheManager(builder.build());
+        var remoteCacheManager = new RemoteCacheManager(builder.build());
+
+        // update the schema before trying to access the caches
+        updateProtoSchema(remoteCacheManager);
 
         // establish connection to all caches
         if (isStartEagerly()) {
@@ -173,6 +186,105 @@ public class CacheManagerFactory {
         return remoteCacheManager;
     }
 
+    private static void createRemoteCaches(org.infinispan.client.hotrod.configuration.ConfigurationBuilder builder) {
+        // fall back for distributed caches if not defined
+        logger.warn("Creating remote cache in external Infinispan server. It should not be used in production!");
+        var baseConfig = defaultRemoteCacheBuilder().build();
+
+        Stream.of(USER_SESSION_CACHE_NAME, OFFLINE_USER_SESSION_CACHE_NAME, CLIENT_SESSION_CACHE_NAME, OFFLINE_CLIENT_SESSION_CACHE_NAME, ACTION_TOKEN_CACHE, WORK_CACHE_NAME)
+                .forEach(name -> builder.remoteCache(name).configuration(baseConfig.toStringConfiguration(name)));
+
+        // We need indexed caches because the delete statement fails for non-indexed cache.
+        createIndexedRemoteCache(builder, LOGIN_FAILURE_CACHE_NAME, List.of(LoginFailureEntity.class));
+        createIndexedRemoteCache(builder, AUTHENTICATION_SESSIONS_CACHE_NAME, List.of(RootAuthenticationSessionEntity.class));
+    }
+
+    private static void createIndexedRemoteCache(org.infinispan.client.hotrod.configuration.ConfigurationBuilder builder, String name, List<Class<?>> entities) {
+        var config = indexedRemoteCacheBuilder(entities).build();
+        builder.remoteCache(name).configuration(config.toStringConfiguration(name));
+    }
+
+    private static ConfigurationBuilder defaultRemoteCacheBuilder() {
+        var builder = new ConfigurationBuilder();
+        builder.clustering().cacheMode(CacheMode.DIST_SYNC);
+        builder.encoding().mediaType(MediaType.APPLICATION_PROTOSTREAM);
+        return builder;
+    }
+
+    private static ConfigurationBuilder indexedRemoteCacheBuilder(List<Class<?>> entities) {
+        var builder = defaultRemoteCacheBuilder();
+        var indexBuilder = builder.indexing().enable();
+        entities.stream()
+                .map(Marshalling::protoEntity)
+                .forEach(indexBuilder::addIndexedEntity);
+        return builder;
+    }
+
+    private void updateProtoSchema(RemoteCacheManager remoteCacheManager) {
+        var key = KeycloakModelSchema.INSTANCE.getProtoFileName();
+        var current = KeycloakModelSchema.INSTANCE.getProtoFile();
+
+        RemoteCache<String, String> protostreamMetadataCache = remoteCacheManager.getCache(InternalCacheNames.PROTOBUF_METADATA_CACHE_NAME);
+        var stored = protostreamMetadataCache.getWithMetadata(key);
+        if (stored == null) {
+            if (protostreamMetadataCache.putIfAbsent(key, current) == null) {
+                logger.info("Infinispan Protostream schema uploaded for the first time.");
+            } else {
+                logger.info("Failed to update Infinispan Protostream schema. Assumed it was updated by other Keycloak server.");
+            }
+            checkForProtoSchemaErrors(protostreamMetadataCache);
+            return;
+        }
+        if (Objects.equals(stored.getValue(), current)) {
+            logger.info("Infinispan Protostream schema is up to date!");
+            return;
+        }
+        if (protostreamMetadataCache.replaceWithVersion(key, current, stored.getVersion())) {
+            logger.info("Infinispan Protostream schema successful updated.");
+            reindexCaches(remoteCacheManager, stored.getValue(), current);
+        } else {
+            logger.info("Failed to update Infinispan Protostream schema. Assumed it was updated by other Keycloak server.");
+        }
+        checkForProtoSchemaErrors(protostreamMetadataCache);
+    }
+
+    private void checkForProtoSchemaErrors(RemoteCache<String, String> protostreamMetadataCache) {
+        String errors = protostreamMetadataCache.get(ProtobufMetadataManagerConstants.ERRORS_KEY_SUFFIX);
+        if (errors != null) {
+            for (String errorFile : errors.split("\n")) {
+                logger.errorf("%nThere was an error in proto file: %s%nError message: %s%nCurrent proto schema: %s%n",
+                        errorFile,
+                        protostreamMetadataCache.get(errorFile + ProtobufMetadataManagerConstants.ERRORS_KEY_SUFFIX),
+                        protostreamMetadataCache.get(errorFile));
+            }
+        }
+    }
+
+    private static void reindexCaches(RemoteCacheManager remoteCacheManager, String oldSchema, String newSchema) {
+        var oldPS = KeycloakModelSchema.parseProtoSchema(oldSchema);
+        var newPS = KeycloakModelSchema.parseProtoSchema(newSchema);
+        var admin = remoteCacheManager.administration();
+
+        if (isEntityChanged(oldPS, newPS, Marshalling.protoEntity(LoginFailureEntity.class))) {
+            updateSchemaAndReIndexCache(admin, LOGIN_FAILURE_CACHE_NAME);
+        }
+
+        if (isEntityChanged(oldPS, newPS, Marshalling.protoEntity(RootAuthenticationSessionAdapter.class))) {
+            updateSchemaAndReIndexCache(admin, AUTHENTICATION_SESSIONS_CACHE_NAME);
+        }
+    }
+
+    private static boolean isEntityChanged(FileDescriptor oldSchema, FileDescriptor newSchema, String entity) {
+        var v1 = KeycloakModelSchema.findEntity(oldSchema, entity);
+        var v2 = KeycloakModelSchema.findEntity(newSchema, entity);
+        return v1.isPresent() && v2.isPresent() && KeycloakIndexSchemaUtil.isIndexSchemaChanged(v1.get(), v2.get());
+    }
+
+    private static void updateSchemaAndReIndexCache(RemoteCacheManagerAdmin admin, String cacheName) {
+        admin.updateIndexSchema(cacheName);
+        admin.reindexCache(cacheName);
+    }
+
     private CompletableFuture<DefaultCacheManager> startEmbeddedCacheManager(String config) {
         logger.info("Starting Infinispan embedded cache manager");
         ConfigurationBuilderHolder builder = new ParserRegistry().parse(config);
@@ -250,7 +362,7 @@ public class CacheManagerFactory {
                 Configuration.getOptionalKcValue(CACHE_REMOTE_PASSWORD_PROPERTY).isPresent();
     }
 
-    private static boolean createRemoteCaches() {
+    private static boolean shouldCreateRemoteCaches() {
         return Boolean.getBoolean("kc.cache-remote-create-caches");
     }
 

testsuite/model/pom.xml

@@ -286,6 +286,12 @@
                     </plugin>
                 </plugins>
             </build>
+            <dependencies>
+                <dependency>
+                    <groupId>org.infinispan</groupId>
+                    <artifactId>infinispan-remote-query-server</artifactId>
+                </dependency>
+            </dependencies>
         </profile>
 
         <profile>