IGNITE-17449 TTL support for CDC

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCachePutAllConflictRequest.java

@@ -33,6 +33,8 @@
 import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.U;
 
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.EXPIRE_TIME_CALCULATE;
+import static org.apache.ignite.internal.processors.cache.GridCacheUtils.TTL_NOT_CHANGED;
 import static org.apache.ignite.internal.processors.platform.utils.PlatformUtils.readCacheObject;
 
 /**
@@ -51,6 +53,8 @@ public class ClientCachePutAllConflictRequest extends ClientCacheDataRequest imp
     public ClientCachePutAllConflictRequest(BinaryReaderExImpl reader, ClientConnectionContext ctx) {
         super(reader);
 
+        boolean expPlc = cachex(ctx).configuration().getExpiryPolicyFactory() != null;
+
         int cnt = reader.readInt();
 
         map = new LinkedHashMap<>(cnt);
@@ -59,11 +63,19 @@ public ClientCachePutAllConflictRequest(BinaryReaderExImpl reader, ClientConnect
             KeyCacheObject key = readCacheObject(reader, true);
             CacheObject val = readCacheObject(reader, false);
             GridCacheVersion ver = (GridCacheVersion)reader.readObjectDetached();
-            Long expireTime = reader.readLong();
+            long expireTime = reader.readLong();
+            long ttl = expireTime == CU.EXPIRE_TIME_ETERNAL
+                ? CU.TTL_ETERNAL
+                : (expireTime - U.currentTimeMillis());
+
+            if (ttl < 0) // Happens if replication lag more than TTL.
+                ttl = CU.TTL_ZERO;
 
             GridCacheDrInfo info = expireTime != CU.EXPIRE_TIME_ETERNAL ?
-                new GridCacheDrExpirationInfo(val, ver, CU.TTL_ETERNAL, expireTime) : //TODO: check params here.
-                new GridCacheDrInfo(val, ver);
+                new GridCacheDrExpirationInfo(val, ver, ttl, expireTime) :
+                (expPlc
+                    ? new GridCacheDrExpirationInfo(val, ver, TTL_NOT_CHANGED, EXPIRE_TIME_CALCULATE)
+                    : new GridCacheDrInfo(val, ver));
 
             map.put(key, info);
         }

modules/core/src/test/java/org/apache/ignite/internal/client/thin/DataReplicationOperationsTest.java

@@ -24,6 +24,7 @@
 import org.apache.ignite.client.IgniteClient;
 import org.apache.ignite.client.Person;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.platform.cache.expiry.PlatformExpiryPolicy;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T3;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -42,6 +43,9 @@ public class DataReplicationOperationsTest extends AbstractThinClientTest {
     /** Keys count. */
     private static final int KEYS_CNT = 10;
 
+    /** TTL. */
+    public static final int TTL = 1000;
+
     /** */
     private static IgniteClient client;
 
@@ -121,39 +125,57 @@ public void testRemoveAllConflict() {
     /** @throws Exception If fails. */
     @Test
     public void testWithExpiryPolicy() throws Exception {
-        long expireTime = System.currentTimeMillis() + 1000;
+        PlatformExpiryPolicy expPlc = new PlatformExpiryPolicy(TTL, TTL, TTL);
 
-        ClientCacheConfiguration ccfgWithTtlEntries = new ClientCacheConfiguration()
-            .setName("cache-with-ttl-entries");
+        ClientCacheConfiguration ccfgWithExpPlc = new ClientCacheConfiguration()
+            .setName("cache-with-expiry-policy")
+            .setExpiryPolicy(expPlc);
 
-        TcpClientCache<Object, Object> cache =
-            (TcpClientCache<Object, Object>)client.getOrCreateCache(ccfgWithTtlEntries);
+        TcpClientCache<Object, Object> cache = (TcpClientCache<Object, Object>)client.getOrCreateCache(ccfgWithExpPlc);
 
-        TcpClientCache<Object, Object> cacheWithTtlEntries = binary ?
+        TcpClientCache<Object, Object> cacheWithExpPlc = binary ?
             (TcpClientCache<Object, Object>)cache.withKeepBinary() : cache;
 
-        Map<Object, T3<Object, GridCacheVersion, Long>> data = createPutAllData(expireTime);
+        Map<Object, T3<Object, GridCacheVersion, Long>> data = createPutAllData(CU.EXPIRE_TIME_ETERNAL);
 
-        cacheWithTtlEntries.putAllConflict(data);
+        cacheWithExpPlc.putAllConflict(data);
 
-        assertTrue(cacheWithTtlEntries.containsKeys(data.keySet()));
+        assertTrue(cacheWithExpPlc.containsKeys(data.keySet()));
 
         assertTrue(waitForCondition(
-            () -> data.keySet().stream().noneMatch(cacheWithTtlEntries::containsKey),
-            2 * expireTime
+            () -> data.keySet().stream().noneMatch(cacheWithExpPlc::containsKey),
+            2 * expPlc.getExpiryForCreation().getDurationAmount()
         ));
     }
 
+    /** @throws Exception If fails. */
+    @Test
+    public void testWithPerEntryExpiry() throws Exception {
+        TcpClientCache<Object, Object> cache0 =
+            (TcpClientCache<Object, Object>)client.getOrCreateCache(DEFAULT_CACHE_NAME);
+
+        TcpClientCache<Object, Object> cache = binary ?
+            (TcpClientCache<Object, Object>)cache0.withKeepBinary() : cache0;
+
+        Map<Object, T3<Object, GridCacheVersion, Long>> data = createPutAllData(System.currentTimeMillis() + TTL);
+
+        cache.putAllConflict(data);
+
+        assertTrue(cache.containsKeys(data.keySet()));
+
+        assertTrue(waitForCondition(() -> data.keySet().stream().noneMatch(cache::containsKey), 2 * TTL));
+    }
+
     /** */
-    private Map<Object, T3<Object, GridCacheVersion, Long>> createPutAllData(long ttl) {
+    private Map<Object, T3<Object, GridCacheVersion, Long>> createPutAllData(long expireTime) {
         Map<Object, T3<Object, GridCacheVersion, Long>> map = new HashMap<>();
 
         for (int i = 0; i < KEYS_CNT; i++) {
             Person key = new Person(i, "Person-" + i);
             Person val = new Person(i, "Person-" + i);
 
             map.put(binary ? client.binary().toBinary(key) : key,
-                new T3<>(binary ? client.binary().toBinary(val) : val, otherVer, ttl));
+                new T3<>(binary ? client.binary().toBinary(val) : val, otherVer, expireTime));
         }
 
         return map;