Skip to content

IGNITE-20913 Previous value, version, their combination or parts should be written as a part of the DataEntry to allow CacheVersionConflictResolver use them on resolve #250

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anton-vinogradov wants to merge 2 commits into
base: master
Choose a base branch
from
Open

IGNITE-20913 Previous value, version, their combination or parts should be written as a part of the DataEntry to allow CacheVersionConflictResolver use them on resolve #250

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
91f60e7
WIP
anton-vinogradov Dec 21, 2023
7a9a0f4
WIP
anton-vinogradov Aug 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 4 additions & 18 deletions modules/cdc-ext/src/main/java/org/apache/ignite/cdc/CdcEventsIgniteApplier.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,8 @@
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectImpl;
import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.KeyCacheObjectImpl;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrExpirationInfo;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
Expand Down Expand Up @@ -84,28 +82,16 @@ public CdcEventsIgniteApplier(IgniteEx ignite, int maxBatchSize, IgniteLogger lo

/** {@inheritDoc} */
@Override protected KeyCacheObject toKey(CdcEvent evt) {
Object key = evt.key();

if (key instanceof KeyCacheObject)
return (KeyCacheObject)key;
else
return new KeyCacheObjectImpl(key, null, evt.partition());
return evt.keyCacheObject();
}

/** {@inheritDoc} */
@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);
CacheObject val = evt.valueCacheObject();

return evt.expireTime() != EXPIRE_TIME_ETERNAL ?
new GridCacheDrExpirationInfo(cacheObj, ver, TTL_ETERNAL, evt.expireTime()) :
new GridCacheDrInfo(cacheObj, ver);
new GridCacheDrExpirationInfo(val, ver, TTL_ETERNAL, evt.expireTime()) :
new GridCacheDrInfo(val, ver);
}

/** @return Cache. */
Expand Down
49 changes: 46 additions & 3 deletions ...src/main/java/org/apache/ignite/cdc/conflictresolve/CacheVersionConflictResolverImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,16 @@

package org.apache.ignite.cdc.conflictresolve;

import java.util.Objects;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.CacheObjectUtils;
import org.apache.ignite.internal.processors.cache.CacheObjectValueContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.version.CacheVersionConflictResolver;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
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;
Expand Down Expand Up @@ -88,11 +94,12 @@ public CacheVersionConflictResolverImpl(
CacheObjectValueContext ctx,
GridCacheVersionedEntryEx<K, V> oldEntry,
GridCacheVersionedEntryEx<K, V> newEntry,
Object prevStateMeta,
boolean atomicVerComparator
) {
GridCacheVersionConflictContext<K, V> res = new GridCacheVersionConflictContext<>(ctx, oldEntry, newEntry);

boolean useNew = isUseNew(ctx, oldEntry, newEntry);
boolean useNew = isUseNew(ctx, oldEntry, newEntry, prevStateMeta);

if (log.isDebugEnabled())
debugResolve(ctx, useNew, oldEntry, newEntry);
Expand All @@ -117,7 +124,8 @@ public CacheVersionConflictResolverImpl(
protected <K, V> boolean isUseNew(
CacheObjectValueContext ctx,
GridCacheVersionedEntryEx<K, V> oldEntry,
GridCacheVersionedEntryEx<K, V> newEntry
GridCacheVersionedEntryEx<K, V> newEntry,
Object prevStateMeta
) {
if (newEntry.dataCenterId() == clusterId) // Update made on the local cluster always win.
return true;
Expand All @@ -139,8 +147,8 @@ protected <K, V> boolean isUseNew(
}

if (conflictResolveFieldEnabled) {
Object oldVal = oldEntry.value(ctx);
Object newVal = newEntry.value(ctx);
Object oldVal = oldEntry.value(ctx);

if (oldVal != null && newVal != null) {
try {
Expand All @@ -153,6 +161,17 @@ protected <K, V> boolean isUseNew(
);
}
}

Object field = oldVal != null ? value(oldVal) : null;

if (Objects.equals(field, prevStateMeta)) // Previous value synchronized.
return true;
}
else {
GridCacheVersion oldVer = oldEntry.value(ctx) != null ? oldEntry.version() : null; // TODO null value version (entry vs row)

if (Objects.equals(oldVer, prevStateMeta)) // Previous value synchronized.
return true;
}

log.error("Conflict can't be resolved, " + (newEntry.value(ctx) == null ? "remove" : "update") + " ignored " +
Expand All @@ -162,6 +181,30 @@ protected <K, V> boolean isUseNew(
return false;
}

/**
* {@inheritDoc}
*/
@Override public Object previousStateMetadata(GridCacheEntryEx entry) {
if (conflictResolveFieldEnabled) {
CacheObjectValueContext ctx = entry.context().cacheObjectContext();
CacheObject val = entry.rawGet();

return val != null ?
value(CacheObjectUtils.unwrapBinaryIfNeeded(ctx, val, true, true, null)) :
null;
}
else {
try {
GridCacheVersion ver = entry.version();

return ver != null ? ver.conflictVersion() : null;
}
catch (GridCacheEntryRemovedException e) { // TODO
throw new RuntimeException(e);
}
}
}

/** @return Conflict resolve field value. */
protected Comparable value(Object val) {
return (val instanceof BinaryObject)
Expand Down
Loading