Repository: ignite
Updated Branches:
  refs/heads/ignite-2.5 37fc72542 -> 76e293654


IGNITE-8114 Add fail recovery mechanism to tracking pages - Fixes #3734.

Signed-off-by: dpavlov <dpav...@apache.org>

(cherry picked from commit 0829397)


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

Branch: refs/heads/ignite-2.5
Commit: 76e293654e34c927d6c9efc85a12e736b58a21f2
Parents: 37fc725
Author: Eduard Shangareev <eshangar...@gridgain.com>
Authored: Fri Apr 6 19:22:07 2018 +0300
Committer: dpavlov <dpav...@apache.org>
Committed: Fri Apr 6 19:26:16 2018 +0300

----------------------------------------------------------------------
 .../TrackingPageIsCorruptedException.java       |  60 +++++++
 .../cache/persistence/tree/io/PageMetaIO.java   |   6 +-
 .../persistence/tree/io/TrackingPageIO.java     | 156 +++++++++++++++----
 .../persistence/tree/io/TrackingPageIOTest.java | 116 +++++++++++---
 4 files changed, 289 insertions(+), 49 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/76e29365/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/TrackingPageIsCorruptedException.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/TrackingPageIsCorruptedException.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/TrackingPageIsCorruptedException.java
new file mode 100644
index 0000000..2b82bff
--- /dev/null
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/TrackingPageIsCorruptedException.java
@@ -0,0 +1,60 @@
+/*
+ * 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.persistence.snapshot;
+
+import org.apache.ignite.IgniteCheckedException;
+
+/**
+ * Thrown when corrupted tracking page was queried.
+ */
+public class TrackingPageIsCorruptedException extends IgniteCheckedException {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Instance. */
+    public static final TrackingPageIsCorruptedException INSTANCE = new 
TrackingPageIsCorruptedException(-1, -1);
+
+    /** Last tag. */
+    private final long lastTag;
+
+    /** Passed tag. */
+    private final long passedTag;
+
+    /**
+     * @param lastTag Last tag.
+     * @param passedTag Passed tag.
+     */
+    public TrackingPageIsCorruptedException(long lastTag, long passedTag) {
+        this.lastTag = lastTag;
+        this.passedTag = passedTag;
+    }
+
+    /**
+     * @return Last tag.
+     */
+    public long lastTag() {
+        return lastTag;
+    }
+
+    /**
+     * @return Passed tag.
+     */
+    public long passedTag() {
+        return passedTag;
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/76e29365/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageMetaIO.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageMetaIO.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageMetaIO.java
index d2921ee..d676cfd 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageMetaIO.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/PageMetaIO.java
@@ -152,10 +152,10 @@ public class PageMetaIO extends PageIO {
 
     /**
      * @param pageAddr Page address.
-     * @param nextSnapshotId Next snapshot id.
+     * @param nextSnapshotTag Next snapshot tag.
      */
-    public void setNextSnapshotTag(long pageAddr, long nextSnapshotId) {
-        PageUtils.putLong(pageAddr, NEXT_SNAPSHOT_TAG_OFF, nextSnapshotId);
+    public void setNextSnapshotTag(long pageAddr, long nextSnapshotTag) {
+        PageUtils.putLong(pageAddr, NEXT_SNAPSHOT_TAG_OFF, nextSnapshotTag);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/76e29365/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIO.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIO.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIO.java
index 1bd70f8..94885e4 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIO.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIO.java
@@ -20,6 +20,7 @@ package 
org.apache.ignite.internal.processors.cache.persistence.tree.io;
 import java.nio.ByteBuffer;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.pagemem.PageIdUtils;
+import 
org.apache.ignite.internal.processors.cache.persistence.snapshot.TrackingPageIsCorruptedException;
 import 
org.apache.ignite.internal.processors.cache.persistence.tree.util.PageHandler;
 import org.apache.ignite.internal.util.GridStringBuilder;
 import org.apache.ignite.internal.util.GridUnsafe;
@@ -50,6 +51,12 @@ public class TrackingPageIO extends PageIO {
         new TrackingPageIO(1)
     );
 
+    /** Corrupt flag mask. */
+    public static final long CORRUPT_FLAG_MASK = 1L << 63;
+
+    /** Corrupt flag mask. */
+    public static final long CORRUPT_FLAG_FILTER_MASK = ~CORRUPT_FLAG_MASK;
+
     /** Last snapshot offset. */
     public static final int LAST_SNAPSHOT_TAG_OFFSET = COMMON_HEADER_END;
 
@@ -77,11 +84,11 @@ public class TrackingPageIO extends PageIO {
      *
      * @param buf Buffer.
      * @param pageId Page id.
-     * @param nextSnapshotTag tag of next snapshot.
+     * @param nextSnapshotTag Tag of next snapshot.
      * @param pageSize Page size.
      */
-    public boolean markChanged(ByteBuffer buf, long pageId, long 
nextSnapshotTag, long lastSuccessfulSnapshotTag, int pageSize) {
-        validateSnapshotId(buf, nextSnapshotTag, lastSuccessfulSnapshotTag, 
pageSize);
+    public void markChanged(ByteBuffer buf, long pageId, long nextSnapshotTag, 
long lastSuccessfulSnapshotTag, int pageSize) {
+        validateSnapshotTag(buf, nextSnapshotTag, lastSuccessfulSnapshotTag, 
pageSize);
 
         int cntOfPage = countOfPageToTrack(pageSize);
 
@@ -98,7 +105,7 @@ public class TrackingPageIO extends PageIO {
         byte newVal =  (byte) (byteToUpdate | updateTemplate);
 
         if (byteToUpdate == newVal)
-            return false;
+            return;
 
         buf.put(idx, newVal);
 
@@ -107,8 +114,6 @@ public class TrackingPageIO extends PageIO {
         buf.putShort(sizeOff, newSize);
 
         assert newSize == countOfChangedPage(buf, nextSnapshotTag, pageSize);
-
-        return true;
     }
 
     /**
@@ -116,22 +121,30 @@ public class TrackingPageIO extends PageIO {
      * @param nextSnapshotTag Next snapshot id.
      * @param lastSuccessfulSnapshotTag Last successful snapshot id.
      * @param pageSize Page size.
+     *
+     * @return <code>-1</code> if everything is ok, otherwise last saved tag.
      */
-    private void validateSnapshotId(ByteBuffer buf, long nextSnapshotTag, long 
lastSuccessfulSnapshotTag, int pageSize) {
+    private long validateSnapshotTag(ByteBuffer buf, long nextSnapshotTag, 
long lastSuccessfulSnapshotTag, int pageSize) {
         assert nextSnapshotTag != lastSuccessfulSnapshotTag : "nextSnapshotTag 
= " + nextSnapshotTag +
             ", lastSuccessfulSnapshotId = " + lastSuccessfulSnapshotTag;
 
         long last = getLastSnapshotTag(buf);
 
-        assert last <= nextSnapshotTag : "last = " + last + ", nextSnapshotTag 
= " + nextSnapshotTag;
+        if(last > nextSnapshotTag) { //we have lost snapshot tag therefore 
should mark this tracking as corrupted
+            PageHandler.zeroMemory(buf, LAST_SNAPSHOT_TAG_OFFSET, 
buf.capacity() - LAST_SNAPSHOT_TAG_OFFSET);
+
+            setLastSnasphotTag(buf, nextSnapshotTag | CORRUPT_FLAG_MASK);
+
+            return last;
+        }
 
         if (nextSnapshotTag == last) //everything is ok
-            return;
+            return -1;
 
         int cntOfPage = countOfPageToTrack(pageSize);
 
         if (last <= lastSuccessfulSnapshotTag) { //we can drop our data
-            buf.putLong(LAST_SNAPSHOT_TAG_OFFSET, nextSnapshotTag);
+            setLastSnasphotTag(buf, nextSnapshotTag);
 
             PageHandler.zeroMemory(buf, SIZE_FIELD_OFFSET, buf.capacity() - 
SIZE_FIELD_OFFSET);
         } else { //we can't drop data, it is still necessary for incremental 
snapshots
@@ -167,37 +180,119 @@ public class TrackingPageIO extends PageIO {
                 buf.putShort(sizeOff2, (short)newSize);
             }
 
-            buf.putLong(LAST_SNAPSHOT_TAG_OFFSET, nextSnapshotTag);
+            setLastSnasphotTag(buf, nextSnapshotTag);
 
             PageHandler.zeroMemory(buf, sizeOff, len + SIZE_FIELD_SIZE);
         }
+
+        return -1;
+    }
+
+    /**
+     * @param buf Buffer.
+     * @param nextSnapshotTag Next snapshot tag.
+     */
+    private void setLastSnasphotTag(ByteBuffer buf, long nextSnapshotTag) {
+        if (isCorrupted(buf))
+            nextSnapshotTag = nextSnapshotTag | CORRUPT_FLAG_MASK;
+
+        buf.putLong(LAST_SNAPSHOT_TAG_OFFSET, nextSnapshotTag);
     }
 
     /**
      * @param buf Buffer.
+     * @param nextSnapshotTag Next snapshot tag.
      */
-    long getLastSnapshotTag(ByteBuffer buf) {
-        return buf.getLong(LAST_SNAPSHOT_TAG_OFFSET);
+    private void setLastSnasphotTag0(ByteBuffer buf, long nextSnapshotTag) {
+        buf.putLong(LAST_SNAPSHOT_TAG_OFFSET, nextSnapshotTag);
     }
 
     /**
-     * @param addr Address.
+     * @param addr address.
+     * @param nextSnapshotTag Next snapshot tag.
      */
-    long getLastSnapshotTag(long addr) {
+    private void setLastSnasphotTag0(long addr, long nextSnapshotTag) {
+        GridUnsafe.putLong(addr + LAST_SNAPSHOT_TAG_OFFSET, nextSnapshotTag);
+    }
+
+    /**
+     * @param buf Buffer.
+     * @return Saved max seen snapshot tag.
+     */
+    private long getLastSnapshotTag(ByteBuffer buf) {
+        return getLastSnapshotTag0(buf) & CORRUPT_FLAG_FILTER_MASK;
+    }
+
+    /**
+     * @param buf Buffer.
+     * @return Saved value in {@link TrackingPageIO#LAST_SNAPSHOT_TAG_OFFSET}.
+     */
+    private long getLastSnapshotTag0(ByteBuffer buf) {
+        return buf.getLong(LAST_SNAPSHOT_TAG_OFFSET) ;
+    }
+
+    /**
+     * @param addr Address of buffer start.
+     * @return Saved max seen snapshot tag.
+     */
+    private long getLastSnapshotTag(long addr) {
+        return getLastSnapshotTag0(addr) & CORRUPT_FLAG_FILTER_MASK;
+    }
+
+    /**
+     * @param addr Address of buffer start.
+     * @return Saved value in {@link TrackingPageIO#LAST_SNAPSHOT_TAG_OFFSET}.
+     */
+    private long getLastSnapshotTag0(long addr) {
         return GridUnsafe.getLong(addr + LAST_SNAPSHOT_TAG_OFFSET);
     }
 
     /**
-     * Check that pageId was marked as changed between previous snapshot 
finish and current snapshot start.
+     * @param buf Buffer.
+     * @return Was tracking page marked as corrupted or not.
+     */
+    public boolean isCorrupted(ByteBuffer buf) {
+        return getLastSnapshotTag0(buf) < 0; //if true it means that first bit 
set to 1
+    }
+
+    /**
+     * Reset corrupted flag to false.
+     *
+     * @param buf Buffer.
+     */
+    public void resetCorruptFlag(ByteBuffer buf) {
+        setLastSnasphotTag0(buf, getLastSnapshotTag(buf));
+    }
+
+    /**
+     * Reset corrupted flag to false.
+     *
+     * @param addr Buffer.
+     */
+    public void resetCorruptFlag(long addr) {
+        setLastSnasphotTag0(addr, getLastSnapshotTag(addr));
+    }
+
+    /**
      *
      * @param buf Buffer.
      * @param pageId Page id.
      * @param curSnapshotTag Snapshot tag.
      * @param lastSuccessfulSnapshotTag Last successful snapshot id.
      * @param pageSize Page size.
+     *
+     * @return Was that pageId marked as changed between previous snapshot 
finish and current snapshot start or not.
+     * @throws TrackingPageIsCorruptedException if this tracking page was 
marked as corrupted.
      */
-    public boolean wasChanged(ByteBuffer buf, long pageId, long 
curSnapshotTag, long lastSuccessfulSnapshotTag, int pageSize) {
-        validateSnapshotId(buf, curSnapshotTag + 1, lastSuccessfulSnapshotTag, 
pageSize);
+    public boolean wasChanged(ByteBuffer buf, long pageId, long 
curSnapshotTag, long lastSuccessfulSnapshotTag, int pageSize)
+        throws TrackingPageIsCorruptedException {
+        if (isCorrupted(buf))
+            throw TrackingPageIsCorruptedException.INSTANCE;
+
+        long lastTag = validateSnapshotTag(buf, curSnapshotTag + 1, 
lastSuccessfulSnapshotTag, pageSize);
+
+        if (lastTag >= 0)
+            throw new TrackingPageIsCorruptedException(lastTag, 
curSnapshotTag);
 
         if (countOfChangedPage(buf, curSnapshotTag, pageSize) < 1)
             return false;
@@ -223,7 +318,7 @@ public class TrackingPageIO extends PageIO {
      * @param snapshotTag Snapshot tag.
      * @param pageSize Page size.
      *
-     * @return count of pages which were marked as change for given snapshotTag
+     * @return Count of pages which were marked as change for given 
snapshotTag.
      */
     public short countOfChangedPage(ByteBuffer buf, long snapshotTag, int 
pageSize) {
         long dif = getLastSnapshotTag(buf) - snapshotTag;
@@ -240,16 +335,16 @@ public class TrackingPageIO extends PageIO {
     /**
      * @param snapshotTag Snapshot id.
      *
-     * @return true if snapshotTag is odd, otherwise - false
+     * @return true if snapshotTag is odd, otherwise - false.
      */
-    boolean useLeftHalf(long snapshotTag) {
+    private boolean useLeftHalf(long snapshotTag) {
         return (snapshotTag & 0b1) == 0;
     }
 
     /**
      * @param pageId Page id.
      * @param pageSize Page size.
-     * @return pageId of tracking page which set pageId belongs to
+     * @return Page id of tracking page which set pageId belongs to.
      */
     public long trackingPageFor(long pageId, int pageSize) {
         assert PageIdUtils.pageIndex(pageId) > 0;
@@ -267,7 +362,7 @@ public class TrackingPageIO extends PageIO {
     /**
      * @param pageSize Page size.
      *
-     * @return how many page we can track with 1 page
+     * @return How many page we can track with 1 page.
      */
     public int countOfPageToTrack(int pageSize) {
         return ((pageSize - SIZE_FIELD_OFFSET) / 2 - SIZE_FIELD_SIZE)  << 3;
@@ -279,11 +374,18 @@ public class TrackingPageIO extends PageIO {
      * @param curSnapshotTag Snapshot id.
      * @param lastSuccessfulSnapshotTag  Last successful snapshot id.
      * @param pageSize Page size.
-     * @return set pageId if it was changed or next closest one, if there is 
no changed page {@code null} will be returned
+     * @return Passed pageId if it was changed or next closest one, if there 
is no changed page {@code null} will be returned.
+     * @throws TrackingPageIsCorruptedException if this tracking page was 
marked as corrupted.
      */
     @Nullable public Long findNextChangedPage(ByteBuffer buf, long start, long 
curSnapshotTag,
-        long lastSuccessfulSnapshotTag, int pageSize) {
-        validateSnapshotId(buf, curSnapshotTag + 1, lastSuccessfulSnapshotTag, 
pageSize);
+        long lastSuccessfulSnapshotTag, int pageSize) throws 
TrackingPageIsCorruptedException {
+        if (isCorrupted(buf))
+            throw TrackingPageIsCorruptedException.INSTANCE;
+
+        long lastTag = validateSnapshotTag(buf, curSnapshotTag + 1, 
lastSuccessfulSnapshotTag, pageSize);
+
+        if (lastTag >= 0)
+            throw new TrackingPageIsCorruptedException(lastTag, 
curSnapshotTag);
 
         int cntOfPage = countOfPageToTrack(pageSize);
 
@@ -332,6 +434,8 @@ public class TrackingPageIO extends PageIO {
     /**
      * @param byteToTest Byte to test.
      * @param firstBitToTest First bit to test.
+     *
+     * @return Index of bit which is set to 1, if there is no such index then 
<code>-1</code>.
      */
     private static int foundSetBit(byte byteToTest, int firstBitToTest) {
         assert firstBitToTest < 8;

http://git-wip-us.apache.org/repos/asf/ignite/blob/76e29365/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIOTest.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIOTest.java
 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIOTest.java
index b50f026..cacea48 100644
--- 
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIOTest.java
+++ 
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/TrackingPageIOTest.java
@@ -26,14 +26,19 @@ import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.ThreadLocalRandom;
 import junit.framework.TestCase;
+import org.apache.ignite.internal.pagemem.PageIdAllocator;
+import org.apache.ignite.internal.pagemem.PageIdUtils;
+import 
org.apache.ignite.internal.processors.cache.persistence.snapshot.TrackingPageIsCorruptedException;
 import org.apache.ignite.internal.util.GridUnsafe;
+import org.jetbrains.annotations.NotNull;
 
 /**
  *
  */
 public class TrackingPageIOTest extends TestCase {
     /** Page size. */
-    public static final int PAGE_SIZE = 2048;
+    public static final int PAGE_SIZE = 4096;
+
 
     /** */
     private final TrackingPageIO io = TrackingPageIO.VERSIONS.latest();
@@ -41,9 +46,8 @@ public class TrackingPageIOTest extends TestCase {
     /**
      *
      */
-    public void testBasics() {
-        ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
-        buf.order(ByteOrder.nativeOrder());
+    public void testBasics() throws Exception {
+        ByteBuffer buf = createBuffer();
 
         io.markChanged(buf, 2, 0, -1, PAGE_SIZE);
 
@@ -55,13 +59,21 @@ public class TrackingPageIOTest extends TestCase {
     }
 
     /**
-     *
+     * @return byte buffer with right order.
      */
-    public void testMarkingRandomly() {
+    @NotNull private ByteBuffer createBuffer() {
         ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
+
         buf.order(ByteOrder.nativeOrder());
 
-        int cntOfPageToTrack = io.countOfPageToTrack(PAGE_SIZE);
+        return buf;
+    }
+
+    /**
+     *
+     */
+    public void testMarkingRandomly() throws Exception {
+        ByteBuffer buf = createBuffer();
 
         for (int i = 0; i < 1001; i++)
             checkMarkingRandomly(buf, i, false);
@@ -70,9 +82,8 @@ public class TrackingPageIOTest extends TestCase {
     /**
      *
      */
-    public void testZeroingRandomly() {
-        ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
-        buf.order(ByteOrder.nativeOrder());
+    public void testZeroingRandomly() throws Exception {
+        ByteBuffer buf = createBuffer();
 
         for (int i = 0; i < 1001; i++)
             checkMarkingRandomly(buf, i, true);
@@ -82,7 +93,7 @@ public class TrackingPageIOTest extends TestCase {
      * @param buf Buffer.
      * @param backupId Backup id.
      */
-    private void checkMarkingRandomly(ByteBuffer buf, int backupId, boolean 
testZeroing) {
+    private void checkMarkingRandomly(ByteBuffer buf, int backupId, boolean 
testZeroing) throws Exception {
         ThreadLocalRandom rand = ThreadLocalRandom.current();
 
         int track = io.countOfPageToTrack(PAGE_SIZE);
@@ -132,8 +143,7 @@ public class TrackingPageIOTest extends TestCase {
      * @throws Exception If failed.
      */
     public void testFindNextChangedPage() throws Exception {
-        ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
-        buf.order(ByteOrder.nativeOrder());
+        ByteBuffer buf = createBuffer();
 
         for (int i = 0; i < 101; i++)
             checkFindingRandomly(buf, i);
@@ -143,7 +153,7 @@ public class TrackingPageIOTest extends TestCase {
      * @param buf Buffer.
      * @param backupId Backup id.
      */
-    private void checkFindingRandomly(ByteBuffer buf, int backupId) {
+    private void checkFindingRandomly(ByteBuffer buf, int backupId) throws 
Exception {
         ThreadLocalRandom rand = ThreadLocalRandom.current();
 
         int track = io.countOfPageToTrack(PAGE_SIZE);
@@ -187,9 +197,8 @@ public class TrackingPageIOTest extends TestCase {
     /**
      *
      */
-    public void testMerging() {
-        ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
-        buf.order(ByteOrder.nativeOrder());
+    public void testMerging() throws Exception {
+        ByteBuffer buf = createBuffer();
 
         ThreadLocalRandom rand = ThreadLocalRandom.current();
 
@@ -226,9 +235,8 @@ public class TrackingPageIOTest extends TestCase {
     /**
      *
      */
-    public void testMerging_MarksShouldBeDropForSuccessfulBackup() {
-        ByteBuffer buf = ByteBuffer.allocateDirect(PAGE_SIZE);
-        buf.order(ByteOrder.nativeOrder());
+    public void testMerging_MarksShouldBeDropForSuccessfulBackup() throws 
Exception {
+        ByteBuffer buf = createBuffer();
 
         ThreadLocalRandom rand = ThreadLocalRandom.current();
 
@@ -260,6 +268,15 @@ public class TrackingPageIOTest extends TestCase {
             assertEquals("pageId = " + i, setIdx2.contains(i), 
io.wasChanged(buf, i, 5, 4, PAGE_SIZE));
     }
 
+    /**
+     * @param buf Buffer.
+     * @param track Track.
+     * @param basePageId Base page id.
+     * @param maxPageId Max page id.
+     * @param setIdx Set index.
+     * @param backupId Backup id.
+     * @param successfulBackupId Successful backup id.
+     */
     private void generateMarking(
         ByteBuffer buf,
         int track,
@@ -280,4 +297,63 @@ public class TrackingPageIOTest extends TestCase {
             }
         }
     }
+
+    /**
+     * We should handle case when we lost snapshot tag and now it's lower than 
saved.
+     *
+     * @throws Exception if failed.
+     */
+    public void testThatWeDontFailIfSnapshotTagWasLost() throws Exception {
+        ByteBuffer buf = createBuffer();
+
+        long basePageId = PageIdUtils.pageId(0, PageIdAllocator.FLAG_IDX, 1);
+
+        assert basePageId >= 0;
+
+        PageIO.setPageId(GridUnsafe.bufferAddress(buf), basePageId);
+
+        int oldTag = 10;
+
+        io.markChanged(buf, basePageId + 1, oldTag, oldTag - 1, PAGE_SIZE);
+
+        for (int i = 1; i < 100; i++)
+            io.markChanged(buf, basePageId + i, oldTag - 1, oldTag - 2, 
PAGE_SIZE);
+
+        assertTrue(io.isCorrupted(buf));
+
+        for (int i = 1; i < 100; i++) {
+            try {
+                long id = basePageId + i;
+
+                io.wasChanged(buf, id, oldTag - 1, oldTag - 2, PAGE_SIZE);
+
+                fail();
+            }
+            catch (TrackingPageIsCorruptedException e) {
+                //ignore
+            }
+        }
+
+        for (int i = 1; i < 100; i++) {
+            long id = basePageId + i + 1000;
+
+            io.markChanged(buf, id, oldTag, oldTag - 2, PAGE_SIZE);
+        }
+
+        io.resetCorruptFlag(buf);
+
+        assertFalse(io.isCorrupted(buf));
+
+        for (int i = 1; i < 100; i++) {
+            long id = basePageId + i + 1000;
+
+            assertTrue(io.wasChanged(buf, id, oldTag, oldTag - 1, PAGE_SIZE));
+        }
+
+        for (int i = 1; i < 100; i++) {
+            long id = basePageId + i;
+
+            assertFalse(io.wasChanged(buf, id, oldTag, oldTag - 1, PAGE_SIZE));
+        }
+    }
 }
\ No newline at end of file

Reply via email to