From 42a6275562ac163f0b4f90a7ad64728ebf07b88a Mon Sep 17 00:00:00 2001 From: nizhikov Date: Thu, 9 Mar 2023 15:51:10 +0300 Subject: [PATCH 1/6] IGNITE-17749 TTL support for CDC --- .../org/apache/ignite/cdc/AbstractCdcEventsApplier.java | 7 +++---- .../org/apache/ignite/cdc/CdcEventsIgniteApplier.java | 8 +++++--- .../ignite/cdc/thin/CdcEventsIgniteClientApplier.java | 9 +++++---- 3 files changed, 13 insertions(+), 11 deletions(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java index ffef91114..ef994f266 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java @@ -87,7 +87,7 @@ public int apply(Iterable evts) throws IgniteCheckedException { if (evt.value() != null) { evtsApplied += applyIf(currCacheId, () -> isApplyBatch(updBatch, key), hasRemoves); - updBatch.put(key, toValue(currCacheId, evt.value(), ver)); + updBatch.put(key, toValue(currCacheId, evt, ver)); } else { evtsApplied += applyIf(currCacheId, hasUpdates, () -> isApplyBatch(rmvBatch, key)); @@ -109,13 +109,12 @@ public int apply(Iterable evts) throws IgniteCheckedException { * @param applyUpd Apply update batch flag supplier. * @param applyRmv Apply remove batch flag supplier. * @return Number of applied events. - * @throws IgniteCheckedException In case of error. */ private int applyIf( int cacheId, BooleanSupplier applyUpd, BooleanSupplier applyRmv - ) throws IgniteCheckedException { + ) { int evtsApplied = 0; if (applyUpd.getAsBoolean()) { @@ -152,7 +151,7 @@ private boolean isApplyBatch(Map map, K key) { protected abstract K toKey(CdcEvent evt); /** @return Value. */ - protected abstract V toValue(int cacheId, Object val, GridCacheVersion ver); + protected abstract V toValue(int cacheId, CdcEvent evt, GridCacheVersion ver); /** Stores DR data. */ protected abstract void putAllConflict(int cacheId, Map drMap); diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java index bce50d38d..b036d2bf5 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java @@ -93,16 +93,18 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo } /** {@inheritDoc} */ - @Override protected GridCacheDrInfo toValue(int cacheId, Object val, GridCacheVersion ver) { + @Override protected GridCacheDrInfo toValue(int cacheId, CdcEvent evt, GridCacheVersion ver) { CacheObject cacheObj; + Object val = evt.value(); + if (val instanceof CacheObject) cacheObj = (CacheObject)val; else cacheObj = new CacheObjectImpl(val, null); - return cache(cacheId).configuration().getExpiryPolicyFactory() != null ? - new GridCacheDrExpirationInfo(cacheObj, ver, TTL_NOT_CHANGED, EXPIRE_TIME_CALCULATE) : + return evt.ttl() != CU.TTL_ETERNAL ? + new GridCacheDrExpirationInfo(cacheObj, ver, evt.ttl(), EXPIRE_TIME_CALCULATE) : new GridCacheDrInfo(cacheObj, ver); } diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java index 5484277a7..d4238be38 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java @@ -27,6 +27,7 @@ import org.apache.ignite.internal.util.collection.IntHashMap; import org.apache.ignite.internal.util.collection.IntMap; import org.apache.ignite.internal.util.typedef.T2; +import org.apache.ignite.internal.util.typedef.T3; import org.apache.ignite.internal.util.typedef.internal.CU; /** @@ -35,7 +36,7 @@ * @see TcpClientCache#putAllConflict(Map) * @see TcpClientCache#removeAllConflict(Map) */ -public class CdcEventsIgniteClientApplier extends AbstractCdcEventsApplier> { +public class CdcEventsIgniteClientApplier extends AbstractCdcEventsApplier> { /** Client connected to the destination cluster. */ private final IgniteClient client; @@ -59,12 +60,12 @@ public CdcEventsIgniteClientApplier(IgniteClient client, int maxBatchSize, Ignit } /** {@inheritDoc} */ - @Override protected T2 toValue(int cacheId, Object val, GridCacheVersion ver) { - return new T2<>(val, ver); + @Override protected T3 toValue(int cacheId, CdcEvent evt, GridCacheVersion ver) { + return new T3<>(evt.value(), ver, evt.ttl()); } /** {@inheritDoc} */ - @Override protected void putAllConflict(int cacheId, Map> drMap) { + @Override protected void putAllConflict(int cacheId, Map> drMap) { cache(cacheId).putAllConflict(drMap); } From 2d6d7135d18b6c12b70e6d9ae3d9f10aa37fdb35 Mon Sep 17 00:00:00 2001 From: nizhikov Date: Fri, 10 Mar 2023 11:17:15 +0300 Subject: [PATCH 2/6] IGNITE-17749 TTL support for CDC --- .../java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java | 4 ++-- .../apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java | 3 +-- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java index b036d2bf5..66c269b15 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java @@ -103,8 +103,8 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo else cacheObj = new CacheObjectImpl(val, null); - return evt.ttl() != CU.TTL_ETERNAL ? - new GridCacheDrExpirationInfo(cacheObj, ver, evt.ttl(), EXPIRE_TIME_CALCULATE) : + return evt.expireTime() != CU.EXPIRE_TIME_ETERNAL ? + new GridCacheDrExpirationInfo(cacheObj, ver, CU.TTL_ETERNAL, evt.expireTime()) : new GridCacheDrInfo(cacheObj, ver); } diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java index d4238be38..2a9912131 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/thin/CdcEventsIgniteClientApplier.java @@ -26,7 +26,6 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.util.collection.IntHashMap; import org.apache.ignite.internal.util.collection.IntMap; -import org.apache.ignite.internal.util.typedef.T2; import org.apache.ignite.internal.util.typedef.T3; import org.apache.ignite.internal.util.typedef.internal.CU; @@ -61,7 +60,7 @@ public CdcEventsIgniteClientApplier(IgniteClient client, int maxBatchSize, Ignit /** {@inheritDoc} */ @Override protected T3 toValue(int cacheId, CdcEvent evt, GridCacheVersion ver) { - return new T3<>(evt.value(), ver, evt.ttl()); + return new T3<>(evt.value(), ver, evt.expireTime()); } /** {@inheritDoc} */ From 83b9d60e89c722dfea0775ed05883be48674aba4 Mon Sep 17 00:00:00 2001 From: nizhikov Date: Tue, 14 Mar 2023 11:05:52 +0300 Subject: [PATCH 3/6] IGNITE-17749 Expire support for CDC --- .../org/apache/ignite/cdc/CdcEventsIgniteApplier.java | 8 +++++--- .../org/apache/ignite/cdc/AbstractReplicationTest.java | 10 ++++++++-- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java index 66c269b15..fad592436 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java @@ -17,6 +17,7 @@ package org.apache.ignite.cdc; +import java.util.Date; import java.util.Map; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteLogger; @@ -35,9 +36,6 @@ 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; - /** * Contains logic to process {@link CdcEvent} and apply them to the destination cluster. * @@ -103,6 +101,10 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo else cacheObj = new CacheObjectImpl(val, null); + //TODO: check how to minimize test partitions count to reduce test duration. + System.out.println("CdcEventsIgniteApplier.toValue - " + evt.expireTime() + ", date = " + new Date(evt.expireTime())); + System.out.println("(System.currentTimeMillis() - evt.expireTime()) = " + (evt.expireTime() - System.currentTimeMillis())); + return evt.expireTime() != CU.EXPIRE_TIME_ETERNAL ? new GridCacheDrExpirationInfo(cacheObj, ver, CU.TTL_ETERNAL, evt.expireTime()) : new GridCacheDrInfo(cacheObj, ver); diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java index 144b8b872..838fff4c7 100644 --- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java +++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java @@ -446,7 +446,7 @@ public void testActiveActiveReplication() throws Exception { /** Test that destination cluster applies expiration policy on received entries. */ @Test public void testWithExpiryPolicy() throws Exception { - Factory factory = () -> new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 10)); + Factory factory = () -> new CreatedExpiryPolicy(new Duration(TimeUnit.MINUTES, 2)); IgniteCache srcCache = createCache(srcCluster[0], ACTIVE_PASSIVE_CACHE, factory); IgniteCache destCache = createCache(destCluster[0], ACTIVE_PASSIVE_CACHE, factory); @@ -465,7 +465,13 @@ public void testWithExpiryPolicy() throws Exception { assertTrue(waitForCondition(() -> !srcCache.containsKey(0), getTestTimeout())); log.warning(">>>>>> Waiting for removing in destination cache"); - assertTrue(waitForCondition(() -> !destCache.containsKey(0), 20_000)); + + Duration ttl = factory.create().getExpiryForCreation(); + + assertTrue(waitForCondition( + () -> !destCache.containsKey(0), + 2 * TimeUnit.MILLISECONDS.convert(ttl.getDurationAmount(), ttl.getTimeUnit()) + )); } finally { for (IgniteInternalFuture fut : futs) From f48aa6ab1847a0b1ab191853b9c4a77f9ccbc264 Mon Sep 17 00:00:00 2001 From: nizhikov Date: Wed, 15 Mar 2023 16:51:59 +0300 Subject: [PATCH 4/6] IGNITE-17449 Support expireTime in replication --- .../ignite/cdc/AbstractCdcEventsApplier.java | 3 ++- .../ignite/cdc/CdcEventsIgniteApplier.java | 5 ---- .../CacheVersionConflictResolverImpl.java | 27 ++++++++++++++++++- ...DebugCacheVersionConflictResolverImpl.java | 2 ++ .../ignite/cdc/AbstractReplicationTest.java | 2 +- 5 files changed, 31 insertions(+), 8 deletions(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java index ef994f266..479a1db4a 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractCdcEventsApplier.java @@ -109,12 +109,13 @@ public int apply(Iterable evts) throws IgniteCheckedException { * @param applyUpd Apply update batch flag supplier. * @param applyRmv Apply remove batch flag supplier. * @return Number of applied events. + * @throws IgniteCheckedException In case of error. */ private int applyIf( int cacheId, BooleanSupplier applyUpd, BooleanSupplier applyRmv - ) { + ) throws IgniteCheckedException { int evtsApplied = 0; if (applyUpd.getAsBoolean()) { diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java index fad592436..63991d443 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java @@ -17,7 +17,6 @@ package org.apache.ignite.cdc; -import java.util.Date; import java.util.Map; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteLogger; @@ -101,10 +100,6 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo else cacheObj = new CacheObjectImpl(val, null); - //TODO: check how to minimize test partitions count to reduce test duration. - System.out.println("CdcEventsIgniteApplier.toValue - " + evt.expireTime() + ", date = " + new Date(evt.expireTime())); - System.out.println("(System.currentTimeMillis() - evt.expireTime()) = " + (evt.expireTime() - System.currentTimeMillis())); - return evt.expireTime() != CU.EXPIRE_TIME_ETERNAL ? new GridCacheDrExpirationInfo(cacheObj, ver, CU.TTL_ETERNAL, evt.expireTime()) : new GridCacheDrInfo(cacheObj, ver); diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java index 56e764fd5..292f1361e 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java @@ -24,6 +24,7 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext; import org.apache.ignite.internal.processors.cache.version.GridCacheVersionedEntryEx; import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.internal.CU; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; @@ -88,7 +89,31 @@ public CacheVersionConflictResolverImpl(byte clusterId, String conflictResolveFi ) { GridCacheVersionConflictContext res = new GridCacheVersionConflictContext<>(ctx, oldEntry, newEntry); - if (isUseNew(ctx, oldEntry, newEntry)) + boolean expireExists = oldEntry.ttl() != CU.TTL_ETERNAL + || newEntry.ttl() != CU.TTL_ETERNAL + || oldEntry.expireTime() != CU.EXPIRE_TIME_ETERNAL + || newEntry.expireTime() != CU.EXPIRE_TIME_ETERNAL; + + boolean useNew = isUseNew(ctx, oldEntry, newEntry); + + if (expireExists) { + if (newEntry.expireTime() > oldEntry.expireTime()) { + res.merge( + useNew ? newEntry.value(ctx) : oldEntry.value(ctx), + newEntry.ttl(), + newEntry.expireTime() + ); + } + else { + res.merge( + useNew ? newEntry.value(ctx) : oldEntry.value(ctx), + oldEntry.ttl(), + oldEntry.expireTime() + ); + } + + } + else if (useNew) res.useNew(); else res.useOld(); diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/DebugCacheVersionConflictResolverImpl.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/DebugCacheVersionConflictResolverImpl.java index cb3ce3210..000cbf378 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/DebugCacheVersionConflictResolverImpl.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/DebugCacheVersionConflictResolverImpl.java @@ -54,6 +54,8 @@ public DebugCacheVersionConflictResolverImpl(byte clusterId, String conflictReso "start=" + oldEntry.isStartVersion() + ", oldVer=" + oldEntry.version() + ", newVer=" + newEntry.version() + + ", oldExpire=[" + oldEntry.ttl() + "," + oldEntry.expireTime() + ']' + + ", newExpire=[" + newEntry.ttl() + "," + newEntry.expireTime() + ']' + ", old=" + oldVal + ", new=" + newVal + ", res=" + res + ']'); diff --git a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java index 838fff4c7..02d394d84 100644 --- a/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java +++ b/modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java @@ -446,7 +446,7 @@ public void testActiveActiveReplication() throws Exception { /** Test that destination cluster applies expiration policy on received entries. */ @Test public void testWithExpiryPolicy() throws Exception { - Factory factory = () -> new CreatedExpiryPolicy(new Duration(TimeUnit.MINUTES, 2)); + Factory factory = () -> new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 30)); IgniteCache srcCache = createCache(srcCluster[0], ACTIVE_PASSIVE_CACHE, factory); IgniteCache destCache = createCache(destCluster[0], ACTIVE_PASSIVE_CACHE, factory); From 400ac870237dcaf5375fd98eaa6d285eeeaf26cb Mon Sep 17 00:00:00 2001 From: nizhikov Date: Wed, 15 Mar 2023 17:07:46 +0300 Subject: [PATCH 5/6] IGNITE-17449 Support expireTime in replication --- .../java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java index 63991d443..9afda275b 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java @@ -35,6 +35,9 @@ 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_ETERNAL; +import static org.apache.ignite.internal.processors.cache.GridCacheUtils.TTL_ETERNAL; + /** * Contains logic to process {@link CdcEvent} and apply them to the destination cluster. * @@ -100,8 +103,8 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo else cacheObj = new CacheObjectImpl(val, null); - return evt.expireTime() != CU.EXPIRE_TIME_ETERNAL ? - new GridCacheDrExpirationInfo(cacheObj, ver, CU.TTL_ETERNAL, evt.expireTime()) : + return evt.expireTime() != EXPIRE_TIME_ETERNAL ? + new GridCacheDrExpirationInfo(cacheObj, ver, TTL_ETERNAL, evt.expireTime()) : new GridCacheDrInfo(cacheObj, ver); } From fcfbd658c7b31ffe2e00b1a7c70ab07113dcfc67 Mon Sep 17 00:00:00 2001 From: nizhikov Date: Wed, 15 Mar 2023 19:29:03 +0300 Subject: [PATCH 6/6] IGNITE-17449 Support expireTime in replication --- .../cdc/conflictresolve/CacheVersionConflictResolverImpl.java | 1 - 1 file changed, 1 deletion(-) diff --git a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java index 292f1361e..ce1a17fce 100644 --- a/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java +++ b/modules/cdc-ext/src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java @@ -111,7 +111,6 @@ public CacheVersionConflictResolverImpl(byte clusterId, String conflictResolveFi oldEntry.expireTime() ); } - } else if (useNew) res.useNew();