IGNITE-9200: Continuous query notifications can be missed if entry pr…

…ocessor is retried due to binary type registration (cherry picked from commit 9d898b9)
apache · Aug 10, 2018 · e21965d · e21965d
1 parent 93366ed
commit e21965d
Show file tree

Hide file tree

Showing 3 changed files with 185 additions and 62 deletions.
diff --git a/...apache/ignite/internal/processors/cache/distributed/dht/atomic/DhtAtomicUpdateResult.java b/...apache/ignite/internal/processors/cache/distributed/dht/atomic/DhtAtomicUpdateResult.java
@@ -44,6 +44,12 @@ class DhtAtomicUpdateResult {
     /** */
     private IgniteCacheExpiryPolicy expiry;
 
+    /**
+     * If batch update was interrupted in the middle, it should be continued from processedEntriesCount to avoid
+     * extra update closure invocation.
+     */
+    private int processedEntriesCount = 0;
+
     /**
      *
      */
@@ -97,10 +103,19 @@ void addDeleted(GridDhtCacheEntry entry,
     /**
      * @return Deleted entries.
      */
-    Collection<IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion>> deleted() {
+    public Collection<IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion>> deleted() {
         return deleted;
     }
 
+    /**
+     * Sets deleted entries.
+     *
+     * @param deleted deleted entries.
+     */
+    void deleted(Collection<IgniteBiTuple<GridDhtCacheEntry, GridCacheVersion>> deleted) {
+        this.deleted = deleted;
+    }
+
     /**
      * @return DHT future.
      */
@@ -128,4 +143,20 @@ GridCacheReturn returnValue() {
     void dhtFuture(@Nullable GridDhtAtomicAbstractUpdateFuture dhtFut) {
         this.dhtFut = dhtFut;
     }
+
+    /**
+     * Sets processed entries count.
+     * @param idx processed entries count.
+     */
+    public void processedEntriesCount(int idx) {
+        processedEntriesCount = idx;
+    }
+
+    /**
+     * Returns processed entries count.
+     * @return processed entries count.
+     */
+    public int processedEntriesCount() {
+        return processedEntriesCount;
+    }
 }