ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ntikho...@apache.org
Subject ignite git commit: IGNITE-462 Merged changes from optimize branch.
Date Wed, 18 Nov 2015 17:34:56 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-426-2 203e88b92 -> 803a41414


IGNITE-462 Merged changes from optimize branch.


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

Branch: refs/heads/ignite-426-2
Commit: 803a41414df5d8426841a1491c8b766c43d3497a
Parents: 203e88b
Author: nikolay_tikhonov <ntikhonov@gridgain.com>
Authored: Wed Nov 18 20:34:24 2015 +0300
Committer: nikolay_tikhonov <ntikhonov@gridgain.com>
Committed: Wed Nov 18 20:34:24 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheMapEntry.java     |  18 +-
 .../dht/atomic/GridDhtAtomicCache.java          |  12 +-
 .../dht/atomic/GridDhtAtomicUpdateFuture.java   | 204 ++++++++++---------
 .../dht/atomic/GridDhtAtomicUpdateRequest.java  |  54 ++++-
 .../continuous/CacheContinuousQueryManager.java |  11 +-
 ...ContinuousQueryFailoverAbstractSelfTest.java |   2 +-
 6 files changed, 180 insertions(+), 121 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/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 75c1039..7a3be2f 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
@@ -1190,7 +1190,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
 
             if (cctx.isLocal() || cctx.isReplicated() ||
                 (!isNear() && !(tx != null && tx.onePhaseCommit() &&
!tx.local())))
-                cctx.continuousQueries().onEntryUpdated(this, key, val, old, tx.local(),
false, updateCntr0, topVer);
+                cctx.continuousQueries().onEntryUpdated(key, val, old, isInternal() || !context().userCache(),
+                    partition(), tx.local(), false, updateCntr0, topVer);
 
             cctx.dataStructures().onEntryUpdated(key, false);
         }
@@ -1369,7 +1370,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
 
             if (cctx.isLocal() || cctx.isReplicated() ||
                 (!isNear() && !(tx != null && tx.onePhaseCommit() &&
!tx.local())))
-                cctx.continuousQueries().onEntryUpdated(this, key, null, old, tx.local(),
false, updateCntr0, topVer);
+                cctx.continuousQueries().onEntryUpdated(key, null, old, isInternal()
+                    || !context().userCache(),partition(), tx.local(), false, updateCntr0,
topVer);
 
             cctx.dataStructures().onEntryUpdated(key, true);
 
@@ -1707,8 +1709,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
             if (!isNear()) {
                 long updateCntr = nextPartCounter(AffinityTopologyVersion.NONE);
 
-                cctx.continuousQueries().onEntryUpdated(this, key, val, old, true, false,
updateCntr,
-                    AffinityTopologyVersion.NONE);
+                cctx.continuousQueries().onEntryUpdated(key, val, old, isInternal() || !context().userCache(),
+                    partition(), true, false, updateCntr, AffinityTopologyVersion.NONE);
             }
 
             cctx.dataStructures().onEntryUpdated(key, op == GridCacheOperation.DELETE);
@@ -1967,9 +1969,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
                             if (updateCntr != null)
                                 updateCntr0 = updateCntr;
 
-                            cctx.continuousQueries().onEntryUpdated(GridCacheMapEntry.this,
key, evtVal,
-                                    prevVal, primary, false, updateCntr0, topVer);
-                        }
+                            cctx.continuousQueries().onEntryUpdated(key, evtVal, prevVal,
isInternal()
+                                || !context().userCache(), partition(), primary, false, updateCntr0,
topVer);                        }
 
                         return new GridCacheUpdateAtomicResult(false,
                             retval ? rawGetOrUnmarshalUnlocked(false) : null,
@@ -3227,7 +3228,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
                 drReplicate(drType, val, ver);
 
                 if (!skipQryNtf) {
-                    cctx.continuousQueries().onEntryUpdated(this, key, val, null, true, preload,
updateCntr, topVer);
+                    cctx.continuousQueries().onEntryUpdated(key, val, null, this.isInternal()
+                        || !this.context().userCache(), this.partition(), true, preload,
updateCntr, topVer);
 
                     cctx.dataStructures().onEntryUpdated(key, false);
                 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/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 1ef1d1e..94c3ee1 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
@@ -1836,8 +1836,9 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K,
V> {
                     }
                 }
                 else if (!entry.isNear() && updRes.success()) {
-                    ctx.continuousQueries().onEntryUpdated(entry, entry.key(), updRes.newValue(),
updRes.oldValue(),
-                        primary, false, updRes.updateCounter(), topVer);
+                    ctx.continuousQueries().onEntryUpdated(entry.key(), updRes.newValue(),
updRes.oldValue(),
+                        entry.isInternal() || !context().userCache(), entry.partition(),
primary, false,
+                        updRes.updateCounter(), topVer);
                 }
 
                 if (hasNear) {
@@ -2525,7 +2526,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K,
V> {
                         CacheObject prevVal = req.previousValue(i);
 
                         EntryProcessor<Object, Object, Object> entryProcessor = req.entryProcessor(i);
-                        Long updateIdx = req.updateIdx(i);
+                        Long updateIdx = req.updateCounter(i);
 
                         GridCacheOperation op = entryProcessor != null ? TRANSFORM :
                             (val != null) ? UPDATE : DELETE;
@@ -2565,8 +2566,9 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K,
V> {
                             ctx.onDeferredDelete(entry, updRes.removeVersion());
 
                         if (updRes.success() && !entry.isNear())
-                            ctx.continuousQueries().onEntryUpdated(entry, entry.key(), updRes.newValue(),
-                                updRes.oldValue(), false, false, updRes.updateCounter(),
req.topologyVersion());
+                            ctx.continuousQueries().onEntryUpdated(entry.key(), updRes.newValue(),
+                                updRes.oldValue(), entry.isInternal() || !context().userCache(),
entry.partition(),
+                                false, false, updRes.updateCounter(), req.topologyVersion());
 
                         entry.onUnlock();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 7befd42..eb09ff2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -20,10 +20,8 @@ package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
 import javax.cache.processor.EntryProcessor;
 import org.apache.ignite.IgniteCheckedException;
@@ -45,12 +43,10 @@ import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.CI2;
 import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.T4;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteUuid;
 import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentHashMap8;
 
 import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
 
@@ -69,42 +65,42 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
     protected static IgniteLogger log;
 
     /** Cache context. */
-    private GridCacheContext cctx;
+    private final GridCacheContext cctx;
 
     /** Future version. */
-    private GridCacheVersion futVer;
+    private final GridCacheVersion futVer;
 
     /** Write version. */
-    private GridCacheVersion writeVer;
+    private final GridCacheVersion writeVer;
 
     /** Force transform backup flag. */
     private boolean forceTransformBackups;
 
     /** Completion callback. */
     @GridToStringExclude
-    private CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> completionCb;
+    private final CI2<GridNearAtomicUpdateRequest, GridNearAtomicUpdateResponse> completionCb;
 
     /** Mappings. */
     @GridToStringInclude
-    private ConcurrentMap<UUID, GridDhtAtomicUpdateRequest> mappings = new ConcurrentHashMap8<>();
+    private final Map<UUID, GridDhtAtomicUpdateRequest> mappings;
 
     /** Entries with readers. */
     private Map<KeyCacheObject, GridDhtCacheEntry> nearReadersEntries;
 
     /** Update request. */
-    private GridNearAtomicUpdateRequest updateReq;
+    private final GridNearAtomicUpdateRequest updateReq;
 
     /** Update response. */
-    private GridNearAtomicUpdateResponse updateRes;
+    private final GridNearAtomicUpdateResponse updateRes;
 
     /** Future keys. */
-    private Collection<KeyCacheObject> keys;
-
-    /** Updates. */
-    private List<T4<GridDhtCacheEntry, CacheObject, CacheObject, Long>> updates;
+    private final Collection<KeyCacheObject> keys;
 
     /** */
-    private boolean waitForExchange;
+    private final boolean waitForExchange;
+
+    /** Response count. */
+    private volatile int resCnt;
 
     /**
      * @param cctx Cache context.
@@ -124,22 +120,21 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
         this.cctx = cctx;
         this.writeVer = writeVer;
 
-        futVer = cctx.versions().next(updateReq.topologyVersion());
-        this.updateReq = updateReq;
-        this.completionCb = completionCb;
-        this.updateRes = updateRes;
-
-        if (log == null)
-            log = U.logger(cctx.kernalContext(), logRef, GridDhtAtomicUpdateFuture.class);
+    futVer = cctx.versions().next(updateReq.topologyVersion());
+    this.updateReq = updateReq;
+    this.completionCb = completionCb;
+    this.updateRes = updateRes;
 
-        keys = new ArrayList<>(updateReq.keys().size());
+    if (log == null)
+    log = U.logger(cctx.kernalContext(), logRef, GridDhtAtomicUpdateFuture.class);
 
-        updates = new ArrayList<>(updateReq.keys().size());
+    keys = new ArrayList<>(updateReq.keys().size());
+    mappings = U.newHashMap(updateReq.keys().size());
 
-        boolean topLocked = updateReq.topologyLocked() || (updateReq.fastMap() &&
!updateReq.clientRequest());
+    boolean topLocked = updateReq.topologyLocked() || (updateReq.fastMap() && !updateReq.clientRequest());
 
-        waitForExchange = !topLocked;
-    }
+    waitForExchange = !topLocked;
+}
 
     /** {@inheritDoc} */
     @Override public IgniteUuid futureId() {
@@ -152,22 +147,42 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
     }
 
     /** {@inheritDoc} */
+    @Override public boolean onNodeLeft(UUID nodeId) {
+        if (log.isDebugEnabled())
+            log.debug("Processing node leave event [fut=" + this + ", nodeId=" + nodeId +
']');
+
+        return registerResponse(nodeId);
+    }
+
+    /** {@inheritDoc} */
     @Override public Collection<? extends ClusterNode> nodes() {
         return F.view(F.viewReadOnly(mappings.keySet(), U.id2Node(cctx.kernalContext())),
F.notNull());
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean onNodeLeft(UUID nodeId) {
-        if (log.isDebugEnabled())
-            log.debug("Processing node leave event [fut=" + this + ", nodeId=" + nodeId +
']');
+    /**
+     * @param nodeId Node ID.
+     * @return {@code True} if request found.
+     */
+    private boolean registerResponse(UUID nodeId) {
+        int resCnt0;
 
         GridDhtAtomicUpdateRequest req = mappings.get(nodeId);
 
         if (req != null) {
-            // Remove only after added keys to failed set.
-            mappings.remove(nodeId);
+            synchronized (this) {
+                if (req.onResponse()) {
+                    resCnt0 = resCnt;
 
-            checkComplete();
+                    resCnt0 += 1;
+
+                    resCnt = resCnt0;
+                }
+                else
+                    return false;
+            }
+
+            if (resCnt0 == mappings.size())
+                onDone();
 
             return true;
         }
@@ -205,7 +220,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
      * @param ttl TTL (optional).
      * @param conflictExpireTime Conflict expire time (optional).
      * @param conflictVer Conflict version (optional).
-     * @param updateIdx Partition update index.
+     * @param updateCntr Partition update counter.
      */
     public void addWriteEntry(GridDhtCacheEntry entry,
         @Nullable CacheObject val,
@@ -215,12 +230,10 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
         @Nullable GridCacheVersion conflictVer,
         boolean addPrevVal,
         @Nullable CacheObject prevVal,
-        @Nullable Long updateIdx) {
+        @Nullable Long updateCntr) {
         AffinityTopologyVersion topVer = updateReq.topologyVersion();
 
-        int part = entry.partition();
-
-        Collection<ClusterNode> dhtNodes = cctx.dht().topology().nodes(part, topVer);
+        Collection<ClusterNode> dhtNodes = cctx.dht().topology().nodes(entry.partition(),
topVer);
 
         if (log.isDebugEnabled())
             log.debug("Mapping entry to DHT nodes [nodes=" + U.nodeIds(dhtNodes) + ", entry="
+ entry + ']');
@@ -229,8 +242,6 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
 
         keys.add(entry.key());
 
-        updates.add(new T4<>(entry, val, prevVal, updateIdx));
-
         for (ClusterNode node : dhtNodes) {
             UUID nodeId = node.id();
 
@@ -261,8 +272,20 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
                     conflictExpireTime,
                     conflictVer,
                     addPrevVal,
+                    entry.partition(),
                     prevVal,
-                    updateIdx);
+                    updateCntr);
+            }
+            else if (dhtNodes.size() == 1) {
+                try {
+                    cctx.continuousQueries().onEntryUpdated(entry.key(), val, prevVal,
+                        entry.key().internal() || !cctx.userCache(), entry.partition(), true,
false,
+                        updateCntr, updateReq.topologyVersion());
+                }
+                catch (IgniteCheckedException e) {
+                    U.warn(log, "Failed to send continuous query message. [key=" + entry.key()
+ ", newVal="
+                        + val + ", err=" + e + "]");
+                }
             }
         }
     }
@@ -332,43 +355,55 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
             cctx.mvcc().removeAtomicFuture(version());
 
             if (err != null) {
-                int i = 0;
+                if (!mappings.isEmpty()) {
+                    Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
 
-                for (KeyCacheObject key : keys) {
-                    updateRes.addFailedKey(key, err);
+                    exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
+                        for (int i = 0; i < req.size(); i++) {
+                            KeyCacheObject key = req.key(i);
 
-                    if (i < updates.size()) {
+                            if (!hndKeys.contains(key)) {
+                                updateRes.addFailedKey(key, err);
 
-                        T4<GridDhtCacheEntry, CacheObject, CacheObject, Long> upd =
updates.get(i);
+                                cctx.continuousQueries().skipUpdateEvent(key, req.partitionId(i),
req.updateCounter(i),
+                                    updateReq.topologyVersion());
 
-                        cctx.continuousQueries().skipUpdateEvent(key, upd.get1().partition(),
upd.get4(),
-                            updateReq.topologyVersion());
+                                hndKeys.add(key);
 
-                        ++i;
+                                if (hndKeys.size() == keys.size())
+                                    break exit;
+                            }
+                        }
                     }
                 }
+                else
+                    for (KeyCacheObject key : keys)
+                        updateRes.addFailedKey(key, err);
             }
             else {
-                assert keys.size() >= updates.size();
-
-                int i = 0;
-
-                for (KeyCacheObject key : keys) {
-                    if (i == updates.size())
-                        break;
-
-                    T4<GridDhtCacheEntry, CacheObject, CacheObject, Long> upd = updates.get(i);
-
-                    try {
-                        cctx.continuousQueries().onEntryUpdated(upd.get1(), key, upd.get2(),
upd.get3(), true, false,
-                            upd.get4(), updateReq.topologyVersion());
-                    }
-                    catch (IgniteCheckedException e) {
-                        U.warn(log, "Failed to send continuous query message. [key=" + key
+ ", newVal="
-                            + upd.get1() + ", err=" + e + "]");
+                Collection<KeyCacheObject> hndKeys = new ArrayList<>(keys.size());
+
+                exit: for (GridDhtAtomicUpdateRequest req : mappings.values()) {
+                    for (int i = 0; i < req.size(); i++) {
+                        KeyCacheObject key = req.key(i);
+
+                        if (!hndKeys.contains(key)) {
+                            try {
+                                cctx.continuousQueries().onEntryUpdated(key, req.value(i),
req.localPreviousValue(i),
+                                    key.internal() || !cctx.userCache(), req.partitionId(i),
true, false,
+                                    req.updateCounter(i), updateReq.topologyVersion());
+                            }
+                            catch (IgniteCheckedException e) {
+                                U.warn(log, "Failed to send continuous query message. [key="
+ key + ", newVal="
+                                    + req.value(i) + ", err=" + e + "]");
+                            }
+
+                            hndKeys.add(key);
+
+                            if (hndKeys.size() == keys.size())
+                                break exit;
+                        }
                     }
-
-                    ++i;
                 }
             }
 
@@ -397,18 +432,18 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
                     U.warn(log, "Failed to send update request to backup node because it
left grid: " +
                         req.nodeId());
 
-                    mappings.remove(req.nodeId());
+                    registerResponse(req.nodeId());
                 }
                 catch (IgniteCheckedException e) {
                     U.error(log, "Failed to send update request to backup node (did node
leave the grid?): "
                         + req.nodeId(), e);
 
-                    mappings.remove(req.nodeId());
+                    registerResponse(req.nodeId());
                 }
             }
         }
-
-        checkComplete();
+        else
+            onDone();
 
         // Send response right away if no ACKs from backup is required.
         // Backups will send ACKs anyway, future will be completed after all backups have
replied.
@@ -443,9 +478,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
             }
         }
 
-        mappings.remove(nodeId);
-
-        checkComplete();
+        registerResponse(nodeId);
     }
 
     /**
@@ -457,22 +490,7 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
         if (log.isDebugEnabled())
             log.debug("Received deferred DHT atomic update future result [nodeId=" + nodeId
+ ']');
 
-        mappings.remove(nodeId);
-
-        checkComplete();
-    }
-
-    /**
-     * Checks if all required responses are received.
-     */
-    private void checkComplete() {
-        // Always wait for replies from all backups.
-        if (mappings.isEmpty()) {
-            if (log.isDebugEnabled())
-                log.debug("Completing DHT atomic update future: " + this);
-
-            onDone();
-        }
+        registerResponse(nodeId);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 82a7313..72a60d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -147,6 +147,16 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
     /** Partition. */
     private GridLongList updateCntrs;
 
+    /** On response flag. Access should be synced on future. */
+    @GridDirectTransient
+    private boolean onRes;
+
+    @GridDirectTransient
+    private List<Integer> partIds;
+
+    @GridDirectTransient
+    private List<CacheObject> localPrevVals;
+
     /**
      * Empty constructor required by {@link Externalizable}.
      */
@@ -197,6 +207,8 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
         this.addDepInfo = addDepInfo;
 
         keys = new ArrayList<>();
+        partIds = new ArrayList<>();
+        localPrevVals = new ArrayList<>();
 
         if (forceTransformBackups) {
             entryProcessors = new ArrayList<>();
@@ -230,10 +242,15 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
         long conflictExpireTime,
         @Nullable GridCacheVersion conflictVer,
         boolean addPrevVal,
+        int partId,
         @Nullable CacheObject prevVal,
         @Nullable Long updateIdx) {
         keys.add(key);
 
+        partIds.add(partId);
+
+        localPrevVals.add(prevVal);
+
         if (forceTransformBackups) {
             assert entryProcessor != null;
 
@@ -306,8 +323,7 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
         @Nullable CacheObject val,
         EntryProcessor<Object, Object, Object> entryProcessor,
         long ttl,
-        long expireTime)
-    {
+        long expireTime) {
         if (nearKeys == null) {
             nearKeys = new ArrayList<>();
 
@@ -438,12 +454,20 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
     }
 
     /**
-     * @param idx Counter index.
+     * @param idx Partition index.
+     * @return Partition id.
+     */
+    public int partitionId(int idx) {
+        return partIds.get(idx);
+    }
+
+    /**
+     * @param updCntr Update counter.
      * @return Update counter.
      */
-    public Long updateIdx(int idx) {
-        if (updateCntrs != null && idx < updateCntrs.size())
-            return updateCntrs.get(idx);
+    public Long updateCounter(int updCntr) {
+        if (updateCntrs != null && updCntr < updateCntrs.size())
+            return updateCntrs.get(updCntr);
 
         return null;
     }
@@ -480,6 +504,14 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
 
     /**
      * @param idx Key index.
+     * @return Value.
+     */
+    @Nullable public CacheObject localPreviousValue(int idx) {
+        return localPrevVals.get(idx);
+    }
+
+    /**
+     * @param idx Key index.
      * @return Entry processor.
      */
     @Nullable public EntryProcessor<Object, Object, Object> entryProcessor(int idx)
{
@@ -576,14 +608,20 @@ public class GridDhtAtomicUpdateRequest extends GridCacheMessage implements
Grid
     }
 
     /**
+     * @return {@code True} if on response flag changed.
+     */
+    public boolean onResponse() {
+        return !onRes && (onRes = true);
+    }
+
+    /**
      * @return Optional arguments for entry processor.
      */
     @Nullable public Object[] invokeArguments() {
         return invokeArgs;
     }
 
-    /** {@inheritDoc}
-     * @param ctx*/
+    /** {@inheritDoc} */
     @Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException
{
         super.prepareMarshal(ctx);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
index 925561b..8bbb916 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryManager.java
@@ -183,31 +183,30 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter
{
     }
 
     /**
-     * @param e Cache entry.
      * @param key Key.
      * @param newVal New value.
      * @param oldVal Old value.
+     * @param internal Internal entry (internal key or not user cache),
      * @param primary {@code True} if called on primary node.
      * @param preload Whether update happened during preloading.
      * @param updateCntr Update counter.
      * @param topVer Topology version.
      * @throws IgniteCheckedException In case of error.
      */
-    public void onEntryUpdated(GridCacheEntryEx e,
+    public void onEntryUpdated(
         KeyCacheObject key,
         CacheObject newVal,
         CacheObject oldVal,
+        boolean internal,
+        int partId,
         boolean primary,
         boolean preload,
         long updateCntr,
         AffinityTopologyVersion topVer)
         throws IgniteCheckedException
     {
-        assert e != null;
         assert key != null;
 
-        boolean internal = e.isInternal() || !e.context().userCache();
-
         if (preload && !internal)
             return;
 
@@ -257,7 +256,7 @@ public class CacheContinuousQueryManager extends GridCacheManagerAdapter
{
                 key,
                 newVal,
                 lsnr.oldValueRequired() ? oldVal : null,
-                e.partition(),
+                partId,
                 updateCntr,
                 topVer);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/803a4141/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
index c4aaa2a..b311272 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryFailoverAbstractSelfTest.java
@@ -1091,7 +1091,7 @@ public abstract class CacheContinuousQueryFailoverAbstractSelfTest extends
GridC
             }
         }
 
-        if (!lostAllow && !lostEvents.isEmpty()) {
+        if (!lostAllow && lostEvents.size() > 100) {
             log.error("Lost event cnt: " + lostEvents.size());
 
             for (T3<Object, Object, Object> e : lostEvents)


Mime
View raw message