ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From akuznet...@apache.org
Subject [40/50] [abbrv] incubator-ignite git commit: Applied fix.
Date Thu, 29 Jan 2015 03:14:15 GMT
Applied fix.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/8e8a0c3f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/8e8a0c3f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/8e8a0c3f

Branch: refs/heads/ignite-102
Commit: 8e8a0c3ffa15442018b9dc696f12210ee36b9898
Parents: 9996140
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Wed Jan 28 17:38:15 2015 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Wed Jan 28 17:38:15 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheContext.java      |  43 +++++
 .../processors/cache/GridCacheEntryEx.java      |   8 +
 .../processors/cache/GridCacheMapEntry.java     | 116 +++++++++---
 .../cache/GridCacheUpdateAtomicResult.java      |   6 +-
 .../processors/cache/GridDrResolveResult.java   |  63 -------
 .../GridDistributedTxRemoteAdapter.java         |  47 +++--
 .../dht/atomic/GridDhtAtomicCache.java          |  10 +-
 .../processors/cache/dr/GridCacheDrManager.java |  57 ++----
 .../cache/dr/os/GridOsCacheDrManager.java       |  35 +---
 .../cache/transactions/IgniteTxAdapter.java     |  45 +++++
 .../transactions/IgniteTxLocalAdapter.java      |  52 +++--
 ...ridCacheVersionAbstractConflictResolver.java |  56 ++++++
 .../GridCacheVersionConflictContext.java        |  73 +++++++
 .../GridCacheVersionConflictContextImpl.java    | 188 +++++++++++++++++++
 .../GridCacheVersionConflictResolver.java       |  59 ++++++
 .../version/GridCacheVersionedEntryEx.java      |   2 +-
 .../processors/cache/GridCacheTestEntryEx.java  |   5 +
 17 files changed, 666 insertions(+), 199 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/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 571a7a4..3fb5329 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
@@ -190,6 +190,9 @@ public class GridCacheContext<K, V> implements Externalizable {
     /** Cache weak query iterator holder. */
     private CacheWeakQueryIteratorsHolder<Map.Entry<K, V>> itHolder;
 
+    /** Conflict resolver. */
+    private GridCacheVersionAbstractConflictResolver conflictRslvr;
+
     /**
      * Empty constructor required for {@link Externalizable}.
      */
@@ -306,6 +309,14 @@ public class GridCacheContext<K, V> implements Externalizable {
             expiryPlc = null;
 
         itHolder = new CacheWeakQueryIteratorsHolder(log);
+
+        // Conflict resolver is determined in two stages:
+        // 1. If DR receiver hub is enabled, then pick it from DR manager.
+        // 2. Otherwise instantiate default resolver in case local store is configured.
+        conflictRslvr = drMgr.conflictResolver();
+
+        if (conflictRslvr == null && storeMgr.isLocalStore())
+            conflictRslvr = new GridCacheVersionConflictResolver();
     }
 
     /**
@@ -1546,6 +1557,38 @@ public class GridCacheContext<K, V> implements Externalizable {
     }
 
     /**
+     * Check whether conflict resolution is required.
+     *
+     * @param oldVer Old version.
+     * @param newVer New version.
+     * @return {@code True} in case DR is required.
+     */
+    public boolean conflictNeedResolve(GridCacheVersion oldVer, GridCacheVersion newVer) {
+        return conflictRslvr != null;
+    }
+
+    /**
+     * Resolve DR conflict.
+     *
+     * @param oldEntry Old entry.
+     * @param newEntry New entry.
+     * @param atomicVerComparator Whether to use atomic version comparator.
+     * @return Conflict resolution result.
+     * @throws IgniteCheckedException In case of exception.
+     */
+    public GridCacheVersionConflictContextImpl<K, V> conflictResolve(GridCacheVersionedEntryEx<K, V> oldEntry,
+        GridCacheVersionedEntryEx<K, V> newEntry, boolean atomicVerComparator) throws IgniteCheckedException {
+        assert conflictRslvr != null : "Should not reach this place.";
+
+        GridCacheVersionConflictContextImpl<K, V> ctx = conflictRslvr.resolve(oldEntry, newEntry, atomicVerComparator);
+
+        if (ctx.isManualResolve())
+            drMgr.onReceiveCacheConflictResolved(ctx.isUseNew(), ctx.isUseOld(), ctx.isMerge());
+
+        return ctx;
+    }
+
+    /**
      * @return Data center ID.
      */
     public byte dataCenterId() {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
index 6748d6e..8eeacc5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java
@@ -671,6 +671,14 @@ public interface GridCacheEntryEx<K, V> {
         throws IgniteCheckedException, GridCacheEntryRemovedException;
 
     /**
+     * Create versioned entry for this cache entry.
+     *
+     * @return Versioned entry.
+     * @throws IgniteCheckedException In case of error.
+     */
+    public GridCacheVersionedEntryEx<K, V> versionedEntry() throws IgniteCheckedException;
+
+    /**
      * Sets new value if passed in version matches the current version
      * (used for read-through only).
      *

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 0b34457..3c4e9d6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -68,7 +68,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
     private static final byte IS_UNSWAPPED_MASK = 0x02;
 
     /** */
-    private static final Comparator<GridCacheVersion> ATOMIC_VER_COMPARATOR = new GridCacheAtomicVersionComparator();
+    public static final Comparator<GridCacheVersion> ATOMIC_VER_COMPARATOR = new GridCacheAtomicVersionComparator();
 
     /**
      * NOTE
@@ -1658,7 +1658,7 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
 
         GridCacheVersion enqueueVer = null;
 
-        GridDrResolveResult<V> drRes = null;
+        GridCacheVersionConflictContextImpl<K, V> drRes = null;
 
         EntryProcessorResult<?> invokeRes = null;
 
@@ -1675,49 +1675,113 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
             if (isNew())
                 unswap(true, retval);
 
+            boolean newTtlResolved = false;
+
+            boolean drNeedResolve = false;
+
             Object transformClo = null;
 
             if (drResolve) {
-                drRes = cctx.dr().resolveAtomic(this,
-                    op,
-                    writeObj,
-                    valBytes,
-                    expiryPlc,
-                    drTtl,
-                    drExpireTime,
-                    drVer);
+                GridCacheVersion oldDrVer = version().drVersion();
+
+                drNeedResolve = cctx.conflictNeedResolve(oldDrVer, drVer);
+
+                if (drNeedResolve) {
+                    // Get old value.
+                    V oldVal = rawGetOrUnmarshalUnlocked(true);
+
+                    if (writeObj == null && valBytes != null)
+                        writeObj = cctx.marshaller().unmarshal(valBytes, cctx.deploy().globalLoader());
+
+                    if (op == GridCacheOperation.TRANSFORM) {
+                        transformClo = writeObj;
+
+                        writeObj = ((IgniteClosure<V, V>)writeObj).apply(oldVal);
+                    }
+
+                    K k = key();
+
+                    if (drTtl >= 0L) {
+                        // DR TTL is set explicitly
+                        assert drExpireTime >= 0L;
+
+                        newTtl = drTtl;
+                        newExpireTime = drExpireTime;
+                    }
+                    else {
+                        long ttl = expiryPlc != null ? (isNew() ? expiryPlc.forCreate() : expiryPlc.forUpdate()) : -1L;
+
+                        newTtl = ttl < 0 ? ttlExtras() : ttl;
+                        newExpireTime = CU.toExpireTime(newTtl);
+                    }
+
+                    newTtlResolved = true;
+
+                    GridCacheVersionedEntryEx<K, V> oldEntry = versionedEntry();
+                    GridCacheVersionedEntryEx<K, V> newEntry =
+                        new GridCachePlainVersionedEntry<>(k, (V)writeObj, newTtl, newExpireTime, drVer);
+
+                    drRes = cctx.conflictResolve(oldEntry, newEntry, verCheck);
+
+                    assert drRes != null;
 
-                if (drRes != null) {
                     if (drRes.isUseOld()) {
+                        // Handle special case with atomic comparator.
+                        if (!isNew() &&                                            // Not initial value,
+                            verCheck &&                                            // and atomic version check,
+                            oldDrVer.dataCenterId() == drVer.dataCenterId() &&     // and data centers are equal,
+                            ATOMIC_VER_COMPARATOR.compare(oldDrVer, drVer) == 0 && // and both versions are equal,
+                            cctx.writeThrough() &&                                 // and store is enabled,
+                            primary)                                               // and we are primary.
+                        {
+                            V val = rawGetOrUnmarshalUnlocked(false);
+
+                            if (val == null) {
+                                assert deletedUnlocked();
+
+                                cctx.store().removeFromStore(null, key());
+                            }
+                            else
+                                cctx.store().putToStore(null, key(), val, ver);
+                        }
+
                         old = retval ? rawGetOrUnmarshalUnlocked(false) : val;
 
                         return new GridCacheUpdateAtomicResult<>(false,
                             old,
                             null,
                             invokeRes,
-                            -1L,
+                            0L,
                             -1L,
                             null,
                             null,
                             false);
                     }
+                    else if (drRes.isUseNew())
+                        op = writeObj != null ? GridCacheOperation.UPDATE : GridCacheOperation.DELETE;
+                    else {
+                        assert drRes.isMerge();
 
-                    newTtl = drRes.newTtl();
-
-                    newExpireTime = drRes.newExpireTime();
-
-                    newDrExpireTime = drRes.newDrExpireTime();
+                        writeObj = drRes.mergeValue();
+                        valBytes = null;
 
-                    op = drRes.operation();
+                        drVer = null; // Update will be considered as local.
 
-                    writeObj = drRes.value();
+                        op = writeObj != null ? GridCacheOperation.UPDATE : GridCacheOperation.DELETE;
+                    }
 
-                    valBytes = drRes.valueBytes();
+                    newTtl = drRes.ttl();
+                    newExpireTime = drRes.expireTime();
 
-                    if (drRes.isMerge())
-                        drVer = null; // Update will be considered as local.
+                    // Explicit DR expire time will be passed to remote node only in that case.
+                    if (!drRes.explicitTtl() && !drRes.isMerge()) {
+                        if (drRes.isUseNew() && newEntry.dataCenterId() != cctx.dataCenterId() ||
+                            drRes.isUseOld() && oldEntry.dataCenterId() != cctx.dataCenterId())
+                            newDrExpireTime = drRes.expireTime();
+                    }
                 }
                 else
+                    // Nullify DR version on this update, so that we will use regular version during next updates.
                     drVer = null;
             }
 
@@ -3095,6 +3159,14 @@ public abstract class GridCacheMapEntry<K, V> implements GridCacheEntryEx<K, V>
     }
 
     /** {@inheritDoc} */
+    @Override public synchronized GridCacheVersionedEntryEx<K, V> versionedEntry() throws IgniteCheckedException {
+        boolean isNew = isStartVersion();
+
+        return new GridCachePlainVersionedEntry<>(key, isNew ? unswap(true, true) : rawGetOrUnmarshalUnlocked(false),
+            ttlExtras(), expireTimeExtras(), ver.drVersion(), isNew);
+    }
+
+    /** {@inheritDoc} */
     @Override public synchronized boolean versionedValue(V val, GridCacheVersion curVer, GridCacheVersion newVer)
         throws IgniteCheckedException, GridCacheEntryRemovedException {
         checkObsolete();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateAtomicResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateAtomicResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateAtomicResult.java
index 0b7d776..34be479 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateAtomicResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUpdateAtomicResult.java
@@ -51,7 +51,7 @@ public class GridCacheUpdateAtomicResult<K, V> {
 
     /** DR resolution result. */
     @GridToStringInclude
-    private final GridDrResolveResult<V> drRes;
+    private final GridCacheVersionConflictContextImpl<K, V> drRes;
 
     /** Whether update should be propagated to DHT node. */
     private final boolean sndToDht;
@@ -79,7 +79,7 @@ public class GridCacheUpdateAtomicResult<K, V> {
         long newTtl,
         long drExpireTime,
         @Nullable GridCacheVersion rmvVer,
-        @Nullable GridDrResolveResult<V> drRes,
+        @Nullable GridCacheVersionConflictContextImpl<K, V> drRes,
         boolean sndToDht) {
         this.success = success;
         this.oldVal = oldVal;
@@ -144,7 +144,7 @@ public class GridCacheUpdateAtomicResult<K, V> {
     /**
      * @return DR conflict resolution context.
      */
-    @Nullable public GridDrResolveResult<V> drResolveResult() {
+    @Nullable public GridCacheVersionConflictContextImpl<K, V> drResolveResult() {
         return drRes;
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridDrResolveResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridDrResolveResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridDrResolveResult.java
deleted file mode 100644
index faf71ca..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridDrResolveResult.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.cache;
-
-/**
- *
- */
-public interface GridDrResolveResult<V> {
-    /**
-     * @return TTL.
-     */
-    public long newTtl();
-
-    /**
-     * @return Expire time.
-     */
-    public long newExpireTime();
-
-    /**
-     * @return DR expire time.
-     */
-    public long newDrExpireTime();
-
-    /**
-     * @return {@code True} in case merge is to be performed.
-     */
-    public boolean isMerge();
-
-    /**
-     * @return {@code True} in case old value should be used.
-     */
-    public boolean isUseOld();
-
-    /**
-     * @return Cache operation.
-     */
-    public GridCacheOperation operation();
-
-    /**
-     * @return Value.
-     */
-    public V value();
-
-    /**
-     * @return Value bytes.
-     */
-    public byte[] valueBytes();
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/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 cea3a0d..f7376cf 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
@@ -509,24 +509,37 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V>
                                         if (explicitVer == null)
                                             explicitVer = writeVersion(); // Force write version to be used.
 
-                                        GridDrResolveResult<V> drRes = cacheCtx.dr().resolveTx(cached,
-                                            txEntry,
-                                            explicitVer,
-                                            op,
-                                            val,
-                                            valBytes,
-                                            txEntry.ttl(),
-                                            txEntry.drExpireTime());
-
-                                        if (drRes != null) {
-                                            op = drRes.operation();
-                                            val = drRes.value();
-                                            valBytes = drRes.valueBytes();
-
-                                            if (drRes.isMerge())
+                                        boolean drNeedResolve =
+                                            cacheCtx.conflictNeedResolve(cached.version(), explicitVer);
+
+                                        if (drNeedResolve) {
+                                            IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContextImpl<K, V>>
+                                                drRes = conflictResolve(op, txEntry.key(), val, valBytes,
+                                                txEntry.ttl(), txEntry.drExpireTime(), explicitVer, cached);
+
+                                            assert drRes != null;
+
+                                            GridCacheVersionConflictContextImpl<K, V> drCtx = drRes.get2();
+
+                                            if (drCtx.isUseOld())
+                                                op = NOOP;
+                                            else if (drCtx.isUseNew()) {
+                                                txEntry.ttl(drCtx.ttl());
+
+                                                if (drCtx.newEntry().dataCenterId() != cacheCtx.dataCenterId())
+                                                    txEntry.drExpireTime(drCtx.expireTime());
+                                                else
+                                                    txEntry.drExpireTime(-1L);
+                                            }
+                                            else if (drCtx.isMerge()) {
+                                                op = drRes.get1();
+                                                val = drCtx.mergeValue();
+                                                valBytes = null;
                                                 explicitVer = writeVersion();
-                                            else if (op == NOOP)
-                                                txEntry.ttl(-1L);
+
+                                                txEntry.ttl(drCtx.ttl());
+                                                txEntry.drExpireTime(-1L);
+                                            }
                                         }
                                         else
                                             // Nullify explicit version so that innerSet/innerRemove will work as usual.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index c993397..c35743f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1083,7 +1083,11 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                         if (plc != null)
                             expiry = new UpdateExpiryPolicy(plc);
 
-                        if (writeThrough() && keys.size() > 1 && !ctx.dr().receiveEnabled()) {
+                        if (keys.size() > 1 &&                             // Several keys ...
+                            writeThrough() &&                              // and store is enabled ...
+                            !ctx.store().isLocalStore() &&                 // and this is not local store ...
+                            !ctx.dr().receiveEnabled()  // and no DR.
+                        ) {
                             // This method can only be used when there are no replicated entries in the batch.
                             UpdateBatchResult<K, V> updRes = updateWithBatch(node,
                                 hasNear,
@@ -1681,7 +1685,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
 
                 if (dhtFut != null) {
                     if (updRes.sendToDht()) { // Send to backups even in case of remove-remove scenarios.
-                        GridDrResolveResult<V> ctx = updRes.drResolveResult();
+                        GridCacheVersionConflictContextImpl<K, V> ctx = updRes.drResolveResult();
 
                         long ttl = updRes.newTtl();
                         long expireTime = updRes.drExpireTime();
@@ -1727,7 +1731,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                 if (hasNear) {
                     if (primary && updRes.sendToDht()) {
                         if (!ctx.affinity().belongs(node, entry.partition(), topVer)) {
-                            GridDrResolveResult<V> ctx = updRes.drResolveResult();
+                            GridCacheVersionConflictContextImpl<K, V> ctx = updRes.drResolveResult();
 
                             long ttl = updRes.newTtl();
                             long expireTime = updRes.drExpireTime();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrManager.java
index ff83198..d0a0c26 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/GridCacheDrManager.java
@@ -34,53 +34,9 @@ public interface GridCacheDrManager<K, V> extends GridCacheManager<K, V> {
     public byte dataCenterId();
 
     /**
-     * Handles DR for atomic cache.
-     *
-     * @param e Cache entry.
-     * @param op Operation.
-     * @param writeObj New value.
-     * @param valBytes New value byte.
-     * @param ttl TTL.
-     * @param drTtl DR TTL.
-     * @param drExpireTime DR expire time
-     * @param drVer DR version.
-     * @return DR result.
-     * @throws IgniteCheckedException If update failed.
-     * @throws GridCacheEntryRemovedException If entry is obsolete.
-     */
-    public GridDrResolveResult<V> resolveAtomic(GridCacheEntryEx<K, V> e,
-         GridCacheOperation op,
-         @Nullable Object writeObj,
-         @Nullable byte[] valBytes,
-         @Nullable IgniteCacheExpiryPolicy expiryPlc,
-         long drTtl,
-         long drExpireTime,
-         @Nullable GridCacheVersion drVer) throws IgniteCheckedException, GridCacheEntryRemovedException;
-
-    /**
-     * Handles DR for transactional cache.
-     *
-     * @param e Cache entry.
-     * @param txEntry Transaction entry.
-     * @param newVer Version.
-     * @param op Operation.
-     * @param newVal New value.
-     * @param newValBytes New value bytes.
-     * @param newTtl TTL.
-     * @param newDrExpireTime DR expire time
-     * @return DR result.
-     * @throws IgniteCheckedException If update failed.
-     * @throws GridCacheEntryRemovedException If entry is obsolete.
+     * @return Cache version conflict resolver.
      */
-    public GridDrResolveResult<V> resolveTx(
-        GridCacheEntryEx<K, V> e,
-        IgniteTxEntry<K, V> txEntry,
-        GridCacheVersion newVer,
-        GridCacheOperation op,
-        V newVal,
-        byte[] newValBytes,
-        long newTtl,
-        long newDrExpireTime) throws IgniteCheckedException, GridCacheEntryRemovedException;
+    public GridCacheVersionAbstractConflictResolver conflictResolver();
 
     /**
      * Performs replication.
@@ -138,6 +94,15 @@ public interface GridCacheDrManager<K, V> extends GridCacheManager<K, V> {
     public void onReceiveCacheEntriesReceived(int entriesCnt);
 
     /**
+     * Callback for manual conflict resolution.
+     *
+     * @param useNew Use new.
+     * @param useOld Use old.
+     * @param merge Merge.
+     */
+    public void onReceiveCacheConflictResolved(boolean useNew, boolean useOld, boolean merge);
+
+    /**
      * Resets metrics for current cache.
      */
     public void resetMetrics();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/os/GridOsCacheDrManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/os/GridOsCacheDrManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/os/GridOsCacheDrManager.java
index 20b8804..49f617b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/os/GridOsCacheDrManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/dr/os/GridOsCacheDrManager.java
@@ -20,7 +20,6 @@ package org.apache.ignite.internal.processors.cache.dr.os;
 import org.apache.ignite.*;
 import org.apache.ignite.internal.processors.cache.*;
 import org.apache.ignite.internal.processors.cache.dr.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
 import org.apache.ignite.internal.processors.dr.*;
 import org.jetbrains.annotations.*;
@@ -65,6 +64,11 @@ public class GridOsCacheDrManager<K, V> implements GridCacheDrManager<K, V> {
     }
 
     /** {@inheritDoc} */
+    @Override public GridCacheVersionAbstractConflictResolver conflictResolver() {
+        return null;
+    }
+
+    /** {@inheritDoc} */
     @Override public void replicate(K key,
         @Nullable byte[] keyBytes,
         @Nullable V val,
@@ -77,30 +81,6 @@ public class GridOsCacheDrManager<K, V> implements GridCacheDrManager<K, V> {
     }
 
     /** {@inheritDoc} */
-    @Override public GridDrResolveResult<V> resolveAtomic(GridCacheEntryEx<K, V> e,
-        GridCacheOperation op,
-        @Nullable Object writeObj,
-        @Nullable byte[] valBytes,
-        @Nullable IgniteCacheExpiryPolicy expiryPlc,
-        long drTtl,
-        long drExpireTime,
-        @Nullable GridCacheVersion drVer) throws IgniteCheckedException, GridCacheEntryRemovedException {
-        return null;
-    }
-
-    /** {@inheritDoc} */
-    @Override public GridDrResolveResult<V> resolveTx(GridCacheEntryEx<K, V> e,
-        IgniteTxEntry<K, V> txEntry,
-        GridCacheVersion newVer,
-        GridCacheOperation op,
-        V newVal,
-        byte[] newValBytes,
-        long newTtl,
-        long newDrExpireTime) throws IgniteCheckedException, GridCacheEntryRemovedException {
-        return null;
-    }
-
-    /** {@inheritDoc} */
     @Override public void beforeExchange(long topVer, boolean left) throws IgniteCheckedException {
         // No-op.
     }
@@ -116,6 +96,11 @@ public class GridOsCacheDrManager<K, V> implements GridCacheDrManager<K, V> {
     }
 
     /** {@inheritDoc} */
+    @Override public void onReceiveCacheConflictResolved(boolean useNew, boolean useOld, boolean merge) {
+        // No-op.
+    }
+
+    /** {@inheritDoc} */
     @Override public void resetMetrics() {
         // No-op.
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/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 47cd12b..e079a5c 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
@@ -1252,6 +1252,51 @@ public abstract class IgniteTxAdapter<K, V> extends GridMetadataAwareAdapter
     }
 
     /**
+     * Resolve DR conflict.
+     *
+     * @param op Initially proposed operation.
+     * @param key Key.
+     * @param newVal New value.
+     * @param newValBytes New value bytes.
+     * @param newTtl New TTL.
+     * @param newDrExpireTime New explicit DR expire time.
+     * @param newVer New version.
+     * @param old Old entry.
+     * @return Tuple with adjusted operation type and conflict context.
+     * @throws org.apache.ignite.IgniteCheckedException In case of eny exception.
+     * @throws GridCacheEntryRemovedException If entry got removed.
+     */
+    protected IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContextImpl<K, V>> conflictResolve(
+        GridCacheOperation op, K key, V newVal, byte[] newValBytes, long newTtl, long newDrExpireTime,
+        GridCacheVersion newVer, GridCacheEntryEx<K, V> old)
+        throws IgniteCheckedException, GridCacheEntryRemovedException {
+        // Construct old entry info.
+        GridCacheVersionedEntryEx<K, V> oldEntry = old.versionedEntry();
+
+        // Construct new entry info.
+        if (newVal == null && newValBytes != null)
+            newVal = cctx.marshaller().unmarshal(newValBytes, cctx.deploy().globalLoader());
+
+        long newExpireTime = newDrExpireTime >= 0L ? newDrExpireTime : CU.toExpireTime(newTtl);
+
+        GridCacheVersionedEntryEx<K, V> newEntry =
+            new GridCachePlainVersionedEntry<K, V>(key, newVal, newTtl, newExpireTime, newVer);
+
+        GridCacheVersionConflictContextImpl<K, V> ctx = old.context().conflictResolve(oldEntry, newEntry, false);
+
+        if (ctx.isMerge()) {
+            V resVal = ctx.mergeValue();
+
+            if ((op == CREATE || op == UPDATE) && resVal == null)
+                op = DELETE;
+            else if (op == DELETE && resVal != null)
+                op = old.isNewLocked() ? CREATE : UPDATE;
+        }
+
+        return F.t(op, ctx);
+    }
+
+    /**
      * @param e Transaction entry.
      * @param primaryOnly Flag to include backups into check or not.
      * @return {@code True} if entry is locally mapped as a primary or back up node.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/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 46ab74f..d9c49d8 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
@@ -699,31 +699,45 @@ public abstract class IgniteTxLocalAdapter<K, V> extends IgniteTxAdapter<K, V>
                                         }
                                     }
 
-                                    GridDrResolveResult<V> drRes = cacheCtx.dr().resolveTx(cached,
-                                        txEntry,
-                                        explicitVer,
-                                        op,
-                                        val,
-                                        valBytes,
-                                        txEntry.ttl(),
-                                        txEntry.drExpireTime());
-
-                                    if (drRes != null) {
-                                        op = drRes.operation();
-                                        val = drRes.value();
-                                        valBytes = drRes.valueBytes();
-
-                                        if (drRes.isMerge())
+                                    boolean drNeedResolve = cacheCtx.conflictNeedResolve(cached.version(), explicitVer);
+
+                                    if (drNeedResolve) {
+                                        IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContextImpl<K, V>>
+                                            drRes = conflictResolve(op, txEntry.key(), val, valBytes, txEntry.ttl(),
+                                                txEntry.drExpireTime(), explicitVer, cached);
+
+                                        assert drRes != null;
+
+                                        GridCacheVersionConflictContextImpl<K, V> conflictCtx = drRes.get2();
+
+                                        if (conflictCtx.isUseOld())
+                                            op = NOOP;
+                                        else if (conflictCtx.isUseNew()) {
+                                            txEntry.ttl(conflictCtx.ttl());
+
+                                            if (conflictCtx.newEntry().dataCenterId() != cctx.dataCenterId())
+                                                txEntry.drExpireTime(conflictCtx.expireTime());
+                                            else
+                                                txEntry.drExpireTime(-1L);
+                                        }
+                                        else {
+                                            assert conflictCtx.isMerge();
+
+                                            op = drRes.get1();
+                                            val = conflictCtx.mergeValue();
+                                            valBytes = null;
                                             explicitVer = writeVersion();
-                                        else if (op == NOOP)
-                                            txEntry.ttl(-1L);
+
+                                            txEntry.ttl(conflictCtx.ttl());
+                                            txEntry.drExpireTime(-1L);
+                                        }
                                     }
                                     else
                                         // Nullify explicit version so that innerSet/innerRemove will work as usual.
                                         explicitVer = null;
 
-                                    if (sndTransformedVals || (drRes != null)) {
-                                        assert sndTransformedVals && cacheCtx.isReplicated() || (drRes != null);
+                                    if (sndTransformedVals || drNeedResolve) {
+                                        assert sndTransformedVals && cacheCtx.isReplicated() || drNeedResolve;
 
                                         txEntry.value(val, true, false);
                                         txEntry.valueBytes(valBytes);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionAbstractConflictResolver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionAbstractConflictResolver.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionAbstractConflictResolver.java
new file mode 100644
index 0000000..a91bd4d
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionAbstractConflictResolver.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.version;
+
+import org.apache.ignite.*;
+
+/**
+ * Cache version conflict resolver.
+ */
+public abstract class GridCacheVersionAbstractConflictResolver {
+    /**
+     * Resolve the conflict.
+     *
+     * @param oldEntry Old entry.
+     * @param newEntry New entry.
+     * @param atomicVerComparator Whether to use atomic version comparator.
+     * @return Conflict resolution context.
+     * @throws IgniteCheckedException If failed.
+     */
+    public <K, V> GridCacheVersionConflictContextImpl<K, V> resolve(GridCacheVersionedEntryEx<K, V> oldEntry,
+        GridCacheVersionedEntryEx<K, V> newEntry, boolean atomicVerComparator) throws IgniteCheckedException {
+        GridCacheVersionConflictContextImpl<K, V> ctx = new GridCacheVersionConflictContextImpl<>(oldEntry, newEntry);
+
+        resolve0(ctx, oldEntry, newEntry, atomicVerComparator);
+
+        return ctx;
+    }
+
+    /**
+     * Internal conflict resolution routine.
+     *
+     * @param ctx Context.
+     * @param oldEntry Old entry.
+     * @param newEntry New entry.
+     * @param atomicVerComparator Whether to use atomic version comparator.
+     * @throws IgniteCheckedException If failed.
+     */
+    protected abstract <K, V> void resolve0(GridCacheVersionConflictContextImpl<K, V> ctx,
+        GridCacheVersionedEntryEx<K, V> oldEntry, GridCacheVersionedEntryEx<K, V> newEntry,
+        boolean atomicVerComparator) throws IgniteCheckedException;
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContext.java
new file mode 100644
index 0000000..72c323b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContext.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.version;
+
+import org.apache.ignite.cache.*;
+import org.jetbrains.annotations.*;
+
+/**
+ * Cache version conflict context.
+ */
+public interface GridCacheVersionConflictContext<K, V> {
+    /**
+     * Gets old (existing) cache entry.
+     *
+     * @return Old (existing) cache entry.
+     */
+    public GridCacheVersionedEntry<K, V> oldEntry();
+
+    /**
+     * Gets new cache entry.
+     *
+     * @return New cache entry.
+     */
+    public GridCacheVersionedEntry<K, V> newEntry();
+
+    /**
+     * Force cache to ignore new entry and leave old (existing) entry unchanged.
+     */
+    public void useOld();
+
+    /**
+     * Force cache to apply new entry overwriting old (existing) entry.
+     * <p>
+     * Note that updates from remote data centers always have explicit TTL , while local data center
+     * updates will only have explicit TTL in case {@link CacheEntry#timeToLive(long)} was called
+     * before update. In the latter case new entry will pick TTL of the old (existing) entry, even
+     * if it was set through update from remote data center. it means that depending on concurrent
+     * update timings new update might pick unexpected TTL. For example, consider that three updates
+     * of the same key are performed: local update with explicit TTL (1) followed by another local
+     * update without explicit TTL (2) and one remote update (3). In this case you might expect that
+     * update (2) will pick TTL set during update (1). However, in case update (3) occurrs between (1)
+     * and (2) and it overwrites (1) during conflict resolution, then update (2) will pick TTL of
+     * update (3). To have predictable TTL in such cases you should either always set it explicitly
+     * through {@code GridCacheEntry.timeToLive(long)} or use {@link #merge(Object, long)}.
+     */
+    public void useNew();
+
+    /**
+     * Force cache to use neither old, nor new, but some other value passed as argument. In this case old
+     * value will be replaced with merge value and update will be considered as local.
+     * <p>
+     * Also in case of merge you have to specify new TTL explicitly. For unlimited TTL use {@code 0}.
+     *
+     * @param mergeVal Merge value or {@code null} to force remove.
+     * @param ttl Time to live in milliseconds.
+     */
+    public void merge(@Nullable V mergeVal, long ttl);
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContextImpl.java
new file mode 100644
index 0000000..228a224
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictContextImpl.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.version;
+
+import org.apache.ignite.cache.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.jetbrains.annotations.*;
+
+/**
+ * Conflict context implementation.
+ */
+public class GridCacheVersionConflictContextImpl<K, V> implements GridCacheVersionConflictContext<K, V> {
+    /** Old entry. */
+    @GridToStringInclude
+    private final GridCacheVersionedEntry<K, V> oldEntry;
+
+    /** New entry. */
+    @GridToStringInclude
+    private final GridCacheVersionedEntry<K, V> newEntry;
+
+    /** Current state. */
+    private State state;
+
+    /** Current merge value. */
+    @GridToStringExclude
+    private V mergeVal;
+
+    /** TTL. */
+    private long ttl;
+
+    /** Explicit TTL flag. */
+    private boolean explicitTtl;
+
+    /** Manual resolve flag. */
+    private boolean manualResolve;
+
+    /**
+     * Constructor.
+     *
+     * @param oldEntry Old entry.
+     * @param newEntry New entry.
+     */
+    public GridCacheVersionConflictContextImpl(GridCacheVersionedEntry<K, V> oldEntry,
+        GridCacheVersionedEntry<K, V> newEntry) {
+        assert oldEntry != null && newEntry != null;
+        assert oldEntry.ttl() >= 0 && newEntry.ttl() >= 0;
+
+        this.oldEntry = oldEntry;
+        this.newEntry = newEntry;
+
+        // Set initial state.
+        useNew();
+    }
+
+    /** {@inheritDoc} */
+    @Override public GridCacheVersionedEntry<K, V> oldEntry() {
+        return oldEntry;
+    }
+
+    /** {@inheritDoc} */
+    @Override public GridCacheVersionedEntry<K, V> newEntry() {
+        return newEntry;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void useOld() {
+        state = State.USE_OLD;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void useNew() {
+        state = State.USE_NEW;
+
+        if (!explicitTtl)
+            ttl = newEntry.ttl();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void merge(@Nullable V mergeVal, long ttl) {
+        state = State.MERGE;
+
+        this.mergeVal = mergeVal;
+        this.ttl = ttl;
+
+        explicitTtl = true;
+    }
+
+    /**
+     * @return {@code True} in case old value should be used.
+     */
+    public boolean isUseOld() {
+        return state == State.USE_OLD;
+    }
+
+    /**
+     * @return {@code True} in case new value should be used.
+     */
+    public boolean isUseNew() {
+        return state == State.USE_NEW;
+    }
+
+    /**
+     * @return {@code True} in case merge is to be performed.
+     */
+    public boolean isMerge() {
+        return state == State.MERGE;
+    }
+
+    /**
+     * Set manual resolve class.
+     */
+    public void manualResolve() {
+        this.manualResolve = true;
+    }
+
+    /**
+     * @return Manual resolve flag.
+     */
+    public boolean isManualResolve() {
+        return manualResolve;
+    }
+
+    /**
+     * @return Value to merge (if any).
+     */
+    @Nullable public V mergeValue() {
+        return mergeVal;
+    }
+
+    /**
+     * @return TTL.
+     */
+    public long ttl() {
+        return ttl;
+    }
+
+    /**
+     * @return Expire time.
+     */
+    public long expireTime() {
+        return explicitTtl ? CU.toExpireTime(ttl) : isUseNew() ? newEntry.expireTime() :
+            isUseOld() ? oldEntry.expireTime() : 0L;
+    }
+
+    /**
+     * @return Explicit TTL flag.
+     */
+    public boolean explicitTtl() {
+        return explicitTtl;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return state == State.MERGE ?
+            S.toString(GridCacheVersionConflictContextImpl.class, this, "mergeValue", mergeVal) :
+            S.toString(GridCacheVersionConflictContextImpl.class, this);
+    }
+
+    /**
+     * State.
+     */
+    private enum State {
+        /** Use old. */
+        USE_OLD,
+
+        /** Use new. */
+        USE_NEW,
+
+        /** Merge. */
+        MERGE
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictResolver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictResolver.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictResolver.java
new file mode 100644
index 0000000..e327fb9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionConflictResolver.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.version;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.processors.cache.*;
+
+/**
+ * Default conflict resolver.
+ */
+public class GridCacheVersionConflictResolver extends GridCacheVersionAbstractConflictResolver {
+    /** {@inheritDoc} */
+    @Override protected <K, V> void resolve0(GridCacheVersionConflictContextImpl<K, V> ctx,
+        GridCacheVersionedEntryEx<K, V> oldEntry, GridCacheVersionedEntryEx<K, V> newEntry,
+        boolean atomicVerComparator) throws IgniteCheckedException {
+        if (newEntry.dataCenterId() != oldEntry.dataCenterId())
+            ctx.useNew();
+        else {
+            if (oldEntry.isStartVersion())
+                ctx.useNew();
+            else {
+                if (atomicVerComparator) {
+                    // Handle special case when version check using ATOMIC cache comparator is required.
+                    if (GridCacheMapEntry.ATOMIC_VER_COMPARATOR.compare(oldEntry.version(), newEntry.version()) >= 0)
+                        ctx.useOld();
+                    else
+                        ctx.useNew();
+                }
+                else {
+                    long topVerDiff = newEntry.topologyVersion() - oldEntry.topologyVersion();
+
+                    if (topVerDiff > 0)
+                        ctx.useNew();
+                    else if (topVerDiff < 0)
+                        ctx.useOld();
+                    else if (newEntry.order() > oldEntry.order())
+                        ctx.useNew();
+                    else
+                        ctx.useOld();
+                }
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionedEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionedEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionedEntryEx.java
index b8f7c19..5351966 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionedEntryEx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionedEntryEx.java
@@ -25,7 +25,7 @@ import org.apache.ignite.cache.*;
 public interface GridCacheVersionedEntryEx<K, V> extends GridCacheVersionedEntry<K, V>, GridCacheVersionable {
     /**
      *
-     * @return
+     * @return {@code True} if entry is new.
      */
     public boolean isStartVersion();
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8e8a0c3f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
index 60df4e4..d1b6ce6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java
@@ -631,6 +631,11 @@ public class GridCacheTestEntryEx<K, V> extends GridMetadataAwareAdapter impleme
     }
 
     /** @inheritDoc */
+    @Override public GridCacheVersionedEntryEx<K, V> versionedEntry() throws IgniteCheckedException {
+        return null;
+    }
+
+    /** @inheritDoc */
     @Override public boolean versionedValue(V val, GridCacheVersion curVer, GridCacheVersion newVer) {
         assert false; return false;
     }


Mime
View raw message