Repository: incubator-ignite Updated Branches: refs/heads/ignite-283 [created] eee156a0f
# ignite-283: Refactoring started. Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/eee156a0 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/eee156a0 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/eee156a0 Branch: refs/heads/ignite-283 Commit: eee156a0f7ba1dfa2ae8982880024ecee3eee23e Parents: 862fac9 Author: vozerov-gridgain <voze...@gridgain.com> Authored: Tue Feb 17 13:59:04 2015 +0300 Committer: vozerov-gridgain <voze...@gridgain.com> Committed: Tue Feb 17 13:59:04 2015 +0300 ---------------------------------------------------------------------- .../processors/cache/GridCacheContext.java | 10 +++ .../GridDistributedTxRemoteAdapter.java | 20 +++--- .../distributed/dht/GridDhtTxLocalAdapter.java | 5 +- .../distributed/dht/GridDhtTxPrepareFuture.java | 14 ++-- .../near/GridNearTxPrepareFuture.java | 5 +- .../cache/transactions/IgniteTxAdapter.java | 5 +- .../cache/transactions/IgniteTxEntry.java | 76 +++++++++++--------- .../transactions/IgniteTxLocalAdapter.java | 70 ++++++------------ 8 files changed, 98 insertions(+), 107 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java index e5cc9c3..02624d7 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java @@ -338,6 +338,16 @@ public class GridCacheContext<K, V> implements Externalizable { } /** + * @param txEntry TX entry. + * @return Expiry policy for the given TX entry. + */ + @Nullable public ExpiryPolicy expiryForTxEntry(IgniteTxEntry txEntry) { + ExpiryPolicy plc = txEntry.expiry(); + + return plc != null ? plc : expiryPlc; + } + + /** * @param mgr Manager to add. * @return Added manager. */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java index 0513cc7..4f8357a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java @@ -39,6 +39,7 @@ import java.util.*; import java.util.concurrent.atomic.*; import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*; +import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry.*; import static org.apache.ignite.internal.processors.dr.GridDrType.*; import static org.apache.ignite.transactions.TransactionState.*; @@ -345,9 +346,9 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V> entry.explicitVersion(e.explicitVersion()); entry.groupLockEntry(e.groupLockEntry()); - // DR stuff. - entry.drVersion(e.drVersion()); - entry.drExpireTime(e.drExpireTime()); + // Conflict resolution stuff. + entry.conflictVersion(e.conflictVersion()); + entry.conflictExpireTime(e.conflictExpireTime()); } addExplicit(e); @@ -504,19 +505,18 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V> V val = res.get2(); byte[] valBytes = res.get3(); - GridCacheVersion explicitVer = txEntry.drVersion(); + GridCacheVersion explicitVer = txEntry.conflictVersion(); if (txEntry.ttl() == CU.TTL_ZERO) op = DELETE; - boolean drNeedResolve = cacheCtx.conflictNeedResolve(cached.version(), explicitVer); if (drNeedResolve) { IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContext<K, V>> drRes = conflictResolve(op, txEntry.key(), val, valBytes, - txEntry.ttl(), txEntry.drExpireTime(), explicitVer, cached); + txEntry.ttl(), txEntry.conflictExpireTime(), explicitVer, cached); assert drRes != null; @@ -528,9 +528,9 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V> txEntry.ttl(drCtx.ttl()); if (drCtx.newEntry().dataCenterId() != cacheCtx.dataCenterId()) - txEntry.drExpireTime(drCtx.expireTime()); + txEntry.conflictExpireTime(drCtx.expireTime()); else - txEntry.drExpireTime(-1L); + txEntry.conflictExpireTime(CONFLICT_EXPIRE_TIME_NOT_SET); } else if (drCtx.isMerge()) { op = drRes.get1(); @@ -539,7 +539,7 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V> explicitVer = writeVersion(); txEntry.ttl(drCtx.ttl()); - txEntry.drExpireTime(-1L); + txEntry.conflictExpireTime(CONFLICT_EXPIRE_TIME_NOT_SET); } } else @@ -556,7 +556,7 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V> else { cached.innerSet(this, eventNodeId(), nodeId, val, valBytes, false, false, txEntry.ttl(), true, true, topVer, txEntry.filters(), - replicate ? DR_BACKUP : DR_NONE, txEntry.drExpireTime(), + replicate ? DR_BACKUP : DR_NONE, txEntry.conflictExpireTime(), near() ? null : explicitVer, CU.subjectId(this, cctx), resolveTaskName()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java index aab5680..1c71f12 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java @@ -452,8 +452,9 @@ public abstract class GridDhtTxLocalAdapter<K, V> extends IgniteTxLocalAdapter<K entry.ttl(e.ttl()); entry.filters(e.filters()); entry.expiry(e.expiry()); - entry.drExpireTime(e.drExpireTime()); - entry.drVersion(e.drVersion()); + + entry.conflictExpireTime(e.conflictExpireTime()); + entry.conflictVersion(e.conflictVersion()); } else { entry = e; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java index c6e91a3..a9d898b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java @@ -42,6 +42,7 @@ import java.util.*; import java.util.concurrent.atomic.*; import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*; +import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry.*; import static org.apache.ignite.transactions.TransactionState.*; import static org.apache.ignite.events.EventType.*; import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*; @@ -279,13 +280,11 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu GridCacheEntryEx<K, V> cached = txEntry.cached(); - ExpiryPolicy expiry = txEntry.expiry(); - - if (expiry == null) - expiry = cacheCtx.expiry(); + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); try { - if (txEntry.op() == CREATE || txEntry.op() == UPDATE && txEntry.drExpireTime() == -1L) { + if ((txEntry.op() == CREATE || txEntry.op() == UPDATE) && + txEntry.conflictExpireTime() == CONFLICT_EXPIRE_TIME_NOT_SET) { if (expiry != null) { Duration duration = cached.hasValue() ? expiry.getExpiryForUpdate() : expiry.getExpiryForCreation(); @@ -994,10 +993,7 @@ public final class GridDhtTxPrepareFuture<K, V> extends GridCompoundIdentityFutu GridDhtCacheAdapter<K, V> dht = cacheCtx.isNear() ? cacheCtx.near().dht() : cacheCtx.dht(); - ExpiryPolicy expiry = entry.expiry(); - - if (expiry == null) - expiry = cacheCtx.expiry(); + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(entry); if (expiry != null && entry.op() == READ) { entry.op(NOOP); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java index a94ffba..18fda47 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java @@ -983,10 +983,9 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut txEntry.op(NOOP); - ExpiryPolicy expiry = txEntry.expiry(); + assert txEntry.context() != null; - if (expiry == null) - expiry = txEntry.context().expiry(); + ExpiryPolicy expiry = txEntry.context().expiryForTxEntry(txEntry); if (expiry != null) txEntry.ttl(CU.toTtl(expiry.getExpiryForAccess())); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java index 62d4e3b..ab0b2a5 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java @@ -1244,10 +1244,7 @@ public abstract class IgniteTxAdapter<K, V> extends GridMetadataAwareAdapter GridCacheOperation op = modified ? (val == null ? DELETE : UPDATE) : NOOP; if (op == NOOP) { - ExpiryPolicy expiry = txEntry.expiry(); - - if (expiry == null) - expiry = cacheCtx.expiry(); + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); if (expiry != null) { long ttl = CU.toTtl(expiry.getExpiryForAccess()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java index eac07be..7022c4c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxEntry.java @@ -44,6 +44,9 @@ import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*; * equality. */ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, OptimizedMarshallable { + /** Special value indicating that conflict expire time is not set. */ + public static final long CONFLICT_EXPIRE_TIME_NOT_SET = -1; + /** */ private static final long serialVersionUID = 0L; @@ -90,8 +93,11 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, /** Time to live. */ private long ttl; - /** DR expire time (explicit) */ - private long drExpireTime = -1L; + /** Conflict expire time (explicit) */ + private long conflictExpireTime = CONFLICT_EXPIRE_TIME_NOT_SET; + + /** Conflict version. */ + private GridCacheVersion conflictVer; /** Explicit lock version if there is one. */ @GridToStringInclude @@ -135,9 +141,6 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, /** Deployment enabled flag. */ private boolean depEnabled; - /** Data center replication version. */ - private GridCacheVersion drVer; - /** Expiry policy. */ private ExpiryPolicy expiryPlc; @@ -159,18 +162,18 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, * @param op Operation. * @param val Value. * @param ttl Time to live. - * @param drExpireTime DR expire time. + * @param conflictExpireTime DR expire time. * @param entry Cache entry. - * @param drVer Data center replication version. + * @param conflictVer Data center replication version. */ public IgniteTxEntry(GridCacheContext<K, V> ctx, IgniteInternalTx<K, V> tx, GridCacheOperation op, V val, long ttl, - long drExpireTime, + long conflictExpireTime, GridCacheEntryEx<K, V> entry, - @Nullable GridCacheVersion drVer) { + @Nullable GridCacheVersion conflictVer) { assert ctx != null; assert tx != null; assert op != null; @@ -181,8 +184,8 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, this.val.value(op, val, false, false); this.entry = entry; this.ttl = ttl; - this.drExpireTime = drExpireTime; - this.drVer = drVer; + this.conflictExpireTime = conflictExpireTime; + this.conflictVer = conflictVer; key = entry.key(); keyBytes = entry.keyBytes(); @@ -204,7 +207,7 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, * @param ttl Time to live. * @param entry Cache entry. * @param filters Put filters. - * @param drVer Data center replication version. + * @param conflictVer Data center replication version. */ public IgniteTxEntry(GridCacheContext<K, V> ctx, IgniteInternalTx<K, V> tx, @@ -215,7 +218,7 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, long ttl, GridCacheEntryEx<K, V> entry, IgnitePredicate<Cache.Entry<K, V>>[] filters, - GridCacheVersion drVer) { + GridCacheVersion conflictVer) { assert ctx != null; assert tx != null; assert op != null; @@ -227,7 +230,7 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, this.entry = entry; this.ttl = ttl; this.filters = filters; - this.drVer = drVer; + this.conflictVer = conflictVer; if (entryProcessor != null) addEntryProcessor(entryProcessor, invokeArgs); @@ -296,11 +299,11 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, cp.val.valueBytes(val.valueBytes()); cp.entryProcessorsCol = entryProcessorsCol; cp.ttl = ttl; - cp.drExpireTime = drExpireTime; + cp.conflictExpireTime = conflictExpireTime; cp.explicitVer = explicitVer; cp.grpLock = grpLock; cp.depEnabled = depEnabled; - cp.drVer = drVer; + cp.conflictVer = conflictVer; cp.expiryPlc = expiryPlc; return cp; @@ -568,17 +571,17 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, } /** - * @return DR expire time. + * @return Conflict expire time. */ - public long drExpireTime() { - return drExpireTime; + public long conflictExpireTime() { + return conflictExpireTime; } /** - * @param drExpireTime DR expire time. + * @param conflictExpireTime Conflict expire time. */ - public void drExpireTime(long drExpireTime) { - this.drExpireTime = drExpireTime; + public void conflictExpireTime(long conflictExpireTime) { + this.conflictExpireTime = conflictExpireTime; } /** @@ -692,17 +695,17 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, } /** - * @return DR version. + * @return Conflict version. */ - @Nullable public GridCacheVersion drVersion() { - return drVer; + @Nullable public GridCacheVersion conflictVersion() { + return conflictVer; } /** - * @param drVer DR version. + * @param conflictVer Conflict version. */ - public void drVersion(@Nullable GridCacheVersion drVer) { - this.drVer = drVer; + public void conflictVersion(@Nullable GridCacheVersion conflictVer) { + this.conflictVer = conflictVer; } /** @@ -849,11 +852,18 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, val.writeTo(out); out.writeLong(ttl); - out.writeLong(drExpireTime); CU.writeVersion(out, explicitVer); out.writeBoolean(grpLock); - CU.writeVersion(out, drVer); + + if (conflictExpireTime != CONFLICT_EXPIRE_TIME_NOT_SET) { + out.writeBoolean(true); + out.writeLong(conflictExpireTime); + } + else + out.writeBoolean(false); + + CU.writeVersion(out, conflictVer); out.writeObject(transferExpiryPlc ? new IgniteExternalizableExpiryPolicy(expiryPlc) : null); } @@ -879,11 +889,12 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, val.readFrom(in); ttl = in.readLong(); - drExpireTime = in.readLong(); explicitVer = CU.readVersion(in); grpLock = in.readBoolean(); - drVer = CU.readVersion(in); + + conflictExpireTime = in.readBoolean() ? in.readLong() : CONFLICT_EXPIRE_TIME_NOT_SET; + conflictVer = CU.readVersion(in); expiryPlc = (ExpiryPolicy)in.readObject(); } @@ -1103,6 +1114,7 @@ public class IgniteTxEntry<K, V> implements GridPeerDeployAware, Externalizable, * @throws IOException If failed. * @throws ClassNotFoundException If failed. */ + @SuppressWarnings("unchecked") public void readFrom(ObjectInput in) throws IOException, ClassNotFoundException { hasWriteVal = in.readBoolean(); valBytesSent = in.readBoolean(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/eee156a0/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java index 8ffe224..3a63236 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java @@ -48,6 +48,7 @@ import static org.apache.ignite.events.EventType.*; import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*; import static org.apache.ignite.internal.processors.dr.GridDrType.*; import static org.apache.ignite.transactions.TransactionState.*; +import static org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry.*; /** * Transaction adapter for cache transactions. @@ -724,11 +725,9 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> if (cacheCtx.isNear()) { ((GridNearCacheEntry<K, V>)cached).recordDhtVersion(txEntry.dhtVersion()); - if (txEntry.op() == CREATE || txEntry.op() == UPDATE && txEntry.drExpireTime() == -1L) { - ExpiryPolicy expiry = txEntry.expiry(); - - if (expiry == null) - expiry = cacheCtx.expiry(); + if ((txEntry.op() == CREATE || txEntry.op() == UPDATE) && + txEntry.conflictExpireTime() == CONFLICT_EXPIRE_TIME_NOT_SET) { + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); if (expiry != null) { Duration duration = cached.hasValue() ? @@ -744,14 +743,12 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> byte[] valBytes = res.get3(); // Deal with DR conflicts. - GridCacheVersion explicitVer = txEntry.drVersion() != null ? - txEntry.drVersion() : writeVersion(); - - if (op == CREATE || op == UPDATE && txEntry.drExpireTime() == -1L) { - ExpiryPolicy expiry = txEntry.expiry(); + GridCacheVersion explicitVer = txEntry.conflictVersion() != null ? + txEntry.conflictVersion() : writeVersion(); - if (expiry == null) - expiry = cacheCtx.expiry(); + if ((op == CREATE || op == UPDATE) && + txEntry.conflictExpireTime() == CONFLICT_EXPIRE_TIME_NOT_SET) { + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); if (expiry != null) { Duration duration = cached.hasValue() ? @@ -771,7 +768,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> if (drNeedResolve) { IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContext<K, V>> drRes = conflictResolve(op, txEntry.key(), val, valBytes, txEntry.ttl(), - txEntry.drExpireTime(), explicitVer, cached); + txEntry.conflictExpireTime(), explicitVer, cached); assert drRes != null; @@ -783,9 +780,9 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> txEntry.ttl(conflictCtx.ttl()); if (conflictCtx.newEntry().dataCenterId() != cctx.dataCenterId()) - txEntry.drExpireTime(conflictCtx.expireTime()); + txEntry.conflictExpireTime(conflictCtx.expireTime()); else - txEntry.drExpireTime(-1L); + txEntry.conflictExpireTime(CONFLICT_EXPIRE_TIME_NOT_SET); } else { assert conflictCtx.isMerge(); @@ -796,7 +793,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> explicitVer = writeVersion(); txEntry.ttl(conflictCtx.ttl()); - txEntry.drExpireTime(-1L); + txEntry.conflictExpireTime(CONFLICT_EXPIRE_TIME_NOT_SET); } } else @@ -810,7 +807,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> txEntry.valueBytes(valBytes); txEntry.op(op); txEntry.entryProcessors(null); - txEntry.drVersion(explicitVer); + txEntry.conflictVersion(explicitVer); } if (op == CREATE || op == UPDATE) { @@ -828,7 +825,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> topVer, null, cached.detached() ? DR_NONE : drType, - txEntry.drExpireTime(), + txEntry.conflictExpireTime(), cached.isNear() ? null : explicitVer, CU.subjectId(this, cctx), resolveTaskName()); @@ -848,7 +845,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> topVer, CU.<K, V>empty(), DR_NONE, - txEntry.drExpireTime(), + txEntry.conflictExpireTime(), null, CU.subjectId(this, cctx), resolveTaskName()); @@ -892,10 +889,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> nearCached.innerReload(); } else if (op == READ) { - ExpiryPolicy expiry = txEntry.expiry(); - - if (expiry == null) - expiry = cacheCtx.expiry(); + ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); if (expiry != null) { Duration duration = expiry.getExpiryForAccess(); @@ -2380,7 +2374,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> if (updateTtl) { if (!read) { - ExpiryPolicy expiryPlc = txEntry.expiry() != null ? txEntry.expiry() : cacheCtx.expiry(); + ExpiryPolicy expiryPlc = cacheCtx.expiryForTxEntry(txEntry); if (expiryPlc != null) txEntry.ttl(CU.toTtl(expiryPlc.getExpiryForAccess())); @@ -3192,7 +3186,7 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> filter, drVer); - txEntry.drExpireTime(drExpireTime); + txEntry.conflictExpireTime(drExpireTime); if (!hasDrTtl) txEntry.expiry(expiryPlc); @@ -3257,24 +3251,6 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> /** * @param key Key. - * @param ttl Time to live. - * @return {@code true} if tx entry exists for this key, {@code false} otherwise. - */ - public boolean entryTtl(IgniteTxKey<K> key, long ttl) { - assert key != null; - - IgniteTxEntry<K, V> e = entry(key); - - if (e != null) { - e.ttl(ttl); - e.drExpireTime(-1L); - } - - return e != null; - } - - /** - * @param key Key. * @param expiryPlc Expiry policy. */ void entryExpiry(IgniteTxKey<K> key, @Nullable ExpiryPolicy expiryPlc) { @@ -3282,8 +3258,10 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> IgniteTxEntry<K, V> e = entry(key); - if (e != null) + if (e != null) { e.expiry(expiryPlc); + e.conflictExpireTime(CONFLICT_EXPIRE_TIME_NOT_SET); + } } /** @@ -3300,10 +3278,8 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V> if (e != null) { e.ttl(ttl); - - e.drExpireTime(expireTime); - e.expiry(null); + e.conflictExpireTime(expireTime); } return e != null;
