mhajas · Jun 13, 2024 · Mar 29, 2024
@@ -9,3 +9,12 @@ to different organizations.
 When creating a social broker, you should now provide an `Alias` and optionally a `Display name` just like any other
 broker.
 
+= Infinispan marshalling changes to Infinispan Protostream
+
+Marshalling is the process of converting Java objects into bytes to send them across the network between {project_name} servers.
+With {project_name} 26, we changed the marshalling format from JBoss Marshalling to Infinispan Protostream.
+
+WARNING: JBoss Marshalling and Infinispan Protostream are not compatible with each other and incorrect usage may lead to data loss.
+Consequently, all caches are cleared when upgrading to this version.
+
+Infinispan Protostream is based on https://protobuf.dev/programming-guides/proto3/[Protocol Buffers] (proto 3), which has the advantage of backwards/forwards compatibility.
@@ -0,0 +1,10 @@
+= Infinispan marshalling changes
+
+Marshalling is the process of converting Java objects into bytes to send them across the network between {project_name} servers.
+With {project_name} 26, the marshalling library has changed from JBoss Marshalling to Infinispan Protostream.
+The libraries are not compatible between each other and, it requires some steps to ensure the session data is not lost.
+
+WARNING: JBoss Marshalling and Infinispan Protostream are not compatible with each other and incorrect usage may lead to data loss.
+Consequently, all caches are cleared when upgrading to this version.
+
+To prevent losing user sessions upgrade to Keycloak 25 first and enable the persistent sessions feature as outlined in the migration guide for {project_name} 25.
@@ -1,6 +1,10 @@
 [[migration-changes]]
 == Migration Changes
 
+=== Migrating to 26.0.0
+
+include::changes-26_0_0.adoc[leveloffset=3]
+
 === Migrating to 25.0.0
 
 include::changes-25_0_0.adoc[leveloffset=3]

@@ -67,7 +67,23 @@
         </dependency>
         <dependency>
             <groupId>org.infinispan</groupId>
-            <artifactId>infinispan-jboss-marshalling</artifactId>
+            <artifactId>infinispan-component-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.infinispan.protostream</groupId>
+            <artifactId>protostream</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan.protostream</groupId>
+            <artifactId>protostream-types</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan.protostream</groupId>
+            <artifactId>protostream-processor</artifactId>
+            <!-- compile-only dependency -->
+            <scope>provided</scope>
         </dependency>
 
         <dependency>

@@ -17,7 +17,6 @@
 
 package org.keycloak.cluster.infinispan;
 
-import java.io.Serializable;
 import java.util.Set;
 
 import org.infinispan.Cache;
@@ -35,10 +34,10 @@ abstract class CrossDCAwareCacheFactory {
     protected static final Logger logger = Logger.getLogger(CrossDCAwareCacheFactory.class);
 
 
-    abstract BasicCache<String, Serializable> getCache();
+    abstract BasicCache<String, Object> getCache();
 
 
-    static CrossDCAwareCacheFactory getFactory(Cache<String, Serializable> workCache, Set<RemoteStore> remoteStores) {
+    static CrossDCAwareCacheFactory getFactory(Cache<String, Object> workCache, Set<RemoteStore> remoteStores) {
         if (remoteStores.isEmpty()) {
             logger.debugf("No configured remoteStore available. Cross-DC scenario is not used");
             return new InfinispanCacheWrapperFactory(workCache);
@@ -66,14 +65,14 @@ static CrossDCAwareCacheFactory getFactory(Cache<String, Serializable> workCache
     // We don't have external JDG configured. No cross-DC.
     private static class InfinispanCacheWrapperFactory extends CrossDCAwareCacheFactory {
 
-        private final Cache<String, Serializable> workCache;
+        private final Cache<String, Object> workCache;
 
-        InfinispanCacheWrapperFactory(Cache<String, Serializable> workCache) {
+        InfinispanCacheWrapperFactory(Cache<String, Object> workCache) {
             this.workCache = workCache;
         }
 
         @Override
-        BasicCache<String, Serializable> getCache() {
+        BasicCache<String, Object> getCache() {
             return workCache;
         }
 
@@ -83,14 +82,14 @@ BasicCache<String, Serializable> getCache() {
     // We have external JDG configured. Cross-DC should be enabled
     private static class RemoteCacheWrapperFactory extends CrossDCAwareCacheFactory {
 
-        private final RemoteCache<String, Serializable> remoteCache;
+        private final RemoteCache<String, Object> remoteCache;
 
-        RemoteCacheWrapperFactory(RemoteCache<String, Serializable> remoteCache) {
+        RemoteCacheWrapperFactory(RemoteCache<String, Object> remoteCache) {
             this.remoteCache = remoteCache;
         }
 
         @Override
-        BasicCache<String, Serializable> getCache() {
+        BasicCache<String, Object> getCache() {
             // Flags are per-invocation!
             return remoteCache.withFlags(Flag.FORCE_RETURN_VALUE);
         }

@@ -17,18 +17,17 @@
 
 package org.keycloak.cluster.infinispan;
 
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
 import org.jboss.logging.Logger;
 import org.keycloak.cluster.ClusterEvent;
 import org.keycloak.cluster.ClusterListener;
 import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.cluster.ExecutionResult;
 import org.keycloak.common.util.Retry;
-import org.keycloak.common.util.Time;
-
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 
 /**
  *
@@ -130,26 +129,18 @@ public void notify(String taskKey, ClusterEvent event, boolean ignoreSender, DCN
         this.notificationsManager.notify(taskKey, event, ignoreSender, dcNotify);
     }
 
-    private LockEntry createLockEntry() {
-        LockEntry lock = new LockEntry();
-        lock.setNode(myAddress);
-        lock.setTimestamp(Time.currentTime());
-        return lock;
-    }
-
-
     private boolean tryLock(String cacheKey, int taskTimeoutInSeconds) {
-        LockEntry myLock = createLockEntry();
+        LockEntry myLock = new LockEntry(myAddress);
 
         LockEntry existingLock = InfinispanClusterProviderFactory.putIfAbsentWithRetries(crossDCAwareCacheFactory, cacheKey, myLock, taskTimeoutInSeconds);
         if (existingLock != null) {
             if (logger.isTraceEnabled()) {
-                logger.tracef("Task %s in progress already by node %s. Ignoring task.", cacheKey, existingLock.getNode());
+                logger.tracef("Task %s in progress already by node %s. Ignoring task.", cacheKey, existingLock.node());
             }
             return false;
         } else {
             if (logger.isTraceEnabled()) {
-                logger.tracef("Successfully acquired lock for task %s. Our node is %s", cacheKey, myLock.getNode());
+                logger.tracef("Successfully acquired lock for task %s. Our node is %s", cacheKey, myLock.node());
             }
             return true;
         }

@@ -38,7 +38,6 @@
 import org.keycloak.models.KeycloakSession;
 import org.keycloak.models.KeycloakSessionFactory;
 
-import java.io.Serializable;
 import java.util.Collection;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -59,7 +58,7 @@ public class InfinispanClusterProviderFactory implements ClusterProviderFactory
     protected static final Logger logger = Logger.getLogger(InfinispanClusterProviderFactory.class);
 
     // Infinispan cache
-    private volatile Cache<String, Serializable> workCache;
+    private volatile Cache<String, Object> workCache;
 
     // Ensure that atomic operations (like putIfAbsent) must work correctly in any of: non-clustered, clustered or cross-Data-Center (cross-DC) setups
     private CrossDCAwareCacheFactory crossDCAwareCacheFactory;
@@ -134,7 +133,7 @@ protected int initClusterStartupTime(KeycloakSession session) {
 
     // Will retry few times for the case when backup site not available in cross-dc environment.
     // The site might be taken offline automatically if "take-offline" properly configured
-    static <V extends Serializable> V putIfAbsentWithRetries(CrossDCAwareCacheFactory crossDCAwareCacheFactory, String key, V value, int taskTimeoutInSeconds) {
+    static <V> V putIfAbsentWithRetries(CrossDCAwareCacheFactory crossDCAwareCacheFactory, String key, V value, int taskTimeoutInSeconds) {
         AtomicReference<V> resultRef = new AtomicReference<>();
 
         Retry.executeWithBackoff(iteration -> {

@@ -75,7 +75,7 @@ public class InfinispanNotificationsManager {
 
     private final ConcurrentMap<String, TaskCallback> taskCallbacks = new ConcurrentHashMap<>();
 
-    private final Cache<String, Serializable> workCache;
+    private final Cache<String, Object> workCache;
 
     private final RemoteCache<Object, Serializable> workRemoteCache;
 
@@ -85,7 +85,7 @@ public class InfinispanNotificationsManager {
 
     private final ExecutorService listenersExecutor;
 
-    protected InfinispanNotificationsManager(Cache<String, Serializable> workCache, RemoteCache<Object, Serializable> workRemoteCache, String myAddress, String mySite, ExecutorService listenersExecutor) {
+    protected InfinispanNotificationsManager(Cache<String, Object> workCache, RemoteCache<Object, Serializable> workRemoteCache, String myAddress, String mySite, ExecutorService listenersExecutor) {
         this.workCache = workCache;
         this.workRemoteCache = workRemoteCache;
         this.myAddress = myAddress;
@@ -95,7 +95,7 @@ protected InfinispanNotificationsManager(Cache<String, Serializable> workCache,
 
 
     // Create and init manager including all listeners etc
-    public static InfinispanNotificationsManager create(KeycloakSession session, Cache<String, Serializable> workCache, String myAddress, String mySite, Set<RemoteStore> remoteStores) {
+    public static InfinispanNotificationsManager create(KeycloakSession session, Cache<String, Object> workCache, String myAddress, String mySite, Set<RemoteStore> remoteStores) {
         RemoteCache<Object, Serializable> workRemoteCache = null;
 
         if (!remoteStores.isEmpty()) {
@@ -141,13 +141,7 @@ TaskCallback registerTaskCallback(String taskKey, TaskCallback callback) {
 
 
     void notify(String taskKey, ClusterEvent event, boolean ignoreSender, ClusterProvider.DCNotify dcNotify) {
-        WrapperClusterEvent wrappedEvent = new WrapperClusterEvent();
-        wrappedEvent.setEventKey(taskKey);
-        wrappedEvent.setDelegateEvent(event);
-        wrappedEvent.setIgnoreSender(ignoreSender);
-        wrappedEvent.setIgnoreSenderSite(dcNotify == ClusterProvider.DCNotify.ALL_BUT_LOCAL_DC);
-        wrappedEvent.setSender(myAddress);
-        wrappedEvent.setSenderSite(mySite);
+        var wrappedEvent = WrapperClusterEvent.wrap(taskKey, event, myAddress, mySite, dcNotify, ignoreSender);
 
         String eventKey = UUID.randomUUID().toString();
 
@@ -186,17 +180,17 @@ void notify(String taskKey, ClusterEvent event, boolean ignoreSender, ClusterPro
     public class CacheEntryListener {
 
         @CacheEntryCreated
-        public void cacheEntryCreated(CacheEntryCreatedEvent<String, Serializable> event) {
+        public void cacheEntryCreated(CacheEntryCreatedEvent<String, Object> event) {
             eventReceived(event.getKey(), event.getValue());
         }
 
         @CacheEntryModified
-        public void cacheEntryModified(CacheEntryModifiedEvent<String, Serializable> event) {
+        public void cacheEntryModified(CacheEntryModifiedEvent<String, Object> event) {
             eventReceived(event.getKey(), event.getNewValue());
         }
 
         @CacheEntryRemoved
-        public void cacheEntryRemoved(CacheEntryRemovedEvent<String, Serializable> event) {
+        public void cacheEntryRemoved(CacheEntryRemovedEvent<String, Object> event) {
             taskFinished(event.getKey());
         }
 
@@ -267,7 +261,7 @@ private void hotrodEventReceived(String key) {
         }
     }
 
-    private void eventReceived(String key, Serializable obj) {
+    private void eventReceived(String key, Object obj) {
         if (!(obj instanceof WrapperClusterEvent event)) {
             // Items with the TASK_KEY_PREFIX might be gone fast once the locking is complete, therefore, don't log them.
             // It is still good to have the warning in case of real events return null because they have been, for example, expired
@@ -277,16 +271,8 @@ private void eventReceived(String key, Serializable obj) {
             return;
         }
 
-        if (event.isIgnoreSender()) {
-            if (this.myAddress.equals(event.getSender())) {
-                return;
-            }
-        }
-
-        if (event.isIgnoreSenderSite()) {
-            if (this.mySite == null || this.mySite.equals(event.getSenderSite())) {
-                return;
-            }
+        if (event.rejectEvent(myAddress, mySite)) {
+            return;
         }
 
         String eventKey = event.getEventKey();

@@ -17,67 +17,14 @@
 
 package org.keycloak.cluster.infinispan;
 
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.io.Serializable;
-
-import org.infinispan.commons.marshall.Externalizer;
-import org.infinispan.commons.marshall.MarshallUtil;
-import org.infinispan.commons.marshall.SerializeWith;
-import org.keycloak.models.sessions.infinispan.util.KeycloakMarshallUtil;
+import org.infinispan.protostream.annotations.Proto;
+import org.infinispan.protostream.annotations.ProtoTypeId;
+import org.keycloak.marshalling.Marshalling;
 
 /**
  * @author <a href="mailto:mposolda@redhat.com">Marek Posolda</a>
  */
-@SerializeWith(LockEntry.ExternalizerImpl.class)
-public class LockEntry implements Serializable {
-
-    private String node;
-    private int timestamp;
-
-    public String getNode() {
-        return node;
-    }
-
-    public void setNode(String node) {
-        this.node = node;
-    }
-
-    public int getTimestamp() {
-        return timestamp;
-    }
-
-    public void setTimestamp(int timestamp) {
-        this.timestamp = timestamp;
-    }
-
-    public static class ExternalizerImpl implements Externalizer<LockEntry> {
-
-        private static final int VERSION_1 = 1;
-
-        @Override
-        public void writeObject(ObjectOutput output, LockEntry obj) throws IOException {
-            output.writeByte(VERSION_1);
-            MarshallUtil.marshallString(obj.node, output);
-            KeycloakMarshallUtil.marshall(obj.timestamp, output);
-        }
-
-        @Override
-        public LockEntry readObject(ObjectInput input) throws IOException, ClassNotFoundException {
-            switch (input.readByte()) {
-                case VERSION_1:
-                    return readObjectVersion1(input);
-                default:
-                    throw new IOException("Unknown version");
-            }
-        }
-
-        public LockEntry readObjectVersion1(ObjectInput input) throws IOException, ClassNotFoundException {
-            LockEntry entry = new LockEntry();
-            entry.setNode(MarshallUtil.unmarshallString(input));
-            entry.setTimestamp(KeycloakMarshallUtil.unmarshallInteger(input));
-            return entry;
-        }
-    }
+@ProtoTypeId(Marshalling.LOCK_ENTRY)
+@Proto
+public record LockEntry(String node) {
 }