Change Delete by Id with @SoftDelete to performs soft delete rather than hard delete

Co-authored-by: robin.bygrave <robin.bygrave@eroad.com>

ebean-core/src/main/java/io/ebeaninternal/server/core/DefaultServer.java

@@ -1242,6 +1242,9 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer {
   }
 
   private <T> int delete(SpiQuery<T> query, boolean permanent) {
+    if (query.getId() != null && query.whereExpressions() == null && query.descriptor().isSoftDelete()) {
+      return executeInTrans((txn) -> persister.delete(query.getBeanType(), query.getId(), txn, permanent), query.transaction());
+    }
     SpiOrmQueryRequest<T> request = createQueryRequest(Type.DELETE, query);
     try {
       request.initTransIfRequired();

ebean-core/src/main/java/io/ebeaninternal/server/core/PersistRequestBean.java

@@ -132,6 +132,13 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
    * Cascade to children must be suppressed to avoid FK violations.
    */
   private boolean insertConflictSkipped;
+  /**
+   * Set true once controller.preDelete() has been invoked so that it is
+   * only ever fired once (as it is fired early, prior to cascading the
+   * delete to children/many's rather than as part of executing the delete).
+   */
+  private boolean preDeleteCalled;
+  private boolean preDeleteResult = true;
 
   public PersistRequestBean(SpiEbeanServer server, T bean, Object parentBean, BeanManager<T> mgr, SpiTransaction t,
                             PersistExecute persistExecute, PersistRequest.Type type, int flags) {
@@ -1276,14 +1283,29 @@ public final class PersistRequestBean<T> extends PersistRequest implements BeanP
   }
 
   private int executeDelete() {
-    setTenantId();
-    if (controller == null || controller.preDelete(this)) {
+    if (controllerPreDelete()) {
       return beanManager.getBeanPersister().delete(this);
     }
     // delete handled by the BeanController so return 0
     return 0;
   }
 
+  /**
+   * Invoke controller.preDelete() if not already invoked.
+   * <p>
+   * This is called prior to cascading the delete to children (assoc many's /
+   * many-to-many intersection rows) so that the persist controller can still
+   * see those collections/relationships as they were before the cascade delete.
+   */
+  public boolean controllerPreDelete() {
+    if (!preDeleteCalled) {
+      preDeleteCalled = true;
+      setTenantId();
+      preDeleteResult = controller == null || controller.preDelete(this);
+    }
+    return preDeleteResult;
+  }
+
   /**
    * Persist to the document store now (via buffer, not post commit).
    */

ebean-core/src/main/java/io/ebeaninternal/server/persist/DefaultPersister.java

@@ -907,6 +907,11 @@ public final class DefaultPersister implements Persister {
    * </p>
    */
   private int delete(PersistRequestBean<?> request) {
+    // fire preDelete now, before cascading to children/many's so that the
+    // BeanPersistController/Adapter still sees the bean's collections and
+    // relationships as they are prior to the cascade delete
+    request.controllerPreDelete();
+
     DeleteUnloadedForeignKeys unloadedForeignKeys = null;
     if (request.isPersistCascade()) {
       // delete children first ... register the

ebean-redisson/pom.xml

@@ -0,0 +1,100 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <artifactId>ebean-parent</artifactId>
+    <groupId>io.ebean</groupId>
+    <version>18.2.0</version>
+  </parent>
+
+  <artifactId>ebean-redisson</artifactId>
+  <name>ebean redisson</name>
+  <description>Ebean Redis L2 Cache (Redisson implementation)</description>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>ch.qos.logback</groupId>
+      <artifactId>logback-classic</artifactId>
+      <version>1.5.6</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.redisson</groupId>
+      <artifactId>redisson</artifactId>
+      <version>4.3.0</version>
+    </dependency>
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>ebean-api</artifactId>
+      <version>18.2.0</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>ebean-core</artifactId>
+      <version>18.2.0</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>ebean-querybean</artifactId>
+      <version>18.2.0</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>ebean-test</artifactId>
+      <version>18.2.0</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.ebean</groupId>
+      <artifactId>querybean-generator</artifactId>
+      <version>18.2.0</version>
+      <scope>provided</scope>
+    </dependency>
+
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>io.ebean</groupId>
+        <artifactId>ebean-maven-plugin</artifactId>
+        <version>${ebean-maven-plugin.version}</version>
+        <executions>
+          <execution>
+            <id>test</id>
+            <phase>process-test-classes</phase>
+            <configuration>
+              <transformArgs>debug=0</transformArgs>
+            </configuration>
+            <goals>
+              <goal>testEnhance</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-source-plugin</artifactId>
+        <version>3.2.1</version>
+        <executions>
+          <execution>
+            <id>attach-sources</id>
+            <goals>
+              <goal>jar</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+</project>

ebean-redisson/src/main/java/io/ebean/redisson/DuelCache.java

@@ -0,0 +1,145 @@
+package io.ebean.redisson;
+
+import io.ebean.cache.ServerCache;
+import io.ebean.redisson.near.NearCacheInvalidate;
+import io.ebean.redisson.near.NearCacheNotify;
+import io.ebean.meta.MetricVisitor;
+import io.ebeaninternal.server.cache.DefaultServerCache;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public final class DuelCache implements ServerCache, NearCacheInvalidate {
+
+    private final DefaultServerCache near;
+    private final RedissonCache remote;
+    private final NearCacheNotify cacheNotify;
+    private final String cacheKey;
+
+    public DuelCache(DefaultServerCache near, RedissonCache remote, String cacheKey, NearCacheNotify cacheNotify) {
+        this.near = near;
+        this.remote = remote;
+        this.cacheKey = cacheKey;
+        this.cacheNotify = cacheNotify;
+    }
+
+    @Override
+    public void visit(MetricVisitor visitor) {
+        near.visit(visitor);
+        remote.visit(visitor);
+    }
+
+    @Override
+    public void invalidateKeys(Set<Object> keySet) {
+        near.removeAll(keySet);
+    }
+
+    @Override
+    public void invalidateKey(Object id) {
+        near.remove(id);
+    }
+
+    @Override
+    public void invalidateClear() {
+        near.clear();
+    }
+
+    @Override
+    public Map<Object, Object> getAll(Set<Object> keys) {
+        Map<Object, Object> resultMap = near.getAll(keys);
+        Set<Object> localKeys = resultMap.keySet();
+        Set<Object> remainingKeys = new HashSet<>();
+        for (Object key : keys) {
+            if (!localKeys.contains(key)) {
+                remainingKeys.add(key);
+            }
+        }
+        if (!remainingKeys.isEmpty()) {
+            // fetch missing ones from a remote cache and merge results
+            Map<Object, Object> remoteMap = remote.getAll(remainingKeys);
+            if (!remoteMap.isEmpty()) {
+                near.putAll(remoteMap);
+                resultMap.putAll(remoteMap);
+            }
+        }
+        return resultMap;
+    }
+
+    @Override
+    public Object get(Object id) {
+        Object val = near.get(id);
+        if (val != null) {
+            return val;
+        }
+        Object remoteVal = remote.get(id);
+        if (remoteVal != null) {
+            near.put(id, remoteVal);
+        }
+        return remoteVal;
+    }
+
+    @Override
+    public void putAll(Map<Object, Object> keyValues) {
+        near.putAll(keyValues);
+        remote.putAll(keyValues);
+        cacheNotify.invalidateKeys(cacheKey, keyValues.keySet());
+    }
+
+    @Override
+    public void put(Object id, Object value) {
+        near.put(id, value);
+        remote.put(id, value);
+        cacheNotify.invalidateKey(cacheKey, id);
+    }
+
+    @Override
+    public void removeAll(Set<Object> keys) {
+        near.removeAll(keys);
+        remote.removeAll(keys);
+        cacheNotify.invalidateKeys(cacheKey, keys);
+    }
+
+    @Override
+    public void remove(Object id) {
+        near.remove(id);
+        remote.remove(id);
+        cacheNotify.invalidateKey(cacheKey, id);
+    }
+
+    @Override
+    public void clear() {
+        near.clear();
+        remote.clear();
+        cacheNotify.invalidateClear(cacheKey);
+    }
+
+    /**
+     * Return the near cache hit count.
+     */
+    public long getNearHitCount() {
+        return near.getHitCount();
+    }
+
+    /**
+     * Return the near cache miss count.
+     */
+    public long getNearMissCount() {
+        return near.getMissCount();
+    }
+
+    /**
+     * Return the redis cache hit count.
+     */
+    public long getRemoteHitCount() {
+        return remote.getHitCount();
+    }
+
+    /**
+     * Return the redis cache miss count.
+     */
+    public long getRemoteMissCount() {
+        return remote.getMissCount();
+    }
+
+}

ebean-redisson/src/main/java/io/ebean/redisson/ModId.java

@@ -0,0 +1,48 @@
+package io.ebean.redisson;
+
+import java.security.SecureRandom;
+import java.util.Base64;
+
+/**
+ * Provides a modified base64 encoded UUID and shorter 12 character random unique value.
+ * <p>
+ * <h3>newId()</h3>
+ * <p>
+ * It produces a 22 character string that is a base64 encoded UUID with the +
+ * and / characters replaced with - and _ so as to be URL safe without requiring
+ * encoding.
+ * </p>
+ * <h3>newShortId()</h3>
+ * <p>
+ * It produces a 12 character string that base64 encoded random number (72 bit).
+ * </p>
+ * <p>
+ * Note that this now internally uses java.util.Base64 to encode the values.
+ * </p>
+ */
+public final class ModId {
+
+    private static final SecureRandom shortIdSecureRandom = new SecureRandom();
+
+    private static final Base64.Encoder urlEncoder = Base64.getUrlEncoder();
+
+    /**
+     * Return a 12 character string using a 72 bit randomly generated ID encoded
+     * in modified base64.
+     * <p>
+     * A UUID is 128 bits and this is 72 bits so quite a bit smaller but still
+     * very random with one in 4.7 * 10^21 chance of a collision.
+     * </p>
+     */
+    public static String id() {
+        // Random 72 bits
+        byte[] randomBytes = new byte[9];
+        shortIdSecureRandom.nextBytes(randomBytes);
+        return encode64(randomBytes);
+    }
+
+    private static String encode64(byte[] bytes) {
+        return urlEncoder.encodeToString(bytes);
+    }
+
+}

ebean-redisson/src/main/java/io/ebean/redisson/RedissonCache.java

@@ -0,0 +1,408 @@
+package io.ebean.redisson;
+
+import io.avaje.applog.AppLog;
+import io.ebean.BackgroundExecutor;
+import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheConfig;
+import io.ebean.cache.ServerCacheStatistics;
+import io.ebean.meta.MetricVisitor;
+import io.ebean.metric.CountMetric;
+import io.ebean.metric.MetricFactory;
+import io.ebean.metric.TimedMetric;
+import io.ebean.metric.TimedMetricStats;
+import io.ebean.redisson.encode.VersionGatedCodec;
+import io.ebeaninternal.server.cache.CachedBeanData;
+import io.netty.buffer.ByteBuf;
+import org.redisson.api.RMapCacheNative;
+import org.redisson.api.RScript;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.map.PutArgs;
+import org.redisson.client.codec.ByteArrayCodec;
+import org.redisson.client.codec.Codec;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.System.Logger.Level.ERROR;
+import static java.lang.System.Logger.Level.WARNING;
+
+/**
+ * Remote (shared) L2 cache region backed by a single Redis hash using
+ * <b>native per-field TTL</b> ({@link RMapCacheNative}). Requires <b>Redis 8.0+</b> / Valkey 9.0+:
+ * writes use {@code HSETEX} (Redis 8.0) and idle refresh uses {@code HEXPIRE}/{@code HGETEX} (Redis 7.4).
+ * <p>
+ * Compared with the scripted {@code RMapCache} this avoids the per-region timeout/idle/last-access
+ * sorted-sets and the client side eviction task: entries are expired by Redis itself. {@code clear()}
+ * remains a single {@code DEL} of the hash.
+ * <p>
+ * Feature handling:
+ * <ul>
+ *   <li><b>maxSecsToLive</b> - authoritative native per-field TTL (set on every writing).</li>
+ *   <li><b>maxIdleSecs</b> - in the shared remote we only slide the TTL on read when there is no
+ *       hard {@code maxSecsToLive} cap. When {@code maxSecsToLive > 0} it is the authoritative bound
+ *       (idle eviction is then handled by the in-heap near cache); this deliberately keeps remote
+ *       reads as plain {@code HGET} instead of turning every read into a Redis writing. When only
+ *       {@code maxIdleSecs} is set it becomes the TTL and is slid forward on each read.</li>
+ *   <li><b>maxSize</b> - native hashes are not bounded, so size is enforced by best-effort periodic
+ *       trim (see {@link #trimCache()}). Eviction order is approximate (Redis scan order) rather than LFU.</li>
+ * </ul>
+ */
+public class RedissonCache implements ServerCache {
+  private static final System.Logger log = AppLog.getLogger(RedissonCache.class);
+  private static final String CACHE_KEY_PREFIX = "EBEAN_CACHE";
+  private static final int TRIM_FREQUENCY_SECS = 60;
+
+  /**
+   * Batch compare-and-set: ARGV[1]=ttlMillis, ARGV[2]=marker bytes, then (field, value) pairs. The stored
+   * format is {@code [marker][8-byte big-endian version][bean]} ({@code VersionGatedCodec}).
+   */
+  private static final String VERSIONED_PUT_LUA =
+    "local ttl = tonumber(ARGV[1]); " +
+    "local marker = ARGV[2]; " +
+    "local mlen = string.len(marker); " +
+    "local i = 3; " +
+    "while i < #ARGV do " +
+    "  local field = ARGV[i]; local val = ARGV[i+1]; " +
+    "  local cur = redis.call('hget', KEYS[1], field); " +
+    "  local skip = false; " +
+    "  if cur ~= false and string.len(cur) >= mlen + 8 and string.sub(cur, 1, mlen) == marker then " +
+    "    if string.sub(cur, mlen + 1, mlen + 8) > string.sub(val, mlen + 1, mlen + 8) then skip = true; end; " +
+    "  end; " +
+    "  if not skip then " +
+    "    redis.call('hset', KEYS[1], field, val); " +
+    "    if ttl > 0 then redis.call('hpexpire', KEYS[1], ttl, 'FIELDS', 1, field); end; " +
+    "  end; " +
+    "  i = i + 2; " +
+    "end; " +
+    "return 1;";
+
+  private final int maxSize;
+  private final Duration writeTtl;
+  private final boolean slideIdle;
+  private final Duration idleTtl;
+  private final RMapCacheNative<String, Object> cacheMap;
+  private final Codec codec;
+  private final boolean versionGated;
+  private final RScript versionScript;
+  private final String mapName;
+  private final String cacheKey;
+  private final TimedMetric metricGet;
+  private final TimedMetric metricGetAll;
+  private final TimedMetric metricPut;
+  private final TimedMetric metricPutAll;
+  private final TimedMetric metricRemove;
+  private final TimedMetric metricRemoveAll;
+  private final TimedMetric metricClear;
+  private final CountMetric hitCount;
+  private final CountMetric missCount;
+
+  RedissonCache(RedissonClient redissonClient, ServerCacheConfig config, Codec codec, BackgroundExecutor executor, boolean versionGated) {
+    this.cacheKey = config.getCacheKey();
+    this.codec = codec;
+    this.versionGated = versionGated;
+    this.versionScript = versionGated ? redissonClient.getScript(ByteArrayCodec.INSTANCE) : null;
+
+    int maxSecsToLive = Math.max(config.getCacheOptions().getMaxSecsToLive(), 0);
+    int maxIdleSecs = Math.max(config.getCacheOptions().getMaxIdleSecs(), 0);
+    this.maxSize = config.getCacheOptions().getMaxSize();
+
+    if (maxSecsToLive > 0) {
+      this.writeTtl = Duration.ofSeconds(maxSecsToLive);
+      this.slideIdle = false;
+      this.idleTtl = null;
+    } else if (maxIdleSecs > 0) {
+      this.writeTtl = Duration.ofSeconds(maxIdleSecs);
+      this.slideIdle = true;
+      this.idleTtl = Duration.ofSeconds(maxIdleSecs);
+    } else {
+      this.writeTtl = null;
+      this.slideIdle = false;
+      this.idleTtl = null;
+    }
+
+    String namePrefix = "l2r." + config.getShortName();
+    MetricFactory factory = MetricFactory.get();
+    hitCount = factory.createCountMetric(namePrefix + ".hit");
+    missCount = factory.createCountMetric(namePrefix + ".miss");
+    metricGet = factory.createTimedMetric(namePrefix + ".get");
+    metricGetAll = factory.createTimedMetric(namePrefix + ".getMany");
+    metricPut = factory.createTimedMetric(namePrefix + ".put");
+    metricPutAll = factory.createTimedMetric(namePrefix + ".putMany");
+    metricRemove = factory.createTimedMetric(namePrefix + ".remove");
+    metricRemoveAll = factory.createTimedMetric(namePrefix + ".removeMany");
+    metricClear = factory.createTimedMetric(namePrefix + ".clear");
+    this.mapName = CACHE_KEY_PREFIX + ":" + cacheKey;
+    cacheMap = redissonClient.getMapCacheNative(mapName, codec);
+
+    if (maxSize > 0 && executor != null) {
+      executor.scheduleWithFixedDelay(this::trimCache, TRIM_FREQUENCY_SECS, TRIM_FREQUENCY_SECS, TimeUnit.SECONDS);
+    }
+  }
+
+  @Override
+  public void visit(MetricVisitor visitor) {
+    hitCount.visit(visitor);
+    missCount.visit(visitor);
+    metricGet.visit(visitor);
+    metricGetAll.visit(visitor);
+    metricPut.visit(visitor);
+    metricPutAll.visit(visitor);
+    metricRemove.visit(visitor);
+    metricRemoveAll.visit(visitor);
+    metricClear.visit(visitor);
+  }
+
+  private void errorOnRead(Exception e) {
+    log.log(ERROR, "Error reading redis cache [" + mapName + "] - treating as miss", e);
+  }
+
+  private void errorOnWrite(Exception e) {
+    log.log(ERROR, "Error writing redis cache [" + mapName + "] - treating as miss", e);
+  }
+
+  @Override
+  public Map<Object, Object> getAll(Set<Object> keys) {
+    try {
+      if (keys.isEmpty()) {
+        return Collections.emptyMap();
+      }
+      long start = System.nanoTime();
+      Map<String, Object> strToOrigKey = new LinkedHashMap<>();
+      for (Object key : keys) {
+        strToOrigKey.put(key.toString(), key);
+      }
+      Map<Object, Object> map = new LinkedHashMap<>();
+      Map<String, Object> values = cacheMap.getAll(strToOrigKey.keySet());
+      for (Map.Entry<String, Object> strEntry : strToOrigKey.entrySet()) {
+        Object value = values.get(strEntry.getKey());
+        if (value != null) {
+          map.put(strEntry.getValue(), value);
+        }
+      }
+      if (slideIdle && !values.isEmpty()) {
+        slideIdleAsync(values.keySet());
+      }
+      int hits = map.size();
+      int miss = keys.size() - hits;
+
+      if (hits > 0) {
+        hitCount.add(hits);
+      }
+      if (miss > 0) {
+        missCount.add(miss);
+      }
+      metricGetAll.addSinceNanos(start);
+      return map;
+    } catch (Exception e) {
+      errorOnRead(e);
+      return Collections.emptyMap();
+    }
+  }
+
+  @Override
+  public Object get(Object id) {
+    long start = System.nanoTime();
+    try {
+      String key = id.toString();
+      Object val = cacheMap.get(key);
+      if (val != null) {
+        hitCount.increment();
+        if (slideIdle) {
+          slideIdleAsync(Collections.singleton(key));
+        }
+      } else {
+        missCount.increment();
+      }
+      metricGet.addSinceNanos(start);
+      return val;
+    } catch (Exception e) {
+      errorOnRead(e);
+      return null;
+    }
+  }
+
+  private void slideIdleAsync(Set<String> keys) {
+    try {
+      if (keys.size() == 1) {
+        cacheMap.expireEntryAsync(keys.iterator().next(), idleTtl)
+          .whenComplete((r, e) -> logSlideError(e));
+      } else {
+        cacheMap.expireEntriesAsync(keys, idleTtl)
+          .whenComplete((r, e) -> logSlideError(e));
+      }
+    } catch (Exception e) {
+      logSlideError(e);
+    }
+  }
+
+  private void logSlideError(Throwable e) {
+    if (e != null) {
+      log.log(WARNING, "Error sliding idle TTL on redis cache [" + mapName + "]", e);
+    }
+  }
+
+  @Override
+  public void put(Object id, Object value) {
+    long start = System.nanoTime();
+    try {
+      String key = id.toString();
+      if (versionGated && value instanceof CachedBeanData) {
+        versionedPut(Map.of(key, value));
+      } else {
+        writePut(key, value);
+      }
+      metricPut.addSinceNanos(start);
+    } catch (Exception e) {
+      errorOnWrite(e);
+    }
+  }
+
+  private void writePut(String key, Object value) {
+    if (writeTtl == null) {
+      cacheMap.fastPut(key, value);
+    } else {
+      cacheMap.fastPut(key, value, writeTtl);
+    }
+  }
+
+  private void writePutAll(Map<String, Object> map) {
+    if (writeTtl == null) {
+      cacheMap.putAll(map);
+    } else {
+      cacheMap.putAll(PutArgs.entries(map).timeToLive(writeTtl));
+    }
+  }
+
+  /**
+   * Version-gated put: never overwrites a strictly newer cached version
+   */
+  private void versionedPut(Map<String, Object> data) {
+    long ttlMillis = (writeTtl == null) ? 0L : writeTtl.toMillis();
+    List<Object> argv = new ArrayList<>(2 + data.size() * 2);
+    argv.add(String.valueOf(ttlMillis).getBytes(StandardCharsets.UTF_8));
+    argv.add(VersionGatedCodec.MARKER.clone());
+    for (Map.Entry<String, Object> entry : data.entrySet()) {
+      argv.add(entry.getKey().getBytes(StandardCharsets.UTF_8));
+      argv.add(encodeValue(entry.getValue()));
+    }
+    versionScript.eval(RScript.Mode.READ_WRITE, VERSIONED_PUT_LUA, RScript.ReturnType.BOOLEAN,
+      Collections.singletonList(mapName), argv.toArray());
+  }
+
+  private byte[] encodeValue(Object value) {
+    try {
+      ByteBuf buf = codec.getValueEncoder().encode(value);
+      try {
+        byte[] bytes = new byte[buf.readableBytes()];
+        buf.getBytes(buf.readerIndex(), bytes);
+        return bytes;
+      } finally {
+        buf.release();
+      }
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public void putAll(Map<Object, Object> keyValues) {
+    long start = System.nanoTime();
+    try {
+      Map<String, Object> map = new LinkedHashMap<>();
+      for (Map.Entry<Object, Object> entry : keyValues.entrySet()) {
+        map.put(entry.getKey().toString(), entry.getValue());
+      }
+      if (versionGated && !keyValues.isEmpty() && keyValues.entrySet().iterator().next().getValue() instanceof CachedBeanData) {
+        versionedPut(map);
+      } else {
+        writePutAll(map);
+      }
+      metricPutAll.addSinceNanos(start);
+    } catch (Exception e) {
+      errorOnWrite(e);
+    }
+  }
+
+  @Override
+  public void remove(Object id) {
+    long start = System.nanoTime();
+    try {
+      cacheMap.fastRemove(id.toString());
+      metricRemove.addSinceNanos(start);
+    } catch (Exception e) {
+      errorOnWrite(e);
+    }
+  }
+
+  @Override
+  public void removeAll(Set<Object> keys) {
+    long start = System.nanoTime();
+    try {
+      var keysArray = keys.stream().map(Object::toString).toArray(String[]::new);
+      cacheMap.fastRemove(keysArray);
+      metricRemoveAll.addSinceNanos(start);
+    } catch (Exception e) {
+      errorOnWrite(e);
+    }
+  }
+
+  @Override
+  public void clear() {
+    long start = System.nanoTime();
+    try {
+      cacheMap.clear();
+      metricClear.addSinceNanos(start);
+    } catch (Exception e) {
+      errorOnWrite(e);
+    }
+  }
+
+  void trimCache() {
+    try {
+      int size = cacheMap.size();
+      int toRemove = size - maxSize;
+      if (toRemove <= 0) {
+        return;
+      }
+      List<String> victims = new ArrayList<>(Math.min(toRemove, 1024));
+      for (String key : cacheMap.keySet()) {
+        victims.add(key);
+        if (victims.size() >= toRemove) {
+          break;
+        }
+      }
+      if (!victims.isEmpty()) {
+        cacheMap.fastRemove(victims.toArray(new String[0]));
+      }
+    } catch (Exception e) {
+      log.log(WARNING, "Error trimming redis cache [" + mapName + "] to maxSize " + maxSize, e);
+    }
+  }
+
+  public long getHitCount() {
+    return hitCount.get(false);
+  }
+
+  public long getMissCount() {
+    return missCount.get(false);
+  }
+
+  @Override
+  public ServerCacheStatistics statistics(boolean reset) {
+    ServerCacheStatistics cacheStats = new ServerCacheStatistics();
+    cacheStats.setCacheName(cacheKey);
+    cacheStats.setHitCount(hitCount.get(reset));
+    cacheStats.setMissCount(missCount.get(reset));
+    cacheStats.setPutCount(count(metricPut.collect(reset)));
+    cacheStats.setRemoveCount(count(metricRemove.collect(reset)));
+    cacheStats.setClearCount(count(metricClear.collect(reset)));
+    return cacheStats;
+  }
+
+  private long count(TimedMetricStats stats) {
+    return stats == null ? 0 : stats.count();
+  }
+}

ebean-redisson/src/main/java/io/ebean/redisson/RedissonCacheFactory.java

@@ -0,0 +1,462 @@
+package io.ebean.redisson;
+
+
+import io.avaje.applog.AppLog;
+import io.ebean.BackgroundExecutor;
+import io.ebean.DatabaseBuilder;
+import io.ebean.cache.*;
+import io.ebean.meta.MetricVisitor;
+import io.ebean.metric.MetricFactory;
+import io.ebean.metric.TimedMetric;
+import io.ebean.redisson.dto.*;
+import io.ebean.redisson.encode.CachedBeanDataCodec;
+import io.ebean.redisson.encode.CachedManyIdsCodec;
+import io.ebean.redisson.encode.SerializableCodec;
+import io.ebean.redisson.encode.VersionGatedCodec;
+import io.ebean.redisson.near.NearCacheInvalidate;
+import io.ebean.redisson.near.NearCacheNotify;
+import io.ebeaninternal.server.cache.DefaultServerCache;
+import io.ebeaninternal.server.cache.DefaultServerCacheConfig;
+import io.ebeaninternal.server.cache.DefaultServerQueryCache;
+import org.redisson.Redisson;
+import org.redisson.api.RReliableTopic;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+
+import java.io.*;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static java.lang.System.Logger.Level.*;
+
+public class RedissonCacheFactory implements ServerCacheFactory {
+
+    private static final System.Logger log = AppLog.getLogger(RedissonCacheFactory.class);
+
+    /**
+     * Channel for standard L2 cache messages.
+     */
+    private static final String CHANNEL_L2 = "ebean.l2cache";
+    /**
+     * Channel specifically for near cache invalidation messages.
+     */
+    private static final String CHANNEL_NEAR = "ebean.l2near";
+
+    private final ConcurrentHashMap<String, RQueryCache> queryCaches = new ConcurrentHashMap<>();
+    private final Map<String, NearCacheInvalidate> nearCacheMap = new ConcurrentHashMap<>();
+    private final SerializableCodec serializableCodec = new SerializableCodec();
+    private final CachedBeanDataCodec cachedBeanDataCodec = new CachedBeanDataCodec();
+    private final CachedManyIdsCodec cachedManyIdsCodec = new CachedManyIdsCodec();
+    private final BackgroundExecutor executor;
+    private final RedissonClient redissonClient;
+    private final NearCacheNotify nearCacheNotify;
+    private final TimedMetric metricOutNearCache;
+    private final TimedMetric metricOutTableMod;
+    private final TimedMetric metricOutQueryCache;
+    private final TimedMetric metricInNearCache;
+    private final TimedMetric metricInTableMod;
+    private final TimedMetric metricInQueryCache;
+    private final String serverId = ModId.id();
+    private final ReentrantLock lock = new ReentrantLock();
+    private final RReliableTopic topicL2;
+    private final RReliableTopic topicNear;
+    private ServerCacheNotify listener;
+
+    RedissonCacheFactory(DatabaseBuilder.Settings config, BackgroundExecutor executor) {
+        this.executor = executor;
+        this.nearCacheNotify = new DNearCacheNotify();
+        MetricFactory factory = MetricFactory.get();
+        this.metricOutTableMod = factory.createTimedMetric("l2a.outTableMod");
+        this.metricOutQueryCache = factory.createTimedMetric("l2a.outQueryCache");
+        this.metricOutNearCache = factory.createTimedMetric("l2a.outNearKeys");
+        this.metricInTableMod = factory.createTimedMetric("l2a.inTableMod");
+        this.metricInQueryCache = factory.createTimedMetric("l2a.inQueryCache");
+        this.metricInNearCache = factory.createTimedMetric("l2a.inNearKeys");
+        this.redissonClient = getRedissonClient(config);
+        this.topicL2 = redissonClient.getReliableTopic(CHANNEL_L2);
+        this.topicNear = redissonClient.getReliableTopic(CHANNEL_NEAR);
+        subscribeToMessages();
+    }
+
+    private RedissonClient getRedissonClient(DatabaseBuilder.Settings config) {
+        RedissonClient existingClient = config.getServiceObject(RedissonClient.class);
+        if (existingClient != null) {
+            return existingClient;
+        }
+
+        Config redisConfig = config.getServiceObject(Config.class);
+        if (redisConfig != null) {
+            return Redisson.create(redisConfig);
+        }
+
+        Config loadedConfig = null;
+        try {
+            ClassLoader cl = Thread.currentThread().getContextClassLoader();
+            InputStream is = cl.getResourceAsStream("redisson-config.yaml");
+
+            if (is != null) {
+                loadedConfig = Config.fromYAML(is);
+                log.log(INFO, "Loaded Redisson config from classpath: redisson-config.yaml");
+            } else {
+                log.log(WARNING, "redisson-config.yaml not found in classpath. Falling back to default config.");
+            }
+        } catch (IllegalArgumentException e) {
+            log.log(WARNING, "Failed to load redisson-config.yaml from classpath. Falling back to default config.", e);
+        }
+
+        if (loadedConfig == null) {
+            loadedConfig = new Config();
+            loadedConfig.useSingleServer().setAddress("redis://localhost:6379");
+            log.log(WARNING, "Using default Redisson config: redis://localhost:6379");
+        }
+
+        return Redisson.create(loadedConfig);
+    }
+
+    @Override
+    public void visit(MetricVisitor visitor) {
+        metricOutQueryCache.visit(visitor);
+        metricOutTableMod.visit(visitor);
+        metricOutNearCache.visit(visitor);
+        metricInTableMod.visit(visitor);
+        metricInQueryCache.visit(visitor);
+        metricInNearCache.visit(visitor);
+    }
+
+    @Override
+    public ServerCache createCache(ServerCacheConfig config) {
+        if (config.isQueryCache()) {
+            return createQueryCache(config);
+        }
+        return createNormalCache(config);
+    }
+
+    private ServerCache createNormalCache(ServerCacheConfig config) {
+        RedissonCache redissonCache = createRedisCache(config);
+        boolean nearCache = config.getCacheOptions().isNearCache();
+        if (!nearCache) {
+            return config.tenantAware(redissonCache);
+        }
+
+        String cacheKey = config.getCacheKey();
+        DefaultServerCache near = new DefaultServerCache(new DefaultServerCacheConfig(config));
+        near.periodicTrim(executor);
+        DuelCache duelCache = new DuelCache(near, redissonCache, cacheKey, nearCacheNotify);
+        nearCacheMap.put(cacheKey, duelCache);
+        return config.tenantAware(duelCache);
+    }
+
+    private RedissonCache createRedisCache(ServerCacheConfig config) {
+        switch (config.getType()) {
+            case NATURAL_KEY:
+                return new RedissonCache(redissonClient, config, serializableCodec, executor, false);
+            case BEAN: {
+                VersionGatedCodec codec = new VersionGatedCodec(cachedBeanDataCodec);
+                return new RedissonCache(redissonClient, config, codec, executor, true);
+            }
+            case COLLECTION_IDS:
+                return new RedissonCache(redissonClient, config, cachedManyIdsCodec, executor, false);
+            default:
+                throw new IllegalArgumentException("Unexpected cache type? " + config.getType());
+        }
+    }
+
+    private ServerCache createQueryCache(ServerCacheConfig config) {
+        lock.lock();
+        try {
+            RQueryCache cache = queryCaches.get(config.getCacheKey());
+            if (cache == null) {
+                log.log(DEBUG, config.getCacheKey());
+                cache = new RQueryCache(new DefaultServerCacheConfig(config));
+                cache.periodicTrim(executor);
+                queryCaches.put(config.getCacheKey(), cache);
+            }
+            return config.tenantAware(cache);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public ServerCacheNotify createCacheNotify(ServerCacheNotify listener) {
+        this.listener = listener;
+        return new RServerCacheNotify();
+    }
+
+    private void sendQueryCacheInvalidation(String name) {
+        long nanos = System.nanoTime();
+        try {
+            L2QueryInvalidMessage message = new L2QueryInvalidMessage();
+            message.setServerId(serverId);
+            message.setKey(name);
+
+            topicL2.publish(message);
+        } finally {
+            metricOutQueryCache.addSinceNanos(nanos);
+        }
+    }
+
+    private void sendTableMod(Set<String> dependentTables) {
+        long nanos = System.nanoTime();
+        try {
+
+            L2TableModMessage message = new L2TableModMessage();
+            message.setTables(dependentTables);
+            message.setServerId(serverId);
+
+            topicL2.publish(message);
+        } finally {
+            metricOutTableMod.addSinceNanos(nanos);
+        }
+    }
+
+    /**
+     * Clear the query cache if we have it.
+     */
+    private void queryCacheInvalidate(L2QueryInvalidMessage message) {
+        if (serverId.equals(message.getServerId())) {
+            // ignore this message as we are the server that sent it
+            return;
+        }
+
+        long nanos = System.nanoTime();
+        try {
+            RQueryCache queryCache = queryCaches.get(message.getKey());
+            if (queryCache != null) {
+                queryCache.invalidate();
+            }
+        } finally {
+            metricInQueryCache.addSinceNanos(nanos);
+        }
+    }
+
+    /**
+     * Process a remote-dependent table modify event.
+     */
+    private void processTableNotify(L2TableModMessage message) {
+        if (serverId.equals(message.getServerId())) {
+            // ignore this message as we are the server that sent it
+            return;
+        }
+        if (listener == null) {
+            log.log(DEBUG, "Ignoring tableMod, listener not registered yet");
+            return;
+        }
+        long nanos = System.nanoTime();
+        try {
+            listener.notify(new ServerCacheNotification(message.getTables()));
+        } finally {
+            metricInTableMod.addSinceNanos(nanos);
+        }
+    }
+
+    /**
+     * Invalidate key for a local near cache.
+     */
+    private void nearCacheInvalidateKey(NearCacheInvalidateKeyMessage message) {
+        String sourceServerId = message.getServerId();
+        if (sourceServerId.equals(serverId)) {
+            // ignore this message as we are the server that sent it
+            return;
+        }
+
+        String cacheKey = message.getCacheKey();
+        long nanos = System.nanoTime();
+        try (ObjectInputStream oi = new ObjectInputStream(new ByteArrayInputStream(message.getKey()))) {
+            Object key = oi.readObject();
+            NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
+            if (invalidate == null) {
+                warnNearCacheNotFound(cacheKey);
+            } else {
+                invalidate.invalidateKey(key);
+            }
+        } catch (IOException | ClassNotFoundException e) {
+            log.log(ERROR, "failed to decode near cache message [" + message + "] for cache:" + cacheKey, e);
+            if (cacheKey != null) {
+                nearCacheInvalidateClear(cacheKey);
+            }
+        } finally {
+            metricInNearCache.addSinceNanos(nanos);
+        }
+    }
+
+    /**
+     * Invalidate keys for a local near cache.
+     */
+    private void nearCacheInvalidateKeys(NearCacheInvalidateKeysMessage message) {
+        String sourceServerId = message.getServerId();
+        if (sourceServerId.equals(serverId)) {
+            // ignore this message as we are the server that sent it
+            return;
+        }
+
+        String cacheKey = message.getCacheKey();
+        long nanos = System.nanoTime();
+        try (ObjectInputStream oi = new ObjectInputStream(new ByteArrayInputStream(message.getKeys()))) {
+            int total = oi.readInt();
+            Set<Object> keys = new LinkedHashSet<>();
+            for (int i = 0; i < total; i++) {
+                keys.add(oi.readObject());
+            }
+            NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
+            if (invalidate == null) {
+                warnNearCacheNotFound(cacheKey);
+            } else {
+                invalidate.invalidateKeys(keys);
+            }
+        } catch (IOException | ClassNotFoundException e) {
+            log.log(ERROR, "failed to decode near cache message [" + message + "] for cache:" + cacheKey, e);
+            if (cacheKey != null) {
+                nearCacheInvalidateClear(cacheKey);
+            }
+        } finally {
+            metricInNearCache.addSinceNanos(nanos);
+        }
+    }
+
+    /**
+     * Invalidate clear for a local near cache.
+     */
+    private void nearCacheInvalidateClear(String cacheKey) {
+        NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
+        if (invalidate == null) {
+            warnNearCacheNotFound(cacheKey);
+        } else {
+            invalidate.invalidateClear();
+        }
+    }
+
+    private void nearCacheInvalidateClear(NearCacheClearMessage message) {
+        String sourceServerId = message.getServerId();
+        if (sourceServerId.equals(serverId)) {
+            // ignore this message as we are the server that sent it
+            return;
+        }
+
+        String cacheKey = message.getCacheKey();
+        long nanos = System.nanoTime();
+        try {
+            NearCacheInvalidate invalidate = nearCacheMap.get(cacheKey);
+            if (invalidate == null) {
+                warnNearCacheNotFound(cacheKey);
+            } else {
+                invalidate.invalidateClear();
+            }
+        } finally {
+            metricInNearCache.addSinceNanos(nanos);
+        }
+    }
+
+    private void warnNearCacheNotFound(String cacheKey) {
+        log.log(WARNING, "No near cache found for cacheKey [" + cacheKey + "] yet - probably on startup");
+    }
+
+    private void subscribeToMessages() {
+        topicL2.addListener(L2QueryInvalidMessage.class, (channel, message) -> queryCacheInvalidate(message));
+        topicL2.addListener(L2TableModMessage.class, (channel, message) -> processTableNotify(message));
+        topicNear.addListener(NearCacheClearMessage.class, (channel, message) -> nearCacheInvalidateClear(message));
+        topicNear.addListener(NearCacheInvalidateKeyMessage.class, (channel, message) -> nearCacheInvalidateKey(message));
+        topicNear.addListener(NearCacheInvalidateKeysMessage.class, (channel, message) -> nearCacheInvalidateKeys(message));
+    }
+
+    /**
+     * Query cache implementation using a Redis channel for message notifications.
+     */
+    private class RQueryCache extends DefaultServerQueryCache {
+
+        RQueryCache(DefaultServerCacheConfig config) {
+            super(config);
+        }
+
+        @Override
+        public void clear() {
+            super.clear();
+            sendQueryCacheInvalidation(name);
+        }
+
+        /**
+         * Process the invalidation message coming from the cluster.
+         */
+        private void invalidate() {
+            super.clear();
+        }
+    }
+
+    /**
+     * Publish table modifications using a Redis channel (to other cluster members)
+     */
+    private class RServerCacheNotify implements ServerCacheNotify {
+
+        @Override
+        public void notify(ServerCacheNotification tableModifications) {
+            Set<String> dependentTables = tableModifications.getDependentTables();
+            if (dependentTables != null && !dependentTables.isEmpty()) {
+                sendTableMod(dependentTables);
+            }
+        }
+    }
+
+    private class DNearCacheNotify implements NearCacheNotify {
+
+        @Override
+        public void invalidateKeys(String cacheKey, Set<Object> keySet) {
+            try {
+                ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
+                ObjectOutputStream os = new ObjectOutputStream(ba);
+                os.writeInt(keySet.size());
+                for (Object key : keySet) {
+                    os.writeObject(key);
+                }
+                os.flush();
+                os.close();
+
+                NearCacheInvalidateKeysMessage message = new NearCacheInvalidateKeysMessage();
+                message.setServerId(serverId);
+                message.setCacheKey(cacheKey);
+                message.setKeys(ba.toByteArray());
+                sendMessage(message);
+            } catch (IOException e) {
+                log.log(ERROR, "failed to transmit invalidateKeys() message", e);
+            }
+        }
+
+        @Override
+        public void invalidateKey(String cacheKey, Object id) {
+            try {
+                ByteArrayOutputStream ba = new ByteArrayOutputStream(100);
+                ObjectOutputStream os = new ObjectOutputStream(ba);
+                os.writeObject(id);
+                os.flush();
+                os.close();
+
+                NearCacheInvalidateKeyMessage message = new NearCacheInvalidateKeyMessage();
+                message.setServerId(serverId);
+                message.setCacheKey(cacheKey);
+                message.setKey(ba.toByteArray());
+                sendMessage(message);
+            } catch (IOException e) {
+                log.log(ERROR, "failed to transmit invalidateKeys() message", e);
+            }
+        }
+
+        @Override
+        public void invalidateClear(String cacheKey) {
+            NearCacheClearMessage message = new NearCacheClearMessage();
+            message.setServerId(serverId);
+            message.setCacheKey(cacheKey);
+            sendMessage(message);
+        }
+
+        private void sendMessage(NearMessage message) {
+            long nanos = System.nanoTime();
+            try {
+                topicNear.publish(message);
+            } finally {
+                metricOutNearCache.addSinceNanos(nanos);
+            }
+        }
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/RedissonCachePlugin.java

@@ -0,0 +1,13 @@
+package io.ebean.redisson;
+
+import io.ebean.BackgroundExecutor;
+import io.ebean.DatabaseBuilder;
+import io.ebean.cache.ServerCacheFactory;
+import io.ebean.cache.ServerCachePlugin;
+
+public class RedissonCachePlugin implements ServerCachePlugin {
+    @Override
+    public ServerCacheFactory create(DatabaseBuilder config, BackgroundExecutor executor) {
+        return new RedissonCacheFactory(config.settings(), executor);
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/L2Message.java

@@ -0,0 +1,4 @@
+package io.ebean.redisson.dto;
+
+public interface L2Message {
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/L2QueryInvalidMessage.java

@@ -0,0 +1,44 @@
+package io.ebean.redisson.dto;
+
+import java.util.Objects;
+
+public class L2QueryInvalidMessage implements L2Message {
+    private String serverId;
+    private String key;
+
+    @Override
+    public String toString() {
+        return "L2QueryInvalidMessage{" +
+               "serverId='" + serverId + '\'' +
+               ", key='" + key + '\'' +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof L2QueryInvalidMessage)) return false;
+        L2QueryInvalidMessage that = (L2QueryInvalidMessage) o;
+        return Objects.equals(getServerId(), that.getServerId()) && Objects.equals(getKey(), that.getKey());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getServerId(), getKey());
+    }
+
+    public String getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(String serverId) {
+        this.serverId = serverId;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/L2TableModMessage.java

@@ -0,0 +1,45 @@
+package io.ebean.redisson.dto;
+
+import java.util.Objects;
+import java.util.Set;
+
+public class L2TableModMessage implements L2Message {
+    private String serverId;
+    private Set<String> tables;
+
+    @Override
+    public String toString() {
+        return "L2TableModMessage{" +
+               "serverId='" + serverId + '\'' +
+               ", tables=" + tables +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof L2TableModMessage)) return false;
+        L2TableModMessage that = (L2TableModMessage) o;
+        return Objects.equals(getServerId(), that.getServerId()) && Objects.equals(getTables(), that.getTables());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getServerId(), getTables());
+    }
+
+    public String getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(String serverId) {
+        this.serverId = serverId;
+    }
+
+    public Set<String> getTables() {
+        return tables;
+    }
+
+    public void setTables(Set<String> tables) {
+        this.tables = tables;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/NearCacheClearMessage.java

@@ -0,0 +1,44 @@
+package io.ebean.redisson.dto;
+
+import java.util.Objects;
+
+public class NearCacheClearMessage implements NearMessage {
+    private String serverId;
+    private String cacheKey;
+
+    @Override
+    public String toString() {
+        return "NearCacheClearMessage{" +
+               "serverId='" + serverId + '\'' +
+               ", cacheKey='" + cacheKey + '\'' +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof NearCacheClearMessage)) return false;
+        NearCacheClearMessage that = (NearCacheClearMessage) o;
+        return Objects.equals(getServerId(), that.getServerId()) && Objects.equals(getCacheKey(), that.getCacheKey());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getServerId(), getCacheKey());
+    }
+
+    public String getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(String serverId) {
+        this.serverId = serverId;
+    }
+
+    public String getCacheKey() {
+        return cacheKey;
+    }
+
+    public void setCacheKey(String cacheKey) {
+        this.cacheKey = cacheKey;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/NearCacheInvalidateKeyMessage.java

@@ -0,0 +1,55 @@
+package io.ebean.redisson.dto;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+public class NearCacheInvalidateKeyMessage implements NearMessage {
+    private String serverId;
+    private String cacheKey;
+    private byte[] key;
+
+    @Override
+    public String toString() {
+        return "NearCacheInvalidateKeyMessage{" +
+               "serverId='" + serverId + '\'' +
+               ", cacheKey='" + cacheKey + '\'' +
+               ", key=" + Arrays.toString(key) +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof NearCacheInvalidateKeyMessage)) return false;
+        NearCacheInvalidateKeyMessage that = (NearCacheInvalidateKeyMessage) o;
+        return Objects.equals(getServerId(), that.getServerId()) && Objects.equals(getCacheKey(), that.getCacheKey()) && Objects.deepEquals(getKey(), that.getKey());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getServerId(), getCacheKey(), Arrays.hashCode(getKey()));
+    }
+
+    public String getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(String serverId) {
+        this.serverId = serverId;
+    }
+
+    public String getCacheKey() {
+        return cacheKey;
+    }
+
+    public void setCacheKey(String cacheKey) {
+        this.cacheKey = cacheKey;
+    }
+
+    public byte[] getKey() {
+        return key;
+    }
+
+    public void setKey(byte[] key) {
+        this.key = key;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/NearCacheInvalidateKeysMessage.java

@@ -0,0 +1,55 @@
+package io.ebean.redisson.dto;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+public class NearCacheInvalidateKeysMessage implements NearMessage {
+    private String serverId;
+    private String cacheKey;
+    private byte[] keys;
+
+    @Override
+    public String toString() {
+        return "NearCacheInvalidateKeysMessage{" +
+               "serverId='" + serverId + '\'' +
+               ", cacheKey='" + cacheKey + '\'' +
+               ", keys=" + Arrays.toString(keys) +
+               '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof NearCacheInvalidateKeysMessage)) return false;
+        NearCacheInvalidateKeysMessage that = (NearCacheInvalidateKeysMessage) o;
+        return Objects.equals(getServerId(), that.getServerId()) && Objects.equals(getCacheKey(), that.getCacheKey()) && Objects.deepEquals(getKeys(), that.getKeys());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getServerId(), getCacheKey(), Arrays.hashCode(getKeys()));
+    }
+
+    public String getServerId() {
+        return serverId;
+    }
+
+    public void setServerId(String serverId) {
+        this.serverId = serverId;
+    }
+
+    public String getCacheKey() {
+        return cacheKey;
+    }
+
+    public void setCacheKey(String cacheKey) {
+        this.cacheKey = cacheKey;
+    }
+
+    public byte[] getKeys() {
+        return keys;
+    }
+
+    public void setKeys(byte[] keys) {
+        this.keys = keys;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/dto/NearMessage.java

@@ -0,0 +1,4 @@
+package io.ebean.redisson.dto;
+
+public interface NearMessage {
+}

ebean-redisson/src/main/java/io/ebean/redisson/encode/CacheCodec.java

@@ -0,0 +1,40 @@
+package io.ebean.redisson.encode;
+
+import io.ebean.cache.TenantAwareKey;
+import io.netty.buffer.Unpooled;
+import org.redisson.client.codec.BaseCodec;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.nio.charset.StandardCharsets;
+
+public abstract class CacheCodec extends BaseCodec {
+
+    @Override
+    public Encoder getMapKeyEncoder() {
+        return in -> {
+            try {
+                if (!(in instanceof String) && !(in instanceof TenantAwareKey.CacheKey)) {
+                    throw new IllegalStateException("Expecting String keys but got type: " + in.getClass());
+                }
+                byte[] bytes = in.toString().getBytes(StandardCharsets.UTF_8);
+                return Unpooled.wrappedBuffer(bytes);
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to encode cache key", e);
+            }
+        };
+    }
+
+    @Override
+    public Decoder<Object> getMapKeyDecoder() {
+        return (buf, state) -> {
+            try {
+                byte[] bytes = new byte[buf.readableBytes()];
+                buf.readBytes(bytes);
+                return new String(bytes, StandardCharsets.UTF_8);
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to decode cache key", e);
+            }
+        };
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/encode/CachedBeanDataCodec.java

@@ -0,0 +1,52 @@
+package io.ebean.redisson.encode;
+
+import io.ebeaninternal.server.cache.CachedBeanData;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.buffer.ByteBufOutputStream;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+public class CachedBeanDataCodec extends CacheCodec {
+
+    private final Encoder encoder = in -> {
+        ByteBuf out = ByteBufAllocator.DEFAULT.buffer();
+        try (ByteBufOutputStream os = new ByteBufOutputStream(out);
+             ObjectOutputStream oos = new ObjectOutputStream(os)) {
+            ((CachedBeanData) in).writeExternal(oos);
+            return os.buffer();
+        } catch (IOException e) {
+            out.release();
+            throw e;
+        } catch (Exception e) {
+            out.release();
+            throw new IOException(e);
+        }
+    };
+    private final Decoder<Object> decoder = (in, state) -> {
+
+        try (ByteBufInputStream is = new ByteBufInputStream(in);
+             ObjectInputStream ois = new ObjectInputStream(is)) {
+            CachedBeanData data = new CachedBeanData();
+            data.readExternal(ois);
+            return data;
+        } catch (IOException | ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    };
+
+    @Override
+    public Encoder getValueEncoder() {
+        return encoder;
+    }
+
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return decoder;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/encode/CachedManyIdsCodec.java

@@ -0,0 +1,52 @@
+package io.ebean.redisson.encode;
+
+import io.ebeaninternal.server.cache.CachedManyIds;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.buffer.ByteBufOutputStream;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+public class CachedManyIdsCodec extends CacheCodec {
+
+    private final Encoder encoder = in -> {
+        ByteBuf out = ByteBufAllocator.DEFAULT.buffer();
+        try (ByteBufOutputStream os = new ByteBufOutputStream(out);
+             ObjectOutputStream oos = new ObjectOutputStream(os)) {
+            ((CachedManyIds) in).writeExternal(oos);
+            return os.buffer();
+        } catch (IOException e) {
+            out.release();
+            throw e;
+        } catch (Exception e) {
+            out.release();
+            throw new IOException(e);
+        }
+    };
+    private final Decoder<Object> decoder = (in, state) -> {
+
+        try (ByteBufInputStream is = new ByteBufInputStream(in);
+             ObjectInputStream ois = new ObjectInputStream(is)) {
+            CachedManyIds data = new CachedManyIds();
+            data.readExternal(ois);
+            return data;
+        } catch (IOException | ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    };
+
+    @Override
+    public Encoder getValueEncoder() {
+        return encoder;
+    }
+
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return decoder;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/encode/SerializableCodec.java

@@ -0,0 +1,50 @@
+package io.ebean.redisson.encode;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.buffer.ByteBufOutputStream;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+public class SerializableCodec extends CacheCodec {
+
+    private final Encoder encoder = in -> {
+        ByteBuf out = ByteBufAllocator.DEFAULT.buffer();
+        try (ByteBufOutputStream os = new ByteBufOutputStream(out);
+             ObjectOutputStream oos = new ObjectOutputStream(os)) {
+            oos.writeObject(in);
+            oos.flush();
+            return os.buffer();
+        } catch (IOException e) {
+            out.release();
+            throw e;
+        } catch (Exception e) {
+            out.release();
+            throw new IOException(e);
+        }
+    };
+    private final Decoder<Object> decoder = (in, state) -> {
+
+        try (ByteBufInputStream is = new ByteBufInputStream(in);
+             ObjectInputStream ois = new ObjectInputStream(is)) {
+            return ois.readObject();
+        } catch (IOException | ClassNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    };
+
+    @Override
+    public Encoder getValueEncoder() {
+        return encoder;
+    }
+
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return decoder;
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/encode/VersionGatedCodec.java

@@ -0,0 +1,82 @@
+package io.ebean.redisson.encode;
+
+import io.ebeaninternal.server.cache.CachedBeanData;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import org.redisson.client.codec.BaseCodec;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+/**
+ * Wraps a bean codec and stores the entity {@code @Version} as a fixed big-endian prefix in front of the
+ * encoded value, behind a 2-byte magic {@link #MARKER}.
+ */
+public class VersionGatedCodec extends BaseCodec {
+    public static final byte[] MARKER = {(byte) 0xEB, (byte) 0x01};
+    public static final int VERSION_BYTES = 8;
+    public static final int PREFIX_BYTES = 2 + VERSION_BYTES;
+
+    private final CacheCodec delegate;
+    private final Encoder valueEncoder;
+    private final Decoder<Object> valueDecoder;
+
+    public VersionGatedCodec(CacheCodec delegate) {
+        this.delegate = delegate;
+        Encoder delegateEncoder = delegate.getValueEncoder();
+        Decoder<Object> delegateDecoder = delegate.getValueDecoder();
+
+        this.valueEncoder = in -> {
+            long version = (in instanceof CachedBeanData) ? ((CachedBeanData) in).getVersion() : 0L;
+            ByteBuf inner = delegateEncoder.encode(in);
+            try {
+                ByteBuf out = ByteBufAllocator.DEFAULT.buffer(PREFIX_BYTES + inner.readableBytes());
+                out.writeBytes(MARKER);
+                out.writeLong(version);
+                out.writeBytes(inner);
+                return out;
+            } finally {
+                inner.release();
+            }
+        };
+
+        this.valueDecoder = (buf, state) -> {
+            if (hasMarker(buf)) {
+                buf.skipBytes(PREFIX_BYTES);
+            }
+            return delegateDecoder.decode(buf, state);
+        };
+    }
+
+    private static boolean hasMarker(ByteBuf buf) {
+        int ri = buf.readerIndex();
+        if (buf.readableBytes() < PREFIX_BYTES) {
+            return false;
+        }
+        for (int i = 0; i < MARKER.length; i++) {
+            if (buf.getByte(ri + i) != MARKER[i]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public Encoder getValueEncoder() {
+        return valueEncoder;
+    }
+
+    @Override
+    public Decoder<Object> getValueDecoder() {
+        return valueDecoder;
+    }
+
+    @Override
+    public Encoder getMapKeyEncoder() {
+        return delegate.getMapKeyEncoder();
+    }
+
+    @Override
+    public Decoder<Object> getMapKeyDecoder() {
+        return delegate.getMapKeyDecoder();
+    }
+}

ebean-redisson/src/main/java/io/ebean/redisson/near/NearCacheInvalidate.java

@@ -0,0 +1,24 @@
+package io.ebean.redisson.near;
+
+import java.util.Set;
+
+/**
+ * Near cache invalidation.
+ */
+public interface NearCacheInvalidate {
+
+    /**
+     * Invalidate from near cache the given keys.
+     */
+    void invalidateKeys(Set<Object> keySet);
+
+    /**
+     * Invalidate from near cache the given key.
+     */
+    void invalidateKey(Object id);
+
+    /**
+     * Clear the near cache.
+     */
+    void invalidateClear();
+}

ebean-redisson/src/main/java/io/ebean/redisson/near/NearCacheNotify.java

@@ -0,0 +1,24 @@
+package io.ebean.redisson.near;
+
+import java.util.Set;
+
+/**
+ * Notify other cluster members to invalidate parts of their near cache.
+ */
+public interface NearCacheNotify {
+
+    /**
+     * Invalidate the given keys.
+     */
+    void invalidateKeys(String cacheKey, Set<Object> keySet);
+
+    /**
+     * Invalidate a single key.
+     */
+    void invalidateKey(String cacheKey, Object id);
+
+    /**
+     * Clear a near cache.
+     */
+    void invalidateClear(String cacheKey);
+}

ebean-redisson/src/main/java/module-info.java

@@ -0,0 +1,13 @@
+import io.ebean.cache.ServerCachePlugin;
+
+/**
+ * Provider of ServerCachePlugin.
+ */
+open module io.ebean.redisson {
+
+  provides ServerCachePlugin with io.ebean.redisson.RedissonCachePlugin;
+
+  requires transitive io.ebean.core;
+  requires transitive redisson;
+  requires io.netty.buffer;
+}

ebean-redisson/src/main/resources/META-INF/services/io.ebean.cache.ServerCachePlugin

@@ -0,0 +1 @@
+io.ebean.redisson.RedissonCachePlugin

ebean-redisson/src/test/java/io/ebean/redisson/RedissonCacheFactoryTest.java

@@ -0,0 +1,148 @@
+package io.ebean.redisson;
+
+import io.ebean.DatabaseBuilder;
+import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheNotification;
+import io.ebean.cache.ServerCacheNotify;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import org.redisson.api.RedissonClient;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class RedissonCacheFactoryTest {
+
+  private static RedissonClient client;
+  private static RedissonCacheFactory factory;
+
+  @BeforeAll
+  static void connect() {
+    RedissonTestFixtures.startRedis();
+    assumeTrue(RedissonTestFixtures.isReachable(), "Skip: Redis not reachable");
+    client = RedissonTestFixtures.createClient();
+    DatabaseBuilder.Settings settings = RedissonTestFixtures.databaseSettings(client);
+    factory = new RedissonCacheFactory(settings, RedissonTestFixtures.backgroundExecutor());
+  }
+
+  @AfterAll
+  static void disconnect() {
+    if (client != null) client.shutdown();
+  }
+
+  @AfterEach
+  void clearCaches() {
+    // Individual tests clear their own caches inline
+  }
+
+  @Test
+  void createsNaturalKeyCache_roundTrip() {
+    String key = RedissonTestFixtures.cacheKey("factory-nk");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.naturalKeyConfig(key));
+    assertThat(cache).isInstanceOf(RedissonCache.class);
+
+    cache.put("1", "one");
+    assertThat(cache.get("1")).isEqualTo("one");
+    assertThat(((RedissonCache) cache).getHitCount()).isEqualTo(1);
+    cache.clear();
+  }
+
+  @Test
+  void createsBeanCache_withVersionGating() {
+    String key = RedissonTestFixtures.cacheKey("factory-bean");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.beanCacheConfig(key));
+    assertThat(cache).isInstanceOf(RedissonCache.class);
+    cache.clear();
+  }
+
+  @Test
+  void createsCollectionIdsCache() {
+    String key = RedissonTestFixtures.cacheKey("factory-coll");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.collectionIdsConfig(key));
+    assertThat(cache).isInstanceOf(RedissonCache.class);
+    cache.clear();
+  }
+
+  @Test
+  void createsNearCache_asDuelCache() {
+    String key = RedissonTestFixtures.cacheKey("factory-near");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.nearNaturalKeyConfig(key));
+    assertThat(cache).isInstanceOf(DuelCache.class);
+
+    cache.put("1", "near");
+    assertThat(cache.get("1")).isEqualTo("near");
+    cache.clear();
+  }
+
+  @Test
+  void createsNearBeanCache_asDuelCache_typeCheck() {
+    String key = RedissonTestFixtures.cacheKey("factory-near-bean");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.nearBeanCacheConfig(key));
+    assertThat(cache).isInstanceOf(DuelCache.class);
+    cache.clear();
+  }
+
+  @Test
+  void queryCache_isSingletonPerKey() {
+    String key = RedissonTestFixtures.cacheKey("factory-query");
+    ServerCache first = factory.createCache(RedissonTestFixtures.queryCacheConfig(key));
+    ServerCache second = factory.createCache(RedissonTestFixtures.queryCacheConfig(key));
+    assertThat(first).isSameAs(second);
+  }
+
+  @Test
+  void queryCacheClear_doesNotThrow() {
+    String key = RedissonTestFixtures.cacheKey("factory-query-clear");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.queryCacheConfig(key));
+    assertNotNull(cache);
+    cache.clear();
+  }
+
+  @Test
+  void cacheNotify_publishTableMod_doesNotThrow() {
+    ServerCacheNotify notify = factory.createCacheNotify(n -> {});
+    assertNotNull(notify);
+    notify.notify(new ServerCacheNotification(Set.of("tableA", "tableB")));
+  }
+
+  @Test
+  void cacheNotify_emptyTables_doesNotThrow() {
+    ServerCacheNotify notify = factory.createCacheNotify(n -> {});
+    assertNotNull(notify);
+    notify.notify(new ServerCacheNotification(Set.of()));
+  }
+
+  @Test
+  void cacheNotify_tableMod_notifiesOtherFactory() throws InterruptedException {
+    DatabaseBuilder.Settings otherSettings = RedissonTestFixtures.databaseSettings(client);
+    RedissonCacheFactory otherFactory = new RedissonCacheFactory(otherSettings, RedissonTestFixtures.backgroundExecutor());
+
+    CopyOnWriteArrayList<ServerCacheNotification> received = new CopyOnWriteArrayList<>();
+    otherFactory.createCacheNotify(received::add);
+
+    Thread.sleep(300);
+
+    ServerCacheNotify notify = factory.createCacheNotify(n -> {});
+    notify.notify(new ServerCacheNotification(Set.of("orders", "items")));
+
+    Thread.sleep(500);
+    assertThat(received).isNotEmpty();
+    assertThat(received.get(0).getDependentTables()).contains("orders", "items");
+  }
+
+  @Test
+  void usesInjectedRedissonClient() {
+    String key = RedissonTestFixtures.cacheKey("factory-inject");
+    ServerCache cache = factory.createCache(RedissonTestFixtures.naturalKeyConfig(key));
+    cache.put("ping", "pong");
+    assertThat(cache.get("ping")).isEqualTo("pong");
+    cache.clear();
+  }
+}

ebean-redisson/src/test/java/io/ebean/redisson/RedissonCacheTest.java

@@ -0,0 +1,219 @@
+package io.ebean.redisson;
+
+import io.ebean.cache.ServerCacheStatistics;
+import io.ebean.cache.ServerCacheType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import org.redisson.api.RedissonClient;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class RedissonCacheTest {
+
+  private static RedissonClient client;
+
+  private String cacheKey;
+  private RedissonCache cache;
+
+  @BeforeAll
+  static void connect() {
+    RedissonTestFixtures.startRedis();
+    assumeTrue(RedissonTestFixtures.isReachable(), "Skip: Redis not reachable");
+    client = RedissonTestFixtures.createClient();
+  }
+
+  @AfterAll
+  static void disconnect() {
+    if (client != null) client.shutdown();
+  }
+
+  void setUp(String suffix) {
+    cacheKey = RedissonTestFixtures.cacheKey(suffix);
+    cache = RedissonTestFixtures.naturalKeyCache(client, cacheKey);
+  }
+
+  @AfterEach
+  void clearCache() {
+    if (cache != null) cache.clear();
+  }
+
+  @Test
+  void putAndGet() {
+    setUp("putAndGet");
+    cache.put("1", "one");
+
+    assertThat(cache.get("1")).isEqualTo("one");
+    assertThat(cache.getHitCount()).isEqualTo(1);
+    assertThat(cache.getMissCount()).isZero();
+  }
+
+  @Test
+  void get_returnsNull_onMiss() {
+    setUp("getMiss");
+    assertThat(cache.get("nope")).isNull();
+    assertThat(cache.getMissCount()).isEqualTo(1);
+    assertThat(cache.getHitCount()).isZero();
+  }
+
+  @Test
+  void getAll_partialHit() {
+    setUp("getAll");
+    cache.put("1", "one");
+    cache.put("2", "two");
+
+    Map<Object, Object> found = cache.getAll(Set.of("1", "2", "3"));
+
+    assertThat(found).containsEntry("1", "one").containsEntry("2", "two").doesNotContainKey("3");
+    assertThat(cache.getHitCount()).isEqualTo(2);
+    assertThat(cache.getMissCount()).isEqualTo(1);
+  }
+
+  @Test
+  void getAll_emptyKeys_returnsEmptyMap() {
+    setUp("getAllEmpty");
+    assertThat(cache.getAll(Set.of())).isEmpty();
+  }
+
+  @Test
+  void getAll_allMiss_returnsEmptyMap() {
+    setUp("getAllMiss");
+    assertThat(cache.getAll(Set.of("x", "y"))).isEmpty();
+    assertThat(cache.getMissCount()).isEqualTo(2);
+  }
+
+  @Test
+  void getAll_resultKeys_areOriginalKeyObjects() {
+    setUp("getAllKeys");
+    cache.put("a", "A");
+    cache.put("b", "B");
+
+    Map<Object, Object> result = cache.getAll(Set.of("a", "b"));
+    assertThat(result.keySet()).containsExactlyInAnyOrder("a", "b");
+  }
+
+  @Test
+  void putAll() {
+    setUp("putAll");
+    Map<Object, Object> entries = new LinkedHashMap<>();
+    entries.put("x", "X");
+    entries.put("y", "Y");
+    cache.putAll(entries);
+
+    assertThat(cache.getAll(Set.of("x", "y")))
+      .containsEntry("x", "X")
+      .containsEntry("y", "Y");
+  }
+
+  @Test
+  void remove() {
+    setUp("remove");
+    cache.put("1", "one");
+    cache.remove("1");
+    assertThat(cache.get("1")).isNull();
+  }
+
+  @Test
+  void removeAll() {
+    setUp("removeAll");
+    cache.put("1", "one");
+    cache.put("2", "two");
+    cache.put("3", "three");
+    cache.removeAll(Set.of("1", "2"));
+
+    assertThat(cache.get("1")).isNull();
+    assertThat(cache.get("2")).isNull();
+    assertThat(cache.get("3")).isEqualTo("three");
+  }
+
+  @Test
+  void clear() {
+    setUp("clear");
+    cache.put("1", "one");
+    cache.put("2", "two");
+    cache.clear();
+    assertThat(cache.getAll(Set.of("1", "2"))).isEmpty();
+  }
+
+  @Test
+  void statistics_countsHitsMissesPutsRemoves() {
+    setUp("stats");
+    cache.put("1", "one");
+    cache.get("1");          // hit
+    cache.get("missing");    // miss
+    cache.remove("1");
+
+    ServerCacheStatistics stats = cache.statistics(true);
+    assertNotNull(stats);
+    assertThat(stats.getCacheName()).isEqualTo(cacheKey);
+    assertThat(stats.getHitCount()).isEqualTo(1);
+    assertThat(stats.getMissCount()).isEqualTo(1);
+    assertThat(stats.getPutCount()).isEqualTo(1);
+    assertThat(stats.getRemoveCount()).isEqualTo(1);
+  }
+
+  @Test
+  void statistics_reset_clearsCounters() {
+    setUp("statsReset");
+    cache.put("k", "v");
+    cache.get("k");
+
+    cache.statistics(true); // reset
+    ServerCacheStatistics after = cache.statistics(false);
+    assertNotNull(after);
+    assertThat(after.getHitCount()).isZero();
+    assertThat(after.getMissCount()).isZero();
+  }
+
+  @Test
+  void ttl_maxSecsToLive_entryStoredWithExpiry() throws InterruptedException {
+    String ttlKey = RedissonTestFixtures.cacheKey("ttl");
+    RedissonCache ttlCache = new RedissonCache(
+      client,
+      RedissonTestFixtures.cacheConfig(ServerCacheType.NATURAL_KEY, ttlKey,
+        RedissonTestFixtures.ttlOptions(2)),
+      new io.ebean.redisson.encode.SerializableCodec(), null, false);
+    try {
+      ttlCache.put("k", "v");
+      assertThat(ttlCache.get("k")).isEqualTo("v");
+
+      Thread.sleep(2500);
+      assertThat(ttlCache.get("k")).isNull(); // expired
+    } finally {
+      ttlCache.clear();
+    }
+  }
+
+  @Test
+  void trimCache_removesExcessEntries() {
+    String sizeKey = RedissonTestFixtures.cacheKey("trim");
+    RedissonCache sizedCache = new RedissonCache(
+      client,
+      RedissonTestFixtures.cacheConfig(ServerCacheType.NATURAL_KEY, sizeKey,
+        RedissonTestFixtures.maxSizeOptions(3)),
+      new io.ebean.redisson.encode.SerializableCodec(), null, false);
+    try {
+      for (int i = 0; i < 10; i++) {
+        sizedCache.put("k" + i, "v" + i);
+      }
+      sizedCache.trimCache();
+
+      // After trim the hash should be at or below maxSize
+      long remaining = 0;
+      for (int i = 0; i < 10; i++) {
+        if (sizedCache.get("k" + i) != null) remaining++;
+      }
+      assertThat(remaining).isLessThanOrEqualTo(3);
+    } finally {
+      sizedCache.clear();
+    }
+  }
+}

ebean-redisson/src/test/java/io/ebean/redisson/RedissonTestFixtures.java

@@ -0,0 +1,163 @@
+package io.ebean.redisson;
+
+import io.ebean.BackgroundExecutor;
+import io.ebean.Database;
+import io.ebean.DatabaseBuilder;
+import io.ebean.cache.ServerCacheConfig;
+import io.ebean.cache.ServerCacheOptions;
+import io.ebean.cache.ServerCacheType;
+import io.ebean.redisson.encode.SerializableCodec;
+import io.ebean.test.containers.RedisContainer;
+import org.jspecify.annotations.NonNull;
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+
+import java.io.InputStream;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+final class RedissonTestFixtures {
+
+  private RedissonTestFixtures() {}
+
+  static String cacheKey(String suffix) {
+    return "ebean-redisson-it:" + suffix + ":" + System.nanoTime();
+  }
+
+  static ServerCacheOptions defaultOptions() {
+    return new ServerCacheOptions();
+  }
+
+  static ServerCacheOptions ttlOptions(int secsToLive) {
+    ServerCacheOptions opts = new ServerCacheOptions();
+    opts.setMaxSecsToLive(secsToLive);
+    return opts;
+  }
+
+  static ServerCacheOptions idleOptions(int maxIdleSecs) {
+    ServerCacheOptions opts = new ServerCacheOptions();
+    opts.setMaxIdleSecs(maxIdleSecs);
+    return opts;
+  }
+
+  static ServerCacheOptions maxSizeOptions(int maxSize) {
+    ServerCacheOptions opts = new ServerCacheOptions();
+    opts.setMaxSize(maxSize);
+    return opts;
+  }
+
+  static ServerCacheConfig cacheConfig(ServerCacheType type, String cacheKey, ServerCacheOptions options) {
+    return new ServerCacheConfig(type, cacheKey, "testCache", options, null, null);
+  }
+
+  static ServerCacheConfig naturalKeyConfig(String cacheKey) {
+    return cacheConfig(ServerCacheType.NATURAL_KEY, cacheKey, defaultOptions());
+  }
+
+  static ServerCacheConfig beanCacheConfig(String cacheKey) {
+    return cacheConfig(ServerCacheType.BEAN, cacheKey, defaultOptions());
+  }
+
+  static ServerCacheConfig collectionIdsConfig(String cacheKey) {
+    return cacheConfig(ServerCacheType.COLLECTION_IDS, cacheKey, defaultOptions());
+  }
+
+  static ServerCacheConfig queryCacheConfig(String cacheKey) {
+    return cacheConfig(ServerCacheType.QUERY, cacheKey, defaultOptions());
+  }
+
+  static ServerCacheConfig nearNaturalKeyConfig(String cacheKey) {
+    ServerCacheOptions opts = defaultOptions();
+    opts.setNearCache(true);
+    return cacheConfig(ServerCacheType.NATURAL_KEY, cacheKey, opts);
+  }
+
+  static ServerCacheConfig nearBeanCacheConfig(String cacheKey) {
+    ServerCacheOptions opts = defaultOptions();
+    opts.setNearCache(true);
+    return cacheConfig(ServerCacheType.BEAN, cacheKey, opts);
+  }
+
+  static RedissonCache naturalKeyCache(RedissonClient client, String cacheKey) {
+    return new RedissonCache(client, naturalKeyConfig(cacheKey), new SerializableCodec(), null, false);
+  }
+
+  static RedissonCache beanCache(RedissonClient client, String cacheKey) {
+    return new RedissonCache(client, beanCacheConfig(cacheKey), new SerializableCodec(), null, false);
+  }
+
+  static DatabaseBuilder.Settings databaseSettings(RedissonClient client) {
+    return Database.builder()
+      .name("redisson-factory-test")
+      .putServiceObject(client)
+      .settings();
+  }
+
+  /**
+   * Starts the Redis test container if it is not already running.
+   * Idempotent: safe to call from multiple test classes; the container
+   * library detects an already-running instance and skips startup.
+   */
+  static void startRedis() {
+    RedisContainer.builder("latest").start();
+  }
+
+  /**
+   * Returns true when Redis is reachable on the configured address.
+   * Uses a 500ms / zero-retry probe so CI skips fast instead of waiting
+   * through the full connectTimeout + retryAttempts in redisson-config.yaml.
+   */
+  static boolean isReachable() {
+    try {
+      Config probe = loadConfig();
+      probe.useSingleServer()
+        .setConnectTimeout(500)
+        .setTimeout(500)
+        .setRetryAttempts(0)
+        .setConnectionMinimumIdleSize(1)
+        .setConnectionPoolSize(1);
+      RedissonClient c = Redisson.create(probe);
+      c.shutdown();
+      return true;
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
+  static RedissonClient createClient() {
+    return Redisson.create(loadConfig());
+  }
+
+  private static Config loadConfig() {
+    InputStream is = RedissonTestFixtures.class.getClassLoader()
+      .getResourceAsStream("redisson-config.yaml");
+    if (is != null) {
+      return Config.fromYAML(is);
+    }
+    Config cfg = new Config();
+    cfg.useSingleServer().setAddress("redis://localhost:6379");
+    return cfg;
+  }
+
+  static BackgroundExecutor backgroundExecutor() {
+    ScheduledExecutorService ex = Executors.newSingleThreadScheduledExecutor(r -> {
+      Thread t = new Thread(r, "ebean-redisson-test-bg");
+      t.setDaemon(true);
+      return t;
+    });
+    return new BackgroundExecutor() {
+      @Override public <T> Future<T> submit(Callable<T> task) { return ex.submit(task); }
+      @Override public Future<?> submit(Runnable task) { return ex.submit(task); }
+      @Override public void execute(Runnable task) { ex.execute(task); }
+      @Override public ScheduledFuture<?> scheduleWithFixedDelay(@NonNull Runnable t, long i, long d, @NonNull TimeUnit u) { return ex.scheduleWithFixedDelay(t, i, d, u); }
+      @Override public ScheduledFuture<?> scheduleAtFixedRate(@NonNull Runnable t, long i, long p, @NonNull TimeUnit u) { return ex.scheduleAtFixedRate(t, i, p, u); }
+      @Override public ScheduledFuture<?> schedule(@NonNull Runnable t, long d, @NonNull TimeUnit u) { return ex.schedule(t, d, u); }
+      @Override public <V> ScheduledFuture<V> schedule(@NonNull Callable<V> t, long d, @NonNull TimeUnit u) { return ex.schedule(t, d, u); }
+    };
+  }
+}

ebean-redisson/src/test/java/io/ebean/redisson/encode/CacheCodecTest.java

@@ -0,0 +1,92 @@
+package io.ebean.redisson.encode;
+
+import io.ebean.cache.TenantAwareKey;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.junit.jupiter.api.Test;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link CacheCodec} key encoder/decoder.
+ *
+ * Regression guard for the decoder bug where "id:tenantId" was decoded to "tenantId"
+ * (only the part after the first colon), causing every tenant-aware getAll() lookup to miss.
+ */
+class CacheCodecTest {
+
+  // Use SerializableCodec as a concrete CacheCodec (only the key codec matters here)
+  private final CacheCodec codec = new SerializableCodec();
+  private final Encoder keyEncoder = codec.getMapKeyEncoder();
+  private final Decoder<Object> keyDecoder = codec.getMapKeyDecoder();
+
+  // ── encoder ──────────────────────────────────────────────────────────────
+
+  @Test
+  void encoder_plainStringKey() throws Exception {
+    ByteBuf buf = keyEncoder.encode("42");
+    byte[] bytes = new byte[buf.readableBytes()];
+    buf.readBytes(bytes);
+    assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("42");
+    buf.release();
+  }
+
+  @Test
+  void encoder_tenantAwareCacheKey() throws Exception {
+    TenantAwareKey.CacheKey key = new TenantAwareKey.CacheKey(123L, "tenantA");
+    ByteBuf buf = keyEncoder.encode(key);
+    byte[] bytes = new byte[buf.readableBytes()];
+    buf.readBytes(bytes);
+    // CacheKey.toString() = "123:tenantA"
+    assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("123:tenantA");
+    buf.release();
+  }
+
+  // ── decoder – regression guard ───────────────────────────────────────────
+
+  @Test
+  void decoder_plainKey_returnsFullString() throws Exception {
+    ByteBuf buf = Unpooled.wrappedBuffer("42".getBytes(StandardCharsets.UTF_8));
+    Object decoded = keyDecoder.decode(buf, null);
+    assertThat(decoded).isEqualTo("42");
+  }
+
+  @Test
+  void decoder_keyContainingColon_returnsFullString() throws Exception {
+    // REGRESSION: old decoder did substring(pos+1) which turned "123:tenantA" -> "tenantA"
+    ByteBuf buf = Unpooled.wrappedBuffer("123:tenantA".getBytes(StandardCharsets.UTF_8));
+    Object decoded = keyDecoder.decode(buf, null);
+    assertThat(decoded).isEqualTo("123:tenantA");  // must NOT be just "tenantA"
+  }
+
+  @Test
+  void decoder_keyWithMultipleColons_returnsFullString() throws Exception {
+    // E.g. UUID-style key with colon in tenantId
+    ByteBuf buf = Unpooled.wrappedBuffer("key:ten:ant".getBytes(StandardCharsets.UTF_8));
+    Object decoded = keyDecoder.decode(buf, null);
+    assertThat(decoded).isEqualTo("key:ten:ant");
+  }
+
+  // ── round-trip ───────────────────────────────────────────────────────────
+
+  @Test
+  void roundTrip_plainString() throws Exception {
+    String original = "99";
+    ByteBuf encoded = keyEncoder.encode(original);
+    Object decoded = keyDecoder.decode(encoded, null);
+    assertThat(decoded).isEqualTo(original);
+  }
+
+  @Test
+  void roundTrip_tenantKey() throws Exception {
+    TenantAwareKey.CacheKey key = new TenantAwareKey.CacheKey(7L, "tenant42");
+    ByteBuf encoded = keyEncoder.encode(key);
+    Object decoded = keyDecoder.decode(encoded, null);
+    // The decoded value is the string representation used as the Redis field name
+    assertThat(decoded).isEqualTo("7:tenant42");
+  }
+}

ebean-redisson/src/test/java/io/ebean/redisson/encode/SerializableCodecTest.java

@@ -0,0 +1,50 @@
+package io.ebean.redisson.encode;
+
+import io.netty.buffer.ByteBuf;
+import org.junit.jupiter.api.Test;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link SerializableCodec} value encoder/decoder.
+ * Mirrors ebean-redis {@code EncodeSerializableTest}.
+ */
+class SerializableCodecTest {
+
+  private final SerializableCodec codec = new SerializableCodec();
+  private final Encoder encoder = codec.getValueEncoder();
+  private final Decoder<Object> decoder = codec.getValueDecoder();
+
+  @Test
+  void roundTrip_string() throws Exception {
+    ByteBuf buf = encoder.encode("HelloWorld");
+    Object result = decoder.decode(buf, null);
+    assertThat(result).isEqualTo("HelloWorld");
+  }
+
+  @Test
+  void roundTrip_long() throws Exception {
+    ByteBuf buf = encoder.encode(42L);
+    Object result = decoder.decode(buf, null);
+    assertThat(result).isEqualTo(42L);
+  }
+
+  @Test
+  void roundTrip_list() throws Exception {
+    List<String> original = List.of("a", "b", "c");
+    ByteBuf buf = encoder.encode(original);
+    Object result = decoder.decode(buf, null);
+    assertThat(result).isEqualTo(original);
+  }
+
+  @Test
+  void roundTrip_null() throws Exception {
+    ByteBuf buf = encoder.encode(null);
+    Object result = decoder.decode(buf, null);
+    assertThat(result).isNull();
+  }
+}

ebean-redisson/src/test/java/io/ebean/redisson/encode/VersionGatedCodecTest.java

@@ -0,0 +1,129 @@
+package io.ebean.redisson.encode;
+
+import io.ebeaninternal.server.cache.CachedBeanData;
+import io.netty.buffer.ByteBuf;
+import org.junit.jupiter.api.Test;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link VersionGatedCodec}.
+ *
+ * Verifies that:
+ * <ul>
+ *   <li>The encoder prepends the magic MARKER + 8-byte big-endian version</li>
+ *   <li>The decoder strips the prefix before delegating to the inner codec</li>
+ *   <li>The decoder tolerates absent marker (backward compat)</li>
+ *   <li>The key encoder/decoder are delegated to the inner codec unchanged</li>
+ * </ul>
+ */
+class VersionGatedCodecTest {
+
+  private final CachedBeanDataCodec inner = new CachedBeanDataCodec();
+  private final VersionGatedCodec codec = new VersionGatedCodec(inner);
+
+  private final Encoder valueEncoder = codec.getValueEncoder();
+  private final Decoder<Object> valueDecoder = codec.getValueDecoder();
+
+  // ── marker structure ─────────────────────────────────────────────────────
+
+  @Test
+  void encoder_prependsMarkerAndVersion() throws Exception {
+    CachedBeanData data = beanData(5L);
+    ByteBuf encoded = valueEncoder.encode(data);
+
+    // First 2 bytes must be the magic marker
+    byte b0 = encoded.getByte(0);
+    byte b1 = encoded.getByte(1);
+    assertThat(b0).isEqualTo(VersionGatedCodec.MARKER[0]);
+    assertThat(b1).isEqualTo(VersionGatedCodec.MARKER[1]);
+
+    // Next 8 bytes are the version (big-endian long = 5)
+    long version = encoded.getLong(2);
+    assertThat(version).isEqualTo(5L);
+
+    // Total length > PREFIX_BYTES
+    assertThat(encoded.readableBytes()).isGreaterThan(VersionGatedCodec.PREFIX_BYTES);
+    encoded.release();
+  }
+
+  @Test
+  void encoder_zeroVersion_whenNoCachedBeanData() throws Exception {
+    // Non-CachedBeanData value → version treated as 0
+    CachedBeanDataCodec codec2 = new CachedBeanDataCodec();
+    VersionGatedCodec gated = new VersionGatedCodec(codec2);
+
+    CachedBeanData data = beanData(0L);
+    ByteBuf encoded = gated.getValueEncoder().encode(data);
+    long version = encoded.getLong(2);
+    assertThat(version).isEqualTo(0L);
+    encoded.release();
+  }
+
+  // ── round-trip ───────────────────────────────────────────────────────────
+
+  @Test
+  void roundTrip_versionedBeanData() throws Exception {
+    CachedBeanData original = beanData(3L);
+    ByteBuf encoded = valueEncoder.encode(original);
+    Object decoded = valueDecoder.decode(encoded, null);
+
+    assertThat(decoded).isInstanceOf(CachedBeanData.class);
+    CachedBeanData result = (CachedBeanData) decoded;
+    assertThat(result.getVersion()).isEqualTo(3L);
+  }
+
+  @Test
+  void roundTrip_zeroVersion() throws Exception {
+    CachedBeanData original = beanData(0L);
+    ByteBuf encoded = valueEncoder.encode(original);
+    Object decoded = valueDecoder.decode(encoded, null);
+
+    assertThat(decoded).isInstanceOf(CachedBeanData.class);
+    assertThat(((CachedBeanData) decoded).getVersion()).isEqualTo(0L);
+  }
+
+  @Test
+  void decoder_toleratesAbsentMarker() throws Exception {
+    // Data without the marker prefix (simulates data stored before VersionGatedCodec was added)
+    CachedBeanData original = beanData(1L);
+    ByteBuf rawEncoded = inner.getValueEncoder().encode(original);
+
+    // Decoder must NOT throw and must still return a valid CachedBeanData
+    Object decoded = valueDecoder.decode(rawEncoded, null);
+    assertThat(decoded).isInstanceOf(CachedBeanData.class);
+  }
+
+  // ── key codec delegation ─────────────────────────────────────────────────
+
+  @Test
+  void keyEncoder_delegatesToInner() throws Exception {
+    // VersionGatedCodec must delegate key encoding to the inner codec
+    ByteBuf fromGated = codec.getMapKeyEncoder().encode("myKey");
+    ByteBuf fromInner = inner.getMapKeyEncoder().encode("myKey");
+
+    byte[] gatedBytes = new byte[fromGated.readableBytes()];
+    fromGated.readBytes(gatedBytes);
+    byte[] innerBytes = new byte[fromInner.readableBytes()];
+    fromInner.readBytes(innerBytes);
+
+    assertThat(gatedBytes).isEqualTo(innerBytes);
+    fromGated.release();
+    fromInner.release();
+  }
+
+  @Test
+  void keyDecoder_delegatesToInner() throws Exception {
+    ByteBuf buf = codec.getMapKeyEncoder().encode("someKey");
+    Object decoded = codec.getMapKeyDecoder().decode(buf, null);
+    assertThat(decoded).isEqualTo("someKey");
+  }
+
+  // ── helper ───────────────────────────────────────────────────────────────
+
+  private static CachedBeanData beanData(long version) {
+    return new CachedBeanData(null, null, java.util.Collections.emptyMap(), version);
+  }
+}

ebean-redisson/src/test/java/org/domain/EBase.java

@@ -0,0 +1,58 @@
+package org.domain;
+
+import io.ebean.Model;
+import io.ebean.annotation.WhenCreated;
+import io.ebean.annotation.WhenModified;
+import jakarta.persistence.Id;
+import jakarta.persistence.MappedSuperclass;
+import jakarta.persistence.Version;
+
+import java.time.Instant;
+
+@MappedSuperclass
+public class EBase extends Model {
+
+  @Id
+  protected long id;
+
+  @Version
+  protected long version;
+
+  @WhenCreated
+  protected Instant whenCreated;
+
+  @WhenModified
+  protected Instant whenModified;
+
+  public long getId() {
+    return id;
+  }
+
+  public void setId(long id) {
+    this.id = id;
+  }
+
+  public long getVersion() {
+    return version;
+  }
+
+  public void setVersion(long version) {
+    this.version = version;
+  }
+
+  public Instant getWhenCreated() {
+    return whenCreated;
+  }
+
+  public void setWhenCreated(Instant whenCreated) {
+    this.whenCreated = whenCreated;
+  }
+
+  public Instant getWhenModified() {
+    return whenModified;
+  }
+
+  public void setWhenModified(Instant whenModified) {
+    this.whenModified = whenModified;
+  }
+}

ebean-redisson/src/test/java/org/domain/OtherOne.java

@@ -0,0 +1,40 @@
+package org.domain;
+
+
+import io.ebean.annotation.Cache;
+import jakarta.persistence.Entity;
+
+/**
+ * Using Natural Key caching but no Near Caching so always hitting Redis.
+ */
+@SuppressWarnings("unused")
+@Cache(naturalKey = {"one", "two"})
+@Entity
+public class OtherOne extends EBase {
+
+  private final String one;
+  private final String two;
+  private String notes;
+
+  public OtherOne(String one, String two, String notes) {
+    this.one = one;
+    this.two = two;
+    this.notes = notes;
+  }
+
+  public String one() {
+    return one;
+  }
+
+  public String two() {
+    return two;
+  }
+
+  public String notes() {
+    return notes;
+  }
+
+  public void setNotes(String notes) {
+    this.notes = notes;
+  }
+}

ebean-redisson/src/test/java/org/domain/Person.java

@@ -0,0 +1,76 @@
+package org.domain;
+
+
+import io.ebean.annotation.Cache;
+import io.ebean.annotation.CacheBeanTuning;
+import io.ebean.annotation.Index;
+import jakarta.persistence.Entity;
+
+import java.time.LocalDate;
+
+@Cache(enableQueryCache = true, nearCache = true, naturalKey = "name")
+@CacheBeanTuning(maxSecsToLive = 1)
+@Entity
+public class Person extends EBase {
+
+  public enum Status {
+    NEW,
+    ACTIVE,
+    INACTIVE
+  }
+
+  @Index(unique = true)
+  String name;
+
+  Status status;
+
+  LocalDate localDate;
+
+  String notes;
+
+  /**
+   * Test that KEY and VALUE are now by default not h2database keywords.
+   */
+  String key;
+
+  public Person(String name) {
+    this.name = name;
+    this.status = Status.NEW;
+  }
+
+  public String toString() {
+    return "[id:" + id + " name:" + name + "date:" + localDate + ']';
+  }
+
+  public Status getStatus() {
+    return status;
+  }
+
+  public void setStatus(Status status) {
+    this.status = status;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public void setName(String name) {
+    this.name = name;
+  }
+
+  public String getNotes() {
+    return notes;
+  }
+
+  public void setNotes(String notes) {
+    this.notes = notes;
+  }
+
+  public LocalDate getLocalDate() {
+    return localDate;
+  }
+
+  public void setLocalDate(LocalDate localDate) {
+    this.localDate = localDate;
+  }
+}

ebean-redisson/src/test/java/org/domain/RCust.java

@@ -0,0 +1,27 @@
+package org.domain;
+
+
+import io.ebean.annotation.Cache;
+import io.ebean.annotation.Index;
+import jakarta.persistence.Entity;
+
+@Cache(naturalKey = "name")
+@Entity
+public class RCust extends EBase {
+
+  @Index(unique = true)
+  String name;
+
+  public RCust(String name) {
+    this.name = name;
+  }
+
+  public String getName() {
+    return name;
+  }
+
+  public void setName(String name) {
+    this.name = name;
+  }
+
+}

ebean-redisson/src/test/java/org/domain/UChild.java

@@ -0,0 +1,38 @@
+package org.domain;
+
+import io.ebean.Model;
+import io.ebean.annotation.Cache;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.ManyToOne;
+
+@Cache
+@Entity
+public class UChild extends Model {
+
+  @Id
+  long id;
+
+  String name;
+
+  @ManyToOne
+  final UParent parent;
+
+  public UChild(UParent parent, String name) {
+    this.parent = parent;
+    this.name = name;
+  }
+
+  public long id() {
+    return id;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public void setName(String name) {
+    this.name = name;
+  }
+
+}

ebean-redisson/src/test/java/org/domain/UParent.java

@@ -0,0 +1,51 @@
+package org.domain;
+
+import io.ebean.Model;
+import io.ebean.annotation.Cache;
+import io.ebean.annotation.CacheBeanTuning;
+import jakarta.persistence.CascadeType;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.OneToMany;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+@Cache(enableQueryCache = true)
+@CacheBeanTuning(maxSecsToLive = 1)
+@Entity
+public class UParent extends Model {
+
+  @Id
+  private UUID id;
+
+  private String name;
+
+  @OneToMany(mappedBy = "parent", cascade = CascadeType.ALL)
+  private final List<UChild> children = new ArrayList<>();
+
+  public UParent(String name) {
+    this.name = name;
+  }
+
+  public UUID id() {
+    return id;
+  }
+
+  public void setId(UUID id) {
+    this.id = id;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public void setName(String name) {
+    this.name = name;
+  }
+
+  public List<UChild> children() {
+    return children;
+  }
+}

ebean-redisson/src/test/java/org/domain/test/TestOne.java

@@ -0,0 +1,47 @@
+package org.domain.test;
+
+import io.ebean.Model;
+import io.ebean.annotation.Cache;
+import jakarta.persistence.*;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Cache
+@Entity
+public class TestOne extends Model {
+
+    @Id
+    private String id;
+
+    @Column(unique = true)
+    private String otherUnique;
+
+    @OneToMany(mappedBy = "testOne", cascade = CascadeType.ALL, orphanRemoval = true)
+    private List<TestTwo> testTwos = new ArrayList<>();
+
+    public TestOne(String id, String otherUnique) {
+        this.id = id;
+        this.otherUnique = otherUnique;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public String getOtherUnique() {
+        return otherUnique;
+    }
+
+    public void setOtherUnique(String otherUnique) {
+        this.otherUnique = otherUnique;
+    }
+
+    public List<TestTwo> getTestTwos() {
+        return testTwos;
+    }
+
+    public void setTestTwos(List<TestTwo> testTwos) {
+        this.testTwos = testTwos;
+    }
+}

ebean-redisson/src/test/java/org/domain/test/TestTwo.java

@@ -0,0 +1,34 @@
+package org.domain.test;
+
+import io.ebean.Model;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToOne;
+
+@Entity
+public class TestTwo extends Model {
+
+    @Id
+    private String id;
+
+    @ManyToOne
+    @JoinColumn
+    private TestOne testOne;
+
+    public TestTwo(String id) {
+        this.id = id;
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public TestOne getTestOne() {
+        return testOne;
+    }
+
+    public void setTestOne(TestOne testOne) {
+        this.testOne = testOne;
+    }
+}

ebean-redisson/src/test/java/org/integration/ClusterTest.java

@@ -0,0 +1,118 @@
+package org.integration;
+
+import io.ebean.DB;
+import io.ebean.Database;
+import io.ebean.redisson.DuelCache;
+import org.domain.Person;
+import org.domain.query.QPerson;
+import org.junit.jupiter.api.Test;
+
+import javax.sql.DataSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ClusterTest {
+
+  private Database createOther(DataSource dataSource) {
+    return Database.builder()
+      .dataSource(dataSource)
+      .loadFromProperties()
+      .defaultDatabase(false)
+      .name("other")
+      .ddlGenerate(false)
+      .ddlRun(false)
+      .build();
+  }
+
+  @Test
+  void testBothNear() throws InterruptedException {
+    // ensure the default server exists first
+    final Database db = DB.getDefault();
+    Database other = createOther(db.pluginApi().dataSource());
+
+    new QPerson()
+      .name.eq("Someone")
+      .delete();
+
+    Person foo = new Person("Someone");
+    foo.save();
+
+    DB.cacheManager().clearAll();
+    db.metaInfo().resetAllMetrics();
+    other.metaInfo().resetAllMetrics();
+
+    Person fooA = DB.find(Person.class, foo.getId());
+    allowAsyncMessaging(); // allow time for background cache load
+    Person fooB = other.find(Person.class, foo.getId());
+
+    DuelCache dualCacheA = db.cacheManager().beanCache(Person.class).unwrap(DuelCache.class);
+    assertCounts(dualCacheA, 0, 1, 0, 1);
+    fooA = DB.find(Person.class, foo.getId());
+    assertCounts(dualCacheA, 1, 1, 0, 1);
+    fooB = other.find(Person.class, foo.getId());
+    fooA = DB.find(Person.class, foo.getId());
+    assertCounts(dualCacheA, 2, 1, 0, 1);
+    fooB = other.find(Person.class, foo.getId());
+    DuelCache dualCacheB = other.cacheManager().beanCache(Person.class).unwrap(DuelCache.class);
+    assertCounts(dualCacheB, 2, 1, 1, 0);
+  }
+
+  @Test
+  void test() throws InterruptedException {
+    // ensure the default server exists first
+    final Database db = DB.getDefault();
+    Database other = createOther(db.pluginApi().dataSource());
+
+    for (int i = 0; i < 10; i++) {
+      Person foo = new Person("name " + i);
+      foo.save();
+    }
+
+    other.cacheManager().clearAll();
+    other.metaInfo().resetAllMetrics();
+
+    DuelCache dualCache = other.cacheManager().beanCache(Person.class).unwrap(DuelCache.class);
+
+    Person foo0 = other.find(Person.class, 1);
+    assertCounts(dualCache, 0, 1, 0, 1);
+
+    other.find(Person.class, 1);
+    assertCounts(dualCache, 1, 1, 0, 1);
+
+    other.find(Person.class, 1);
+    assertCounts(dualCache, 2, 1, 0, 1);
+
+    other.find(Person.class, 1);
+    assertCounts(dualCache, 3, 1, 0, 1);
+
+    other.find(Person.class, 2);
+    assertCounts(dualCache, 3, 2, 0, 2);
+
+    foo0.setName("name2");
+    foo0.save();
+    allowAsyncMessaging();
+
+    Person foo3 = other.find(Person.class, 1);
+    assertThat(foo3.getName()).isEqualTo("name2");
+    assertCounts(dualCache, 3, 3, 1, 2);
+
+    foo0.setName("name3");
+    foo0.save();
+    allowAsyncMessaging();
+
+    foo3 = other.find(Person.class, 1);
+    assertThat(foo3.getName()).isEqualTo("name3");
+    assertCounts(dualCache, 3, 4, 2, 2);
+  }
+
+  private void assertCounts(DuelCache dualCache, int nearHits, int nearMiss, int remoteHit, int remoteMiss) {
+    assertThat(dualCache.getNearHitCount()).isEqualTo(nearHits);
+    assertThat(dualCache.getNearMissCount()).isEqualTo(nearMiss);
+    assertThat(dualCache.getRemoteHitCount()).isEqualTo(remoteHit);
+    assertThat(dualCache.getRemoteMissCount()).isEqualTo(remoteMiss);
+  }
+
+  private void allowAsyncMessaging() throws InterruptedException {
+    Thread.sleep(200);
+  }
+}

ebean-redisson/src/test/java/org/integration/IntegrationTest.java

@@ -0,0 +1,347 @@
+package org.integration;
+
+import io.ebean.DB;
+import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheStatistics;
+import io.ebeaninternal.server.cache.CachedBeanData;
+import org.domain.*;
+import org.domain.query.QOtherOne;
+import org.domain.query.QPerson;
+import org.domain.query.QRCust;
+import org.domain.test.TestOne;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class IntegrationTest {
+
+    private static OtherOne findOther(String a, String b) {
+        return new QOtherOne()
+            .one.eq(a)
+            .two.eq(b)
+            .findOne();
+    }
+
+    @Test
+    void uuid_getPut() {
+        UParent b0 = new UParent("b0");
+        b0.children().add(new UChild(b0, "b0c0"));
+        b0.children().add(new UChild(b0, "b0c1"));
+        b0.save();
+
+        ServerCache beanCache = DB.cacheManager().beanCache(UParent.class);
+        beanCache.clear();
+        beanCache.statistics(true);
+
+        UParent found0 = DB.find(UParent.class, b0.id());
+        assertThat(found0.name()).isEqualTo("b0");
+
+        List<UChild> children = found0.children();
+        assertThat(children).hasSize(2);
+
+        UParent found1 = DB.find(UParent.class, b0.id());
+        assertThat(found1.name()).isEqualTo("b0");
+
+        DB.delete(found1);
+
+        ServerCacheStatistics stats1 = beanCache.statistics(true);
+        assertThat(stats1.getHitCount()).isEqualTo(1);
+    }
+
+    @Test
+    void mget_when_emptyCollectionOfIds() {
+
+        List<RCust> f0 = new QRCust()
+            .setIdIn(Collections.emptyList())
+            .findList();
+
+        assertThat(f0).isEmpty();
+
+        List<RCust> f1 = new QRCust()
+            .id.in(Collections.emptyList())
+            .findList();
+
+        assertThat(f1).isEmpty();
+    }
+
+    @Test
+    void mput_via_setIdIn() throws InterruptedException {
+
+        ServerCache beanCache = DB.cacheManager().beanCache(RCust.class);
+        beanCache.clear();
+        beanCache.statistics(true);
+
+        List<RCust> people = new ArrayList<>();
+        for (String name : new String[]{"mp0", "mp1", "mp2"}) {
+            people.add(new RCust(name));
+        }
+        DB.saveAll(people);
+        List<Long> ids = people.stream().map(RCust::getId).collect(Collectors.toList());
+
+        List<RCust> f0 = new QRCust()
+            .setIdIn(ids) // using collection argument
+            .findList();
+
+        assertThat(f0).hasSize(3);
+        ServerCacheStatistics stats0 = beanCache.statistics(true);
+        assertThat(stats0.getHitCount()).isEqualTo(0);
+
+        Thread.sleep(5);
+
+        // we will hit the cache this time
+        List<RCust> f1 = new QRCust()
+            .setIdIn(ids.toArray()) // using varargs argument
+            .findList();
+
+        assertThat(f1).hasSize(3);
+        ServerCacheStatistics stats1 = beanCache.statistics(true);
+        assertThat(stats1.getHitCount()).isEqualTo(3);
+
+        // we will hit the cache again
+        List<RCust> f2 = new QRCust()
+            .setIdIn(ids) // using collection argument
+            .findList();
+        assertThat(f2).hasSize(3);
+        ServerCacheStatistics stats2 = beanCache.statistics(true);
+        assertThat(stats2.getHitCount()).isEqualTo(3);
+    }
+
+    @Test
+    void mput_via_propertyInExpression() throws InterruptedException {
+
+        ServerCache beanCache = DB.cacheManager().beanCache(RCust.class);
+        beanCache.clear();
+        beanCache.statistics(true);
+
+        List<RCust> people = new ArrayList<>();
+        for (String name : new String[]{"mpx0", "mpx1", "mpx2"}) {
+            people.add(new RCust(name));
+        }
+        DB.saveAll(people);
+        List<Long> ids = people.stream().map(RCust::getId).collect(Collectors.toList());
+
+        List<RCust> f0 = new QRCust()
+            .id.in(ids)
+            .findList();
+
+        assertThat(f0).hasSize(3);
+        ServerCacheStatistics stats0 = beanCache.statistics(true);
+        assertThat(stats0.getHitCount()).isEqualTo(0);
+
+        Thread.sleep(5);
+
+        // we will hit the cache this time
+        List<RCust> f1 = new QRCust()
+            .id.in(ids)
+            .findList();
+
+        assertThat(f1).hasSize(3);
+        ServerCacheStatistics stats1 = beanCache.statistics(true);
+        assertThat(stats1.getHitCount()).isEqualTo(3);
+
+        // we will hit the cache again
+        List<RCust> f2 = new QRCust()
+            .id.isIn(ids)
+            .findList();
+        assertThat(f2).hasSize(3);
+        ServerCacheStatistics stats2 = beanCache.statistics(true);
+        assertThat(stats2.getHitCount()).isEqualTo(3);
+    }
+
+    @Test
+    void testOtherOne() {
+        DB.save(new OtherOne("A", "B", "ab"));
+        DB.save(new OtherOne("A", "C", "ac"));
+        DB.save(new OtherOne("B", "B", "bb"));
+
+        ServerCache nkeyCache = DB.cacheManager().naturalKeyCache(OtherOne.class);
+        nkeyCache.clear();
+        nkeyCache.statistics(true);
+
+        OtherOne ab0 = findOther("A", "B");
+        OtherOne ab1 = findOther("A", "B");
+        OtherOne ab2 = findOther("A", "B");
+        OtherOne bb = findOther("B", "B");
+
+        assertThat(ab0).isNotNull();
+        assertThat(ab1).isNotNull();
+        assertThat(ab2).isNotNull();
+        assertThat(bb).isNotNull();
+
+        ServerCacheStatistics statistics = nkeyCache.statistics(true);
+        assertThat(statistics.getHitCount()).isEqualTo(2);
+    }
+
+    @Test
+    void test() throws InterruptedException {
+
+        insertSomePeople();
+
+        Person fiona = findByName("Fiona");
+        fiona.setName("Fortuna");
+        fiona.setLocalDate(LocalDate.now());
+        fiona.update();
+
+        Thread.sleep(100);
+
+        Person one = findById(1);
+        assertThat(one).isNotNull();
+
+        for (int i = 1; i < 4; i++) {
+            System.out.println("hit " + findById(i));
+        }
+
+        List<Person> one2 = nameStartsWith("fo");
+        assertThat(one2).hasSize(1);
+
+        one2 = nameStartsWith("j");
+        assertThat(one2).hasSize(2);
+
+        one2 = nameStartsWith("j");
+        assertThat(one2).hasSize(2);
+
+
+        List<Person> byNames = findByNames("Jack", "Rob");
+        assertThat(byNames).hasSize(2);
+
+        byNames = findByNames("Jack", "Rob", "Moby");
+        assertThat(byNames).hasSize(3);
+
+        fiona.setName("fo2");
+        fiona.setLocalDate(LocalDate.now());
+        fiona.update();
+
+        byNames = findByNames("Jack", "Rob", "Moby");
+        assertThat(byNames).hasSize(3);
+
+        Thread.sleep(200);
+
+        one2 = nameStartsWith("fo%");
+        System.out.println("one2 " + one2);
+        one2 = nameStartsWith("f0%");
+
+        System.out.println("one2 " + one2);
+
+        DB.cacheManager().clear(Person.class);
+
+        System.out.println("done");
+    }
+
+    private void insertSomePeople() {
+        List<Person> people = new ArrayList<>();
+        for (String name : new String[]{"Jack", "John", "Rob", "Moby", "Fiona"}) {
+            people.add(new Person(name));
+        }
+        DB.saveAll(people);
+    }
+
+    private Person findByName(String name) {
+        return new QPerson()
+            .name.eq(name)
+            .findOne();
+    }
+
+    private List<Person> findByNames(String... names) {
+        return new QPerson()
+            .name.in(names)
+            .setUseCache(true)
+            .findList();
+    }
+
+    private Person findById(int id) {
+        return new QPerson()
+            .id.eq(id)
+            .findOne();
+    }
+
+    private List<Person> nameStartsWith(String pattern) {
+        return new QPerson()
+            .name.istartsWith(pattern)
+            .setUseQueryCache(true)
+            .findList();
+    }
+
+    /**
+     * Verifies the Lua CAS: a stored version 2 must NOT be overwritten by an incoming version 1.
+     * Strict greater-than comparison (stored > incoming → skip) ensures stale cluster writes are ignored.
+     */
+    @Test
+    void versionGated_newerCached_staleWriteIsIgnored() throws InterruptedException {
+        ServerCache beanCache = DB.cacheManager().beanCache(RCust.class);
+        beanCache.clear();
+
+        RCust cust = new RCust("stale-test-orig");
+        DB.save(cust);
+        long id = cust.getId();
+
+        // prime cache at version 1
+        DB.find(RCust.class, id);
+        Thread.sleep(150);
+        Object staleV1 = beanCache.get(id);
+        assertThat(staleV1).isNotNull();
+
+        // update to version 2; ensure v2 is in cache
+        cust.setName("stale-test-updated");
+        DB.save(cust);
+        DB.find(RCust.class, id);
+        Thread.sleep(150);
+
+        // stale write attempt: v2 is cached, v1 should be rejected
+        beanCache.put(id, staleV1);
+
+        // v2 must survive (Lua CAS blocked the stale v1 write)
+        Object staleV2 = beanCache.get(id);
+        assertThat(staleV2).isNotNull();
+        assertThat(staleV2).isInstanceOf(CachedBeanData.class);
+        assertThat(((CachedBeanData) staleV2).getVersion()).isEqualTo(2L);
+
+        beanCache.statistics(true);
+        RCust found = DB.find(RCust.class, id);
+        ServerCacheStatistics stats = beanCache.statistics(true);
+        assertThat(stats.getHitCount()).isEqualTo(1);
+        assertThat(found.getVersion()).isEqualTo(2L);
+        assertThat(found.getName()).isEqualTo("stale-test-updated");
+    }
+
+    /**
+     * Verifies that for beans without {@code @Version} the version is always 0.
+     * Equal-version comparison (stored v0 > incoming v0 is false) must never block a write,
+     * so cache updates always go through for unversioned beans.
+     */
+    @Test
+    void zeroVersion_equalVersion_staleWriteIsNotBlocked() throws InterruptedException {
+        ServerCache beanCache = DB.cacheManager().beanCache(TestOne.class);
+        beanCache.clear();
+
+        TestOne t1 = new TestOne("zvw-test", "unique-a");
+        DB.save(t1);
+
+        // prime cache at version 0 (no @Version field)
+        DB.find(TestOne.class, "zvw-test");
+        Thread.sleep(150);
+        Object staleV0 = beanCache.get("zvw-test");
+        assertThat(staleV0).isNotNull();
+
+        // update; ensure new v0 is in cache
+        t1.setOtherUnique("unique-b");
+        DB.save(t1);
+        DB.find(TestOne.class, "zvw-test");
+        Thread.sleep(150);
+
+        // stale write: v0 in cache, incoming v0 — must NOT be blocked (0 > 0 is false)
+        beanCache.put("zvw-test", staleV0);
+
+        // stale data should now be in cache (unlike the versioned case above)
+        beanCache.statistics(true);
+        TestOne found = DB.find(TestOne.class, "zvw-test");
+        ServerCacheStatistics stats = beanCache.statistics(true);
+        assertThat(stats.getHitCount()).isEqualTo(1);
+        assertThat(found.getOtherUnique()).isEqualTo("unique-a");
+    }
+}

ebean-redisson/src/test/java/org/integration/RedissonTenantAwareCacheTest.java

@@ -0,0 +1,116 @@
+package org.integration;
+
+import io.ebean.DB;
+import io.ebean.Database;
+import io.ebean.cache.ServerCache;
+import io.ebean.cache.ServerCacheStatistics;
+import org.domain.RCust;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+class RedissonTenantAwareCacheTest {
+
+  private static final ThreadLocal<String> TENANT = new ThreadLocal<>();
+
+  private Database buildTenantDb() {
+    return Database.builder()
+      .dataSource(DB.getDefault().pluginApi().dataSource())
+      .loadFromProperties()
+      .defaultDatabase(false)
+      .name("tenant-test")
+      .ddlGenerate(false)
+      .ddlRun(false)
+      .currentTenantProvider(TENANT::get)
+      .build();
+  }
+
+  @Test
+  void singleBean_tenantA_hits_tenantB_misses() {
+    Database db = buildTenantDb();
+    try {
+      RCust cust = new RCust("t-single-iso");
+      DB.save(cust);
+      long id = cust.getId();
+
+      ServerCache beanCache = db.cacheManager().beanCache(RCust.class);
+      beanCache.clear();
+
+      // Tenant A: first load – goes to DB
+      TENANT.set("tenantA");
+      beanCache.statistics(true); // reset counters
+      assertThat(db.find(RCust.class, id)).isNotNull();
+      assertThat(Objects.requireNonNull(beanCache.statistics(true)).getMissCount()).isEqualTo(1);
+
+      // Tenant A: second load – must hit cache
+      assertThat(db.find(RCust.class, id)).isNotNull();
+      assertThat(Objects.requireNonNull(beanCache.statistics(true)).getHitCount()).isEqualTo(1);
+
+      // Tenant B: same ID, different tenant key – must miss
+      TENANT.set("tenantB");
+      beanCache.statistics(true); // reset counters
+      assertThat(db.find(RCust.class, id)).isNotNull();
+      ServerCacheStatistics statsB = beanCache.statistics(true);
+      assertNotNull(statsB);
+      assertThat(statsB.getHitCount()).isEqualTo(0);
+      assertThat(statsB.getMissCount()).isEqualTo(1);
+
+    } finally {
+      TENANT.remove();
+      db.shutdown(false, false);
+    }
+  }
+
+  @Test
+  void getAll_tenantA_hits_tenantB_misses() throws InterruptedException {
+    Database db = buildTenantDb();
+    try {
+      List<RCust> custs = new ArrayList<>();
+      for (String n : new String[]{"tga0", "tga1", "tga2"}) {
+        custs.add(new RCust(n));
+      }
+      DB.saveAll(custs);
+      List<Long> ids = custs.stream().map(RCust::getId).collect(Collectors.toList());
+
+      ServerCache beanCache = db.cacheManager().beanCache(RCust.class);
+      beanCache.clear();
+
+      // Tenant A: first batch load – DB misses, cache populated
+      TENANT.set("tenantA");
+      List<RCust> listA0 = db.find(RCust.class).where().idIn(ids).setUseCache(true).findList();
+      assertThat(listA0).hasSize(3);
+
+      Thread.sleep(10);
+
+      // Tenant A: second batch load – all 3 must be cache hits
+      beanCache.statistics(true); // reset
+      List<RCust> listA1 = db.find(RCust.class).where().idIn(ids).setUseCache(true).findList();
+      assertThat(listA1).hasSize(3);
+      ServerCacheStatistics statsA = beanCache.statistics(true);
+      assertNotNull(statsA);
+      assertThat(statsA.getHitCount()).isEqualTo(3);
+      assertThat(statsA.getMissCount()).isEqualTo(0);
+
+      // Tenant B: same IDs, different tenant – all 3 must miss
+      TENANT.set("tenantB");
+      beanCache.statistics(true); // reset
+      List<RCust> listB = db.find(RCust.class).where().idIn(ids).setUseCache(true).findList();
+      assertThat(listB).hasSize(3);
+      ServerCacheStatistics statsB = beanCache.statistics(true);
+      assertNotNull(statsB);
+      assertThat(statsB.getHitCount()).isEqualTo(0);
+      assertThat(statsB.getMissCount()).isEqualTo(3);
+
+    } finally {
+      TENANT.remove();
+      db.shutdown(false, false);
+    }
+  }
+}

ebean-redisson/src/test/resources/application-test.yaml

@@ -0,0 +1,10 @@
+ebean:
+  dumpMetricsOnShutdown: true
+  dumpMetricsOptions: sql,hash,loc
+  test:
+    registerTestTenantProvider: false
+    redis: latest
+#    shutdown: stop # stop | remove
+    platform: h2 # h2, postgres, mysql, oracle, sqlserver, sqlite
+    ddlMode: dropCreate # none | dropCreate | create | migration | createOnly | migrationDropCreate
+    dbName: myapp

ebean-redisson/src/test/resources/logback-test.xml

@@ -0,0 +1,22 @@
+<configuration>
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <root level="ERROR">
+    <appender-ref ref="STDOUT"/>
+  </root>
+
+  <logger name="io.ebean.SQL" level="TRACE"/>
+  <logger name="io.ebean.TXN" level="TRACE"/>
+  <logger name="io.ebean.SUM" level="TRACE"/>
+
+  <logger name="io.ebean.cache" level="TRACE"/>
+  <logger name="io.ebean.cache.QUERY" level="TRACE"/>
+  <logger name="io.ebean.cache.BEAN" level="TRACE"/>
+  <logger name="io.ebean.cache.COLL" level="TRACE"/>
+  <logger name="io.ebean.cache.NATKEY" level="TRACE"/>
+
+</configuration>

ebean-redisson/src/test/resources/redisson-config.yaml

@@ -0,0 +1,17 @@
+singleServerConfig:
+  address: "redis://127.0.0.1:6379"
+  password: null
+  database: 0
+  connectionMinimumIdleSize: 2
+  connectionPoolSize: 10
+  idleConnectionTimeout: 3000
+  connectTimeout: 3000
+  timeout: 3000
+  retryAttempts: 2
+
+threads: 4
+nettyThreads: 8
+referenceEnabled: false
+keepPubSubOrder: false
+checkLockSyncedSlaves: false
+lockWatchdogTimeout: 10000

ebean-test/src/test/java/io/ebean/xtest/event/BeanPersistControllerTest.java

@@ -9,6 +9,9 @@ import io.ebean.event.BeanPersistRequest;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.tests.model.m2m.MnyA;
+import org.tests.model.m2m.MnyB;
+import org.tests.model.m2m.MnyC;
 import org.tests.model.m2m.MnyTopic;
 import org.tests.model.basic.EBasicVer;
 import org.tests.model.basic.UTDetail;
@@ -166,6 +169,56 @@ public class BeanPersistControllerTest {
     db.shutdown();
   }
 
+  @Test
+  public void preDelete_seesManyToManyCollection_beforeCascadeDelete() {
+    List<List<MnyC>> capturedCs = new ArrayList<>();
+    BeanPersistAdapter mnyBAdapter = new BeanPersistAdapter() {
+      @Override
+      public boolean isRegisterFor(Class<?> cls) {
+        return MnyB.class.isAssignableFrom(cls);
+      }
+
+      @Override
+      public boolean preDelete(BeanPersistRequest<?> request) {
+        // the M2M collection should still be intact here, before the
+        // intersection table rows are cascade deleted
+        capturedCs.add(new ArrayList<>(((MnyB) request.bean()).getCs()));
+        return true;
+      }
+    };
+
+    DatabaseBuilder config = Database.builder();
+    config.setName("h2ebasicver");
+    config.setRegister(false);
+    config.setDefaultServer(false);
+    config.loadFromProperties();
+    config.setDdlGenerate(true);
+    config.setDdlRun(true);
+    config.setDdlExtra(false);
+    config.addClass(MnyA.class);
+    config.addClass(MnyB.class);
+    config.addClass(MnyC.class);
+    config.add(mnyBAdapter);
+    Database db = config.build();
+
+    try {
+      MnyB mnyB = new MnyB();
+      MnyC mnyC = new MnyC();
+      mnyB.setCs(new ArrayList<>());
+      mnyB.getCs().add(mnyC);
+      db.save(mnyC);
+      db.save(mnyB);
+
+      MnyB mnyBFromDB = db.find(MnyB.class, mnyB.getId());
+      db.delete(mnyBFromDB);
+
+      assertThat(capturedCs).hasSize(1);
+      assertThat(capturedCs.get(0)).extracting("id").containsExactly(mnyC.getId());
+    } finally {
+      db.shutdown();
+    }
+  }
+
   private Database createDatabase(PersistAdapter persistAdapter) {
     DatabaseBuilder config = Database.builder();
     config.setName("h2ebasicver");

ebean-test/src/test/java/org/tests/model/onetoone/album/DeleteById_SoftDelete_Tests.java

@@ -88,6 +88,35 @@ public class DeleteById_SoftDelete_Tests extends BaseTestCase {
     cover.deletePermanent();
   }
 
+  @Test
+  public void queryByIdDelete_when_softDelete() {
+
+    Cover cover = new Cover("q1");
+    cover.save();
+
+    LoggedSql.start();
+
+    DB.find(Cover.class).setId(cover.getId()).delete();
+
+    List<String> sql = LoggedSql.stop();
+    assertThat(sql).hasSize(1);
+    if (isPlatformBooleanNative()) {
+      assertSql(sql.get(0)).contains("update cover set deleted=true where id = ?");
+    } else {
+      assertSql(sql.get(0)).contains("update cover set deleted=1 where id = ?");
+    }
+
+    assertNull(DB.find(Cover.class, cover.getId()));
+    Cover softDeleted = DB.find(Cover.class)
+      .setIncludeSoftDeletes()
+      .setId(cover.getId())
+      .findOne();
+    assertNotNull(softDeleted);
+    assertThat(softDeleted.isDeleted()).isTrue();
+
+    cover.deletePermanent();
+  }
+
   @Test
   public void deletePermanentById_when_softDelete() {
 

ebean-test/src/test/java/org/tests/rawsql/TestRawSqlWithPlaceholders.java

@@ -120,14 +120,17 @@ class TestRawSqlWithPlaceholders extends BaseTestCase {
       .columnMapping("total_amount", "totalAmount")
       .create();
 
+    // HAVING is evaluated before SELECT-list aliases exist, so on some platforms (e.g. Postgres)
+    // a HAVING clause cannot reference the "total_amount" output alias - it must reference the
+    // underlying aggregate expression instead.
     havingOnlySql = RawSqlBuilder.withPlaceholders(HAVING_ONLY_SQL)
       .columnMapping("order_id", "order.id")
-      .columnMapping("total_amount", "totalAmount")
+      .columnMapping("sum(d.order_qty * d.unit_price)", "totalAmount")
       .create();
 
     whereAndHavingSql = RawSqlBuilder.withPlaceholders(WHERE_AND_HAVING_SQL)
       .columnMapping("order_id", "order.id")
-      .columnMapping("total_amount", "totalAmount")
+      .columnMapping("sum(d.order_qty * d.unit_price)", "totalAmount")
       .create();
 
     whereAndOrderBySql = RawSqlBuilder.withPlaceholders(WHERE_AND_ORDER_BY_SQL)

pom.xml

@@ -94,6 +94,7 @@
     <module>ebean-net-postgis-types</module>
     <module>ebean-pgvector-types</module>
     <module>ebean-redis</module>
+    <module>ebean-redisson</module>
     <module>platforms</module>
     <module>composites</module>
     <module>ebean-jackson-mapper</module>

tests/test-java16/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <artifactId>tests</artifactId>
     <groupId>io.ebean</groupId>
-    <version>18.1.0</version>
+    <version>18.2.0</version>
   </parent>
 
   <artifactId>test-java16</artifactId>
@@ -18,7 +18,7 @@
     <dependency>
       <groupId>io.ebean</groupId>
       <artifactId>ebean-h2</artifactId>
-      <version>18.1.0</version>
+      <version>${project.version}</version>
     </dependency>
 
     <dependency>
@@ -30,14 +30,14 @@
     <dependency>
       <groupId>io.ebean</groupId>
       <artifactId>ebean-test</artifactId>
-      <version>18.1.0</version>
+      <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
 
     <dependency>
       <groupId>io.ebean</groupId>
       <artifactId>querybean-generator</artifactId>
-      <version>18.1.0</version>
+      <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>