asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mhub...@apache.org
Subject [2/3] asterixdb git commit: [NO ISSUE][STO] Introduce Index Checkpoints
Date Thu, 07 Dec 2017 23:09:26 GMT
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/AbstractLSMIOOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/AbstractLSMIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/AbstractLSMIOOperationCallbackTest.java
index 0f2ea50..1c57d1b 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/AbstractLSMIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/AbstractLSMIOOperationCallbackTest.java
@@ -21,6 +21,8 @@ package org.apache.asterix.test.ioopcallbacks;
 
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback;
+import org.apache.asterix.common.storage.IIndexCheckpointManager;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
@@ -45,8 +47,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(Mockito.mock(AbstractLSMMemoryComponent.class));
-        LSMBTreeIOOperationCallback callback =
-                new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
 
         //request to flush first component
         callback.updateLastLSN(1);
@@ -57,13 +59,14 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         callback.beforeOperation(LSMIOOperationType.FLUSH);
 
         Assert.assertEquals(1, callback.getComponentLSN(null));
-        callback.afterOperation(LSMIOOperationType.FLUSH, null, mockDiskComponent());
-        callback.afterFinalize(LSMIOOperationType.FLUSH, mockDiskComponent());
+        final ILSMDiskComponent diskComponent1 = mockDiskComponent();
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent1);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent1);
 
         Assert.assertEquals(2, callback.getComponentLSN(null));
-
-        callback.afterOperation(LSMIOOperationType.FLUSH, null, mockDiskComponent());
-        callback.afterFinalize(LSMIOOperationType.FLUSH, mockDiskComponent());
+        final ILSMDiskComponent diskComponent2 = mockDiskComponent();
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent2);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent2);
     }
 
     @Test
@@ -72,8 +75,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(Mockito.mock(AbstractLSMMemoryComponent.class));
 
-        LSMBTreeIOOperationCallback callback =
-                new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
 
         //request to flush first component
         callback.updateLastLSN(1);
@@ -90,11 +93,13 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         //the scheduleFlush request would fail this time
 
         Assert.assertEquals(1, callback.getComponentLSN(null));
-        callback.afterOperation(LSMIOOperationType.FLUSH, null, mockDiskComponent());
-        callback.afterFinalize(LSMIOOperationType.FLUSH, mockDiskComponent());
-
+        final ILSMDiskComponent diskComponent1 = mockDiskComponent();
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent1);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent1);
+        final ILSMDiskComponent diskComponent2 = mockDiskComponent();
         Assert.assertEquals(2, callback.getComponentLSN(null));
-        callback.afterFinalize(LSMIOOperationType.FLUSH, mockDiskComponent());
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent2);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent2);
     }
 
     @Test
@@ -103,9 +108,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(Mockito.mock(AbstractLSMMemoryComponent.class));
 
-        LSMBTreeIOOperationCallback callback =
-                new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
-
+        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
         //request to flush first component
         callback.updateLastLSN(1);
         callback.beforeOperation(LSMIOOperationType.FLUSH);
@@ -144,7 +148,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
 
-        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, idGenerator);
+        LSMBTreeIOOperationCallback callback =
+                new LSMBTreeIOOperationCallback(mockIndex, idGenerator, mockIndexCheckpointManagerProvider());
 
         ILSMComponentId initialId = idGenerator.getId();
         // simulate a partition is flushed before allocated
@@ -162,7 +167,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
         ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
-        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, idGenerator);
+        LSMBTreeIOOperationCallback callback =
+                new LSMBTreeIOOperationCallback(mockIndex, idGenerator, mockIndexCheckpointManagerProvider());
 
         ILSMComponentId id = idGenerator.getId();
         callback.allocated(mockComponent);
@@ -178,8 +184,9 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
             callback.beforeOperation(LSMIOOperationType.FLUSH);
             callback.recycled(mockComponent, true);
 
-            callback.afterOperation(LSMIOOperationType.FLUSH, null, mockDiskComponent());
-            callback.afterFinalize(LSMIOOperationType.FLUSH, mockDiskComponent());
+            final ILSMDiskComponent diskComponent = mockDiskComponent();
+            callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent);
+            callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent);
             checkMemoryComponent(expectedId, mockComponent);
         }
     }
@@ -191,7 +198,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
         ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
-        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, idGenerator);
+        LSMBTreeIOOperationCallback callback =
+                new LSMBTreeIOOperationCallback(mockIndex, idGenerator, mockIndexCheckpointManagerProvider());
 
         ILSMComponentId id = idGenerator.getId();
         callback.allocated(mockComponent);
@@ -216,7 +224,8 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         ILSMMemoryComponent mockComponent = Mockito.mock(AbstractLSMMemoryComponent.class);
         Mockito.when(mockIndex.getCurrentMemoryComponent()).thenReturn(mockComponent);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
-        LSMBTreeIOOperationCallback callback = new LSMBTreeIOOperationCallback(mockIndex, idGenerator);
+        LSMBTreeIOOperationCallback callback =
+                new LSMBTreeIOOperationCallback(mockIndex, idGenerator, mockIndexCheckpointManagerProvider());
 
         ILSMComponentId id = idGenerator.getId();
         callback.allocated(mockComponent);
@@ -230,7 +239,9 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
 
         callback.updateLastLSN(0);
         callback.beforeOperation(LSMIOOperationType.FLUSH);
-        callback.afterFinalize(LSMIOOperationType.FLUSH, Mockito.mock(ILSMDiskComponent.class));
+        final ILSMDiskComponent diskComponent = mockDiskComponent();
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent);
 
         // another flush is to be scheduled before the component is recycled
         idGenerator.refresh();
@@ -243,7 +254,9 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         // schedule the next flush
         callback.updateLastLSN(0);
         callback.beforeOperation(LSMIOOperationType.FLUSH);
-        callback.afterFinalize(LSMIOOperationType.FLUSH, Mockito.mock(ILSMDiskComponent.class));
+        final ILSMDiskComponent diskComponent2 = mockDiskComponent();
+        callback.afterOperation(LSMIOOperationType.FLUSH, null, diskComponent2);
+        callback.afterFinalize(LSMIOOperationType.FLUSH, diskComponent2);
         callback.recycled(mockComponent, true);
         checkMemoryComponent(nextId, mockComponent);
     }
@@ -263,5 +276,14 @@ public abstract class AbstractLSMIOOperationCallbackTest extends TestCase {
         return component;
     }
 
-    protected abstract AbstractLSMIOOperationCallback getIoCallback();
+    protected IIndexCheckpointManagerProvider mockIndexCheckpointManagerProvider() throws HyracksDataException {
+        IIndexCheckpointManagerProvider indexCheckpointManagerProvider =
+                Mockito.mock(IIndexCheckpointManagerProvider.class);
+        IIndexCheckpointManager indexCheckpointManager = Mockito.mock(IIndexCheckpointManager.class);
+        Mockito.doNothing().when(indexCheckpointManager).flushed(Mockito.any(), Mockito.anyLong());
+        Mockito.doReturn(indexCheckpointManager).when(indexCheckpointManagerProvider).get(Mockito.any());
+        return indexCheckpointManagerProvider;
+    }
+
+    protected abstract AbstractLSMIOOperationCallback getIoCallback() throws HyracksDataException;
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeIOOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeIOOperationCallbackTest.java
index c22e2e3..a4bc399 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeIOOperationCallbackTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.ioopcallbacks;
 
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentIdGenerator;
 import org.mockito.Mockito;
@@ -28,10 +29,11 @@ import org.mockito.Mockito;
 public class LSMBTreeIOOperationCallbackTest extends AbstractLSMIOOperationCallbackTest {
 
     @Override
-    protected AbstractLSMIOOperationCallback getIoCallback() {
+    protected AbstractLSMIOOperationCallback getIoCallback() throws HyracksDataException {
         ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
-        return new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        return new LSMBTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
     }
 
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackTest.java
index 356c80a..5f37c78 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMBTreeWithBuddyIOOperationCallbackTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.ioopcallbacks;
 
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeWithBuddyIOOperationCallback;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentIdGenerator;
 import org.mockito.Mockito;
@@ -28,10 +29,11 @@ import org.mockito.Mockito;
 public class LSMBTreeWithBuddyIOOperationCallbackTest extends AbstractLSMIOOperationCallbackTest {
 
     @Override
-    protected AbstractLSMIOOperationCallback getIoCallback() {
+    protected AbstractLSMIOOperationCallback getIoCallback() throws HyracksDataException {
         ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
-        return new LSMBTreeWithBuddyIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        return new LSMBTreeWithBuddyIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
     }
 
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMInvertedIndexIOOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMInvertedIndexIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMInvertedIndexIOOperationCallbackTest.java
index ac4595e..343bc59 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMInvertedIndexIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMInvertedIndexIOOperationCallbackTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.ioopcallbacks;
 
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMInvertedIndexIOOperationCallback;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentIdGenerator;
 import org.mockito.Mockito;
@@ -28,10 +29,11 @@ import org.mockito.Mockito;
 public class LSMInvertedIndexIOOperationCallbackTest extends AbstractLSMIOOperationCallbackTest {
 
     @Override
-    protected AbstractLSMIOOperationCallback getIoCallback() {
+    protected AbstractLSMIOOperationCallback getIoCallback() throws HyracksDataException {
         ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
-        return new LSMInvertedIndexIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        return new LSMInvertedIndexIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
     }
 
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMRTreeIOOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMRTreeIOOperationCallbackTest.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMRTreeIOOperationCallbackTest.java
index 0131e3f..10d95d8 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMRTreeIOOperationCallbackTest.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/LSMRTreeIOOperationCallbackTest.java
@@ -21,6 +21,7 @@ package org.apache.asterix.test.ioopcallbacks;
 
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMRTreeIOOperationCallback;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentIdGenerator;
 import org.mockito.Mockito;
@@ -28,10 +29,11 @@ import org.mockito.Mockito;
 public class LSMRTreeIOOperationCallbackTest extends AbstractLSMIOOperationCallbackTest {
 
     @Override
-    protected AbstractLSMIOOperationCallback getIoCallback() {
+    protected AbstractLSMIOOperationCallback getIoCallback() throws HyracksDataException {
         ILSMIndex mockIndex = Mockito.mock(ILSMIndex.class);
         Mockito.when(mockIndex.getNumberOfAllMemoryComponents()).thenReturn(2);
-        return new LSMRTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator());
+        return new LSMRTreeIOOperationCallback(mockIndex, new LSMComponentIdGenerator(),
+                mockIndexCheckpointManagerProvider());
     }
 
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-installer/src/test/java/org/apache/asterix/installer/test/AsterixLifecycleIT.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-installer/src/test/java/org/apache/asterix/installer/test/AsterixLifecycleIT.java b/asterixdb/asterix-installer/src/test/java/org/apache/asterix/installer/test/AsterixLifecycleIT.java
index 814e109..1434629 100644
--- a/asterixdb/asterix-installer/src/test/java/org/apache/asterix/installer/test/AsterixLifecycleIT.java
+++ b/asterixdb/asterix-installer/src/test/java/org/apache/asterix/installer/test/AsterixLifecycleIT.java
@@ -19,43 +19,18 @@
 package org.apache.asterix.installer.test;
 
 import java.io.File;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.logging.Logger;
 
-import org.apache.asterix.event.error.VerificationUtil;
-import org.apache.asterix.event.model.AsterixInstance;
 import org.apache.asterix.event.model.AsterixInstance.State;
-import org.apache.asterix.event.model.AsterixRuntimeState;
-import org.apache.asterix.event.service.ServiceProvider;
-import org.apache.asterix.installer.command.CommandHandler;
-import org.apache.asterix.test.common.TestExecutor;
-import org.apache.asterix.testframework.context.TestCaseContext;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
-import org.junit.FixMethodOrder;
-import org.junit.Test;
-import org.junit.runners.MethodSorters;
-import org.junit.runners.Parameterized.Parameters;
-
-@FixMethodOrder(MethodSorters.NAME_ASCENDING)
 
 public class AsterixLifecycleIT {
 
-    private static final int NUM_NC = 2;
-    private static final CommandHandler cmdHandler = new CommandHandler();
-    private static final String PATH_BASE = "src/test/resources/integrationts/lifecycle";
     private static final String PATH_ACTUAL = "target" + File.separator + "ittest" + File.separator;
-    private static final Logger LOGGER = Logger.getLogger(AsterixLifecycleIT.class.getName());
-    private static List<TestCaseContext> testCaseCollection;
-    private final TestExecutor testExecutor = new TestExecutor();
 
     @BeforeClass
     public static void setUp() throws Exception {
         AsterixInstallerIntegrationUtil.init(AsterixInstallerIntegrationUtil.LOCAL_CLUSTER_PATH);
-        TestCaseContext.Builder b = new TestCaseContext.Builder();
-        testCaseCollection = b.build(new File(PATH_BASE));
         File outdir = new File(PATH_ACTUAL);
         outdir.mkdirs();
     }
@@ -70,88 +45,8 @@ public class AsterixLifecycleIT {
         }
     }
 
-    @Parameters
-    public static Collection<Object[]> tests() throws Exception {
-        Collection<Object[]> testArgs = new ArrayList<>();
-        return testArgs;
-    }
-
     public static void restartInstance() throws Exception {
         AsterixInstallerIntegrationUtil.transformIntoRequiredState(State.INACTIVE);
         AsterixInstallerIntegrationUtil.transformIntoRequiredState(State.ACTIVE);
     }
-
-    @Test
-    public void test_1_StopActiveInstance() throws Exception {
-        try {
-            LOGGER.info("Starting test: test_1_StopActiveInstance");
-            AsterixInstallerIntegrationUtil.transformIntoRequiredState(State.ACTIVE);
-            String command = "stop -n " + AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME;
-            cmdHandler.processCommand(command.split(" "));
-            Thread.sleep(4000);
-            AsterixInstance instance = ServiceProvider.INSTANCE.getLookupService()
-                    .getAsterixInstance(AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME);
-            AsterixRuntimeState state = VerificationUtil.getAsterixRuntimeState(instance);
-            assert (state.getFailedNCs().size() == NUM_NC && !state.isCcRunning());
-            LOGGER.info("PASSED: test_1_StopActiveInstance");
-        } catch (Exception e) {
-            throw new Exception("Test configure installer " + "\" FAILED!", e);
-        }
-    }
-
-    @Test
-    public void test_2_StartActiveInstance() throws Exception {
-        try {
-            LOGGER.info("Starting test: test_2_StartActiveInstance");
-            AsterixInstallerIntegrationUtil.transformIntoRequiredState(State.INACTIVE);
-            String command = "start -n " + AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME;
-            cmdHandler.processCommand(command.split(" "));
-            AsterixInstance instance = ServiceProvider.INSTANCE.getLookupService()
-                    .getAsterixInstance(AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME);
-            AsterixRuntimeState state = VerificationUtil.getAsterixRuntimeState(instance);
-            assert (state.getFailedNCs().size() == 0 && state.isCcRunning());
-            LOGGER.info("PASSED: test_2_StartActiveInstance");
-        } catch (Exception e) {
-            throw new Exception("Test configure installer " + "\" FAILED!", e);
-        }
-    }
-
-    @Test
-    public void test_3_DeleteActiveInstance() throws Exception {
-        try {
-            LOGGER.info("Starting test: test_3_DeleteActiveInstance");
-            AsterixInstallerIntegrationUtil.transformIntoRequiredState(State.INACTIVE);
-            String command = "delete -n " + AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME;
-            cmdHandler.processCommand(command.split(" "));
-            AsterixInstance instance = ServiceProvider.INSTANCE.getLookupService()
-                    .getAsterixInstance(AsterixInstallerIntegrationUtil.ASTERIX_INSTANCE_NAME);
-            assert (instance == null);
-            LOGGER.info("PASSED: test_3_DeleteActiveInstance");
-        } catch (Exception e) {
-            throw new Exception("Test delete active instance " + "\" FAILED!", e);
-        } finally {
-            // recreate instance
-            AsterixInstallerIntegrationUtil.createInstance();
-        }
-    }
-
-    @Test
-    public void test() throws Exception {
-        for (TestCaseContext testCaseCtx : testCaseCollection) {
-            testExecutor.executeTest(PATH_ACTUAL, testCaseCtx, null, false);
-        }
-    }
-
-    public static void main(String[] args) throws Exception {
-        try {
-            setUp();
-            new AsterixLifecycleIT().test();
-        } catch (Exception e) {
-            e.printStackTrace();
-            LOGGER.info("TEST CASE(S) FAILED");
-        } finally {
-            tearDown();
-        }
-    }
-
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogMapping.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogMapping.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogMapping.java
index 3b2aff7..0444952 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogMapping.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogMapping.java
@@ -27,10 +27,6 @@ public class RemoteLogMapping {
     private long localLSN;
     public AtomicInteger numOfFlushedIndexes = new AtomicInteger();
 
-    public String getRemoteNodeID() {
-        return remoteNodeID;
-    }
-
     public void setRemoteNodeID(String remoteNodeID) {
         this.remoteNodeID = remoteNodeID;
     }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
index 3143284..dfc23d3 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
@@ -27,9 +27,6 @@ import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -39,13 +36,17 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Predicate;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.api.INcApplicationContext;
 import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.config.ReplicationProperties;
 import org.apache.asterix.common.context.IndexInfo;
+import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.exceptions.ACIDException;
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.replication.IReplicaResourcesManager;
@@ -55,6 +56,9 @@ import org.apache.asterix.common.replication.IReplicationStrategy;
 import org.apache.asterix.common.replication.IReplicationThread;
 import org.apache.asterix.common.replication.ReplicaEvent;
 import org.apache.asterix.common.storage.DatasetResourceReference;
+import org.apache.asterix.common.storage.IIndexCheckpointManager;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
+import org.apache.asterix.common.storage.ResourceReference;
 import org.apache.asterix.common.transactions.IAppRuntimeContextProvider;
 import org.apache.asterix.common.transactions.ILogManager;
 import org.apache.asterix.common.transactions.LogRecord;
@@ -72,9 +76,13 @@ import org.apache.asterix.replication.storage.LSMComponentProperties;
 import org.apache.asterix.replication.storage.LSMIndexFileProperties;
 import org.apache.asterix.replication.storage.ReplicaResourcesManager;
 import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
+import org.apache.asterix.transaction.management.service.logging.LogBuffer;
 import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
 import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager;
+import org.apache.hyracks.storage.common.LocalResource;
 import org.apache.hyracks.util.StorageUtil;
 import org.apache.hyracks.util.StorageUtil.StorageUnit;
 
@@ -89,7 +97,6 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
     private final String localNodeID;
     private final ILogManager logManager;
     private final ReplicaResourcesManager replicaResourcesManager;
-    private SocketChannel socketChannel = null;
     private ServerSocketChannel serverSocketChannel = null;
     private final IReplicationManager replicationManager;
     private final ReplicationProperties replicationProperties;
@@ -98,6 +105,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
     private final LinkedBlockingQueue<LSMComponentLSNSyncTask> lsmComponentRemoteLSN2LocalLSNMappingTaskQ;
     private final LinkedBlockingQueue<LogRecord> pendingNotificationRemoteLogsQ;
     private final Map<String, LSMComponentProperties> lsmComponentId2PropertiesMap;
+    private final Map<Long, RemoteLogMapping> localLsn2RemoteMapping;
     private final Map<String, RemoteLogMapping> replicaUniqueLSN2RemoteMapping;
     private final LSMComponentsSyncService lsmComponentLSNMappingService;
     private final Set<Integer> nodeHostedPartitions;
@@ -105,6 +113,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
     private final Object flushLogslock = new Object();
     private final IDatasetLifecycleManager dsLifecycleManager;
     private final PersistentLocalResourceRepository localResourceRep;
+    private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
 
     public ReplicationChannel(String nodeId, ReplicationProperties replicationProperties, ILogManager logManager,
             IReplicaResourcesManager replicaResoucesManager, IReplicationManager replicationManager,
@@ -122,6 +131,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
         pendingNotificationRemoteLogsQ = new LinkedBlockingQueue<>();
         lsmComponentId2PropertiesMap = new ConcurrentHashMap<>();
         replicaUniqueLSN2RemoteMapping = new ConcurrentHashMap<>();
+        localLsn2RemoteMapping = new ConcurrentHashMap<>();
         lsmComponentLSNMappingService = new LSMComponentsSyncService();
         replicationNotifier = new ReplicationNotifier();
         replicationThreads = Executors.newCachedThreadPool(ncServiceContext.getThreadFactory());
@@ -136,6 +146,8 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
         }
         nodeHostedPartitions = new HashSet<>(clientsPartitions.size());
         nodeHostedPartitions.addAll(clientsPartitions);
+        this.indexCheckpointManagerProvider =
+                ((INcApplicationContext) ncServiceContext.getApplicationContext()).getIndexCheckpointManagerProvider();
     }
 
     @Override
@@ -156,7 +168,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
 
             //start accepting replication requests
             while (true) {
-                socketChannel = serverSocketChannel.accept();
+                SocketChannel socketChannel = serverSocketChannel.accept();
                 socketChannel.configureBlocking(true);
                 //start a new thread to handle the request
                 replicationThreads.execute(new ReplicationThread(socketChannel));
@@ -349,16 +361,19 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
                 }
                 if (afp.isLSMComponentFile()) {
                     String componentId = LSMComponentProperties.getLSMComponentID(afp.getFilePath());
-                    if (afp.getLSNByteOffset() > AbstractLSMIOOperationCallback.INVALID) {
-                        LSMComponentLSNSyncTask syncTask = new LSMComponentLSNSyncTask(componentId,
-                                destFile.getAbsolutePath(), afp.getLSNByteOffset());
+                    final LSMComponentProperties lsmComponentProperties = lsmComponentId2PropertiesMap.get(componentId);
+                    // merge operations do not generate flush logs
+                    if (afp.requiresAck() && lsmComponentProperties.getOpType() == LSMOperationType.FLUSH) {
+                        LSMComponentLSNSyncTask syncTask =
+                                new LSMComponentLSNSyncTask(componentId, destFile.getAbsolutePath());
                         lsmComponentRemoteLSN2LocalLSNMappingTaskQ.offer(syncTask);
                     } else {
                         updateLSMComponentRemainingFiles(componentId);
                     }
                 } else {
                     //index metadata file
-                    replicaResourcesManager.initializeReplicaIndexLSNMap(indexPath, logManager.getAppendLSN());
+                    final ResourceReference indexRef = ResourceReference.of(destFile.getAbsolutePath());
+                    indexCheckpointManagerProvider.get(indexRef).init(logManager.getAppendLSN());
                 }
             }
         }
@@ -402,8 +417,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
                         try (RandomAccessFile fromFile = new RandomAccessFile(filePath, "r");
                                 FileChannel fileChannel = fromFile.getChannel();) {
                             long fileSize = fileChannel.size();
-                            fileProperties.initialize(filePath, fileSize, partition.getNodeId(), false,
-                                    AbstractLSMIOOperationCallback.INVALID, false);
+                            fileProperties.initialize(filePath, fileSize, partition.getNodeId(), false, false);
                             outBuffer = ReplicationProtocol.writeFileReplicationRequest(outBuffer, fileProperties,
                                     ReplicationRequestType.REPLICATE_FILE);
 
@@ -462,7 +476,7 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
                 switch (remoteLog.getLogType()) {
                     case LogType.UPDATE:
                     case LogType.ENTITY_COMMIT:
-                        //if the log partition belongs to a partitions hosted on this node, replicate it
+                        //if the log partition belongs to a partitions hosted on this node, replicated it
                         if (nodeHostedPartitions.contains(remoteLog.getResourcePartition())) {
                             logManager.log(remoteLog);
                         }
@@ -479,13 +493,21 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
                     case LogType.FLUSH:
                         //store mapping information for flush logs to use them in incoming LSM components.
                         RemoteLogMapping flushLogMap = new RemoteLogMapping();
+                        LogRecord flushLog = new LogRecord();
+                        TransactionUtil.formFlushLogRecord(flushLog, remoteLog.getDatasetId(), null,
+                                remoteLog.getNodeId(), remoteLog.getNumOfFlushedIndexes());
+                        flushLog.setReplicationThread(this);
+                        flushLog.setLogSource(LogSource.REMOTE);
                         flushLogMap.setRemoteNodeID(remoteLog.getNodeId());
                         flushLogMap.setRemoteLSN(remoteLog.getLSN());
-                        logManager.log(remoteLog);
-                        //the log LSN value is updated by logManager.log(.) to a local value
-                        flushLogMap.setLocalLSN(remoteLog.getLSN());
-                        flushLogMap.numOfFlushedIndexes.set(remoteLog.getNumOfFlushedIndexes());
-                        replicaUniqueLSN2RemoteMapping.put(flushLogMap.getNodeUniqueLSN(), flushLogMap);
+                        synchronized (localLsn2RemoteMapping) {
+                            logManager.log(flushLog);
+                            //the log LSN value is updated by logManager.log(.) to a local value
+                            flushLogMap.setLocalLSN(flushLog.getLSN());
+                            flushLogMap.numOfFlushedIndexes.set(flushLog.getNumOfFlushedIndexes());
+                            replicaUniqueLSN2RemoteMapping.put(flushLogMap.getNodeUniqueLSN(), flushLogMap);
+                            localLsn2RemoteMapping.put(flushLog.getLSN(), flushLogMap);
+                        }
                         synchronized (flushLogslock) {
                             flushLogslock.notify();
                         }
@@ -497,18 +519,55 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
         }
 
         /**
-         * this method is called sequentially by LogPage (notifyReplicationTerminator)
-         * for JOB_COMMIT and JOB_ABORT log types.
+         * this method is called sequentially by {@link LogBuffer#notifyReplicationTermination()}
+         * for JOB_COMMIT, JOB_ABORT, and FLUSH log types.
          */
         @Override
         public void notifyLogReplicationRequester(LogRecord logRecord) {
-            pendingNotificationRemoteLogsQ.offer(logRecord);
+            switch (logRecord.getLogType()) {
+                case LogType.JOB_COMMIT:
+                case LogType.ABORT:
+                    pendingNotificationRemoteLogsQ.offer(logRecord);
+                    break;
+                case LogType.FLUSH:
+                    final RemoteLogMapping remoteLogMapping;
+                    synchronized (localLsn2RemoteMapping) {
+                        remoteLogMapping = localLsn2RemoteMapping.remove(logRecord.getLSN());
+                    }
+                    checkpointReplicaIndexes(remoteLogMapping, logRecord.getDatasetId());
+                    break;
+                default:
+                    throw new IllegalStateException("Unexpected log type: " + logRecord.getLogType());
+            }
         }
 
         @Override
         public SocketChannel getReplicationClientSocket() {
             return socketChannel;
         }
+
+        private void checkpointReplicaIndexes(RemoteLogMapping remoteLogMapping, int datasetId) {
+            try {
+                Predicate<LocalResource> replicaIndexesPredicate = lr -> {
+                    DatasetLocalResource dls = (DatasetLocalResource) lr.getResource();
+                    return dls.getDatasetId() == datasetId && !localResourceRep.getActivePartitions()
+                            .contains(dls.getPartition());
+                };
+                final Map<Long, LocalResource> resources = localResourceRep.getResources(replicaIndexesPredicate);
+                final List<DatasetResourceReference> replicaIndexesRef =
+                        resources.values().stream().map(DatasetResourceReference::of).collect(Collectors.toList());
+                for (DatasetResourceReference replicaIndexRef : replicaIndexesRef) {
+                    final IIndexCheckpointManager indexCheckpointManager =
+                            indexCheckpointManagerProvider.get(replicaIndexRef);
+                    synchronized (indexCheckpointManager) {
+                        indexCheckpointManager
+                                .masterFlush(remoteLogMapping.getRemoteLSN(), remoteLogMapping.getLocalLSN());
+                    }
+                }
+            } catch (Exception e) {
+                LOGGER.log(Level.SEVERE, "Failed to checkpoint replica indexes", e);
+            }
+        }
     }
 
     /**
@@ -541,7 +600,6 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
      * the received LSM components to a local LSN.
      */
     private class LSMComponentsSyncService extends Thread {
-        private static final int BULKLOAD_LSN = 0;
 
         @Override
         public void run() {
@@ -560,90 +618,22 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
                         LOGGER.log(Level.SEVERE, "Unexpected exception during LSN synchronization", e);
                     }
                 }
-
             }
         }
 
         private void syncLSMComponentFlushLSN(LSMComponentProperties lsmCompProp, LSMComponentLSNSyncTask syncTask)
                 throws InterruptedException, IOException {
-            long remoteLSN = lsmCompProp.getOriginalLSN();
-            //LSN=0 (bulkload) does not need to be updated and there is no flush log corresponding to it
-            if (remoteLSN == BULKLOAD_LSN) {
-                //since this is the first LSM component of this index,
-                //then set the mapping in the LSN_MAP to the current log LSN because
-                //no other log could've been received for this index since bulkload replication is synchronous.
-                lsmCompProp.setReplicaLSN(logManager.getAppendLSN());
-                return;
-            }
-
-            //path to the LSM component file
-            Path path = Paths.get(syncTask.getComponentFilePath());
-            if (lsmCompProp.getReplicaLSN() == null) {
-                if (lsmCompProp.getOpType() == LSMOperationType.FLUSH) {
-                    //need to look up LSN mapping from memory
-                    RemoteLogMapping remoteLogMap = replicaUniqueLSN2RemoteMapping.get(lsmCompProp.getNodeUniqueLSN());
-                    //wait until flush log arrives, and verify the LSM component file still exists
-                    //The component file could be deleted if its NC fails.
-                    while (remoteLogMap == null && Files.exists(path)) {
-                        synchronized (flushLogslock) {
-                            flushLogslock.wait();
-                        }
-                        remoteLogMap = replicaUniqueLSN2RemoteMapping.get(lsmCompProp.getNodeUniqueLSN());
-                    }
-
-                    /**
-                     * file has been deleted due to its remote primary replica failure
-                     * before its LSN could've been synchronized.
-                     */
-                    if (remoteLogMap == null) {
-                        return;
-                    }
-                    lsmCompProp.setReplicaLSN(remoteLogMap.getLocalLSN());
-                } else if (lsmCompProp.getOpType() == LSMOperationType.MERGE) {
-                    //need to load the LSN mapping from disk
-                    Map<Long, Long> lsmMap = replicaResourcesManager
-                            .getReplicaIndexLSNMap(lsmCompProp.getReplicaComponentPath(replicaResourcesManager));
-                    Long mappingLSN = lsmMap.get(lsmCompProp.getOriginalLSN());
-                    if (mappingLSN == null) {
-                        /**
-                         * this shouldn't happen unless this node just recovered and
-                         * the first component it received is a merged component due
-                         * to an on-going merge operation while recovery on the remote
-                         * replica. In this case, we use the current append LSN since
-                         * no new records exist for this index, otherwise they would've
-                         * been flushed. This could be prevented by waiting for any IO
-                         * to finish on the remote replica during recovery.
-                         */
-                        mappingLSN = logManager.getAppendLSN();
-                    }
-                    lsmCompProp.setReplicaLSN(mappingLSN);
+            final String componentFilePath = syncTask.getComponentFilePath();
+            final ResourceReference indexRef = ResourceReference.of(componentFilePath);
+            final IIndexCheckpointManager indexCheckpointManager = indexCheckpointManagerProvider.get(indexRef);
+            synchronized (indexCheckpointManager) {
+                long masterLsn = lsmCompProp.getOriginalLSN();
+                // wait until the lsn mapping is flushed to disk
+                while (!indexCheckpointManager.isFlushed(masterLsn)) {
+                    indexCheckpointManager.wait();
                 }
-            }
-
-            if (Files.notExists(path)) {
-                /**
-                 * This could happen when a merged component arrives and deletes
-                 * the flushed component (which we are trying to update) before
-                 * its flush log arrives since logs and components are received
-                 * on different threads.
-                 */
-                return;
-            }
-
-            File destFile = new File(syncTask.getComponentFilePath());
-            //prepare local LSN buffer
-            ByteBuffer metadataBuffer = ByteBuffer.allocate(Long.BYTES);
-            metadataBuffer.putLong(lsmCompProp.getReplicaLSN());
-            metadataBuffer.flip();
-
-            //replace the remote LSN value by the local one
-            try (RandomAccessFile fileOutputStream = new RandomAccessFile(destFile, "rw");
-                    FileChannel fileChannel = fileOutputStream.getChannel()) {
-                long lsnStartOffset = syncTask.getLSNByteOffset();
-                while (metadataBuffer.hasRemaining()) {
-                    lsnStartOffset += fileChannel.write(metadataBuffer, lsnStartOffset);
-                }
-                fileChannel.force(true);
+                indexCheckpointManager
+                        .replicated(AbstractLSMIndexFileManager.getComponentEndTime(indexRef.getName()), masterLsn);
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
index 48c7083..5cf7eab 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationManager.java
@@ -54,7 +54,6 @@ import java.util.stream.Collectors;
 
 import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.config.ReplicationProperties;
-import org.apache.asterix.common.dataflow.LSMIndexUtil;
 import org.apache.asterix.common.replication.IReplicaResourcesManager;
 import org.apache.asterix.common.replication.IReplicationManager;
 import org.apache.asterix.common.replication.IReplicationStrategy;
@@ -63,6 +62,8 @@ import org.apache.asterix.common.replication.Replica.ReplicaState;
 import org.apache.asterix.common.replication.ReplicaEvent;
 import org.apache.asterix.common.replication.ReplicationJob;
 import org.apache.asterix.common.storage.DatasetResourceReference;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
+import org.apache.asterix.common.storage.ResourceReference;
 import org.apache.asterix.common.transactions.IAppRuntimeContextProvider;
 import org.apache.asterix.common.transactions.ILogManager;
 import org.apache.asterix.common.transactions.ILogRecord;
@@ -84,7 +85,6 @@ import org.apache.hyracks.api.replication.IReplicationJob;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationExecutionType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationJobType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexReplicationJob;
 import org.apache.hyracks.util.StorageUtil;
 import org.apache.hyracks.util.StorageUtil.StorageUnit;
@@ -136,6 +136,7 @@ public class ReplicationManager implements IReplicationManager {
     private final ByteBuffer txnLogsBatchSizeBuffer = ByteBuffer.allocate(Integer.BYTES);
     private final IReplicationStrategy replicationStrategy;
     private final PersistentLocalResourceRepository localResourceRepo;
+    private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
 
     //TODO this class needs to be refactored by moving its private classes to separate files
     //and possibly using MessageBroker to send/receive remote replicas events.
@@ -148,6 +149,8 @@ public class ReplicationManager implements IReplicationManager {
         this.replicaResourcesManager = (ReplicaResourcesManager) remoteResoucesManager;
         this.asterixAppRuntimeContextProvider = asterixAppRuntimeContextProvider;
         this.logManager = logManager;
+        this.indexCheckpointManagerProvider =
+                asterixAppRuntimeContextProvider.getAppContext().getIndexCheckpointManagerProvider();
         localResourceRepo =
                 (PersistentLocalResourceRepository) asterixAppRuntimeContextProvider.getLocalResourceRepository();
         this.hostIPAddressFirstOctet = replicationProperties.getReplicaIPAddress(nodeId).substring(0, 3);
@@ -284,7 +287,6 @@ public class ReplicationManager implements IReplicationManager {
             if (!replicationStrategy.isMatch(indexFileRef.getDatasetId())) {
                 return;
             }
-
             int jobPartitionId = indexFileRef.getPartitionId();
 
             ByteBuffer responseBuffer = null;
@@ -327,25 +329,10 @@ public class ReplicationManager implements IReplicationManager {
                                 FileChannel fileChannel = fromFile.getChannel();) {
 
                             long fileSize = fileChannel.size();
-
-                            if (LSMComponentJob != null) {
-                                /**
-                                 * since this is LSM_COMPONENT REPLICATE job, the job will contain
-                                 * only the component being replicated.
-                                 */
-                                ILSMDiskComponent diskComponent = LSMComponentJob.getLSMIndexOperationContext()
-                                        .getComponentsToBeReplicated().get(0);
-                                long lsnOffset = LSMIndexUtil.getComponentFileLSNOffset(LSMComponentJob.getLSMIndex(),
-                                        diskComponent, filePath);
-                                asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile,
-                                        lsnOffset, remainingFiles == 0);
-                            } else {
-                                asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile, -1L,
-                                        remainingFiles == 0);
-                            }
+                            asterixFileProperties.initialize(filePath, fileSize, nodeId, isLSMComponentFile,
+                                    remainingFiles == 0);
                             requestBuffer = ReplicationProtocol.writeFileReplicationRequest(requestBuffer,
                                     asterixFileProperties, ReplicationRequestType.REPLICATE_FILE);
-
                             Iterator<Map.Entry<String, SocketChannel>> iterator = replicasSockets.entrySet().iterator();
                             while (iterator.hasNext()) {
                                 Map.Entry<String, SocketChannel> entry = iterator.next();
@@ -378,8 +365,7 @@ public class ReplicationManager implements IReplicationManager {
                 } else if (job.getOperation() == ReplicationOperation.DELETE) {
                     for (String filePath : job.getJobFiles()) {
                         remainingFiles--;
-                        asterixFileProperties.initialize(filePath, -1, nodeId, isLSMComponentFile, -1L,
-                                remainingFiles == 0);
+                        asterixFileProperties.initialize(filePath, -1, nodeId, isLSMComponentFile, remainingFiles == 0);
                         ReplicationProtocol.writeFileReplicationRequest(requestBuffer, asterixFileProperties,
                                 ReplicationRequestType.DELETE_FILE);
 
@@ -1026,10 +1012,10 @@ public class ReplicationManager implements IReplicationManager {
         ByteBuffer requestBuffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
         for (String replicaId : replicaIds) {
             //1. identify replica indexes with LSN less than nonSharpCheckpointTargetLSN.
-            Map<Long, String> laggingIndexes =
+            Map<Long, DatasetResourceReference> laggingIndexes =
                     replicaResourcesManager.getLaggingReplicaIndexesId2PathMap(replicaId, nonSharpCheckpointTargetLSN);
 
-            if (laggingIndexes.size() > 0) {
+            if (!laggingIndexes.isEmpty()) {
                 //2. send request to remote replicas that have lagging indexes.
                 ReplicaIndexFlushRequest laggingIndexesResponse = null;
                 try (SocketChannel socketChannel = getReplicaSocket(replicaId)) {
@@ -1052,16 +1038,14 @@ public class ReplicationManager implements IReplicationManager {
                 }
 
                 /*
-                 * 4. update the LSN_MAP for indexes that were not flushed
+                 * 4. update checkpoints for indexes that were not flushed
                  * to the current append LSN to indicate no operations happened
                  * since the checkpoint start.
                  */
                 if (laggingIndexesResponse != null) {
                     for (Long resouceId : laggingIndexesResponse.getLaggingRescouresIds()) {
-                        String indexPath = laggingIndexes.get(resouceId);
-                        Map<Long, Long> indexLSNMap = replicaResourcesManager.getReplicaIndexLSNMap(indexPath);
-                        indexLSNMap.put(ReplicaResourcesManager.REPLICA_INDEX_CREATION_LSN, startLSN);
-                        replicaResourcesManager.updateReplicaIndexLSNMap(indexPath, indexLSNMap);
+                        final DatasetResourceReference datasetResourceReference = laggingIndexes.get(resouceId);
+                        indexCheckpointManagerProvider.get(datasetResourceReference).advanceLowWatermark(startLSN);
                     }
                 }
             }
@@ -1141,12 +1125,11 @@ public class ReplicationManager implements IReplicationManager {
                     fileChannel.force(true);
                 }
 
-                //we need to create LSN map for .metadata files that belong to remote replicas
+                //we need to create initial map for .metadata files that belong to remote replicas
                 if (!fileProperties.isLSMComponentFile() && !fileProperties.getNodeId().equals(nodeId)) {
-                    //replica index
-                    replicaResourcesManager.initializeReplicaIndexLSNMap(indexPath, logManager.getAppendLSN());
+                    final ResourceReference indexRef = ResourceReference.of(destFile.getAbsolutePath());
+                    indexCheckpointManagerProvider.get(indexRef).init(logManager.getAppendLSN());
                 }
-
                 responseFunction = ReplicationProtocol.getRequestType(socketChannel, dataBuffer);
             }
 

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentLSNSyncTask.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentLSNSyncTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentLSNSyncTask.java
index f11adc2..08c0ec7 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentLSNSyncTask.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentLSNSyncTask.java
@@ -19,33 +19,21 @@
 package org.apache.asterix.replication.storage;
 
 public class LSMComponentLSNSyncTask {
+
     private String componentFilePath;
     private String componentId;
-    private long LSNByteOffset;
 
-    public LSMComponentLSNSyncTask(String componentId, String componentFilePath, long LSNByteOffset) {
+    public LSMComponentLSNSyncTask(String componentId, String componentFilePath) {
         this.componentId = componentId;
         this.componentFilePath = componentFilePath;
-        this.LSNByteOffset = LSNByteOffset;
     }
 
     public String getComponentFilePath() {
         return componentFilePath;
     }
 
-    public void setComponentFilePath(String componentFilePath) {
-        this.componentFilePath = componentFilePath;
-    }
-
     public String getComponentId() {
         return componentId;
     }
 
-    public void setComponentId(String componentId) {
-        this.componentId = componentId;
-    }
-
-    public long getLSNByteOffset() {
-        return LSNByteOffset;
-    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentProperties.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentProperties.java
index 7ca6f2f..bf987d0 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentProperties.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMComponentProperties.java
@@ -52,9 +52,10 @@ public class LSMComponentProperties {
         this.nodeId = nodeId;
         componentId = LSMComponentProperties.getLSMComponentID((String) job.getJobFiles().toArray()[0]);
         numberOfFiles = new AtomicInteger(job.getJobFiles().size());
-        originalLSN = LSMComponentProperties.getLSMComponentLSN((AbstractLSMIndex) job.getLSMIndex(),
-                job.getLSMIndexOperationContext());
         opType = job.getLSMOpType();
+        originalLSN = opType == LSMOperationType.FLUSH ?
+                LSMComponentProperties.getLSMComponentLSN((AbstractLSMIndex) job.getLSMIndex(),
+                        job.getLSMIndexOperationContext()) : 0;
     }
 
     public LSMComponentProperties() {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
index f2747fe..2ebf2cb 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/LSMIndexFileProperties.java
@@ -31,27 +31,25 @@ public class LSMIndexFileProperties {
     private boolean lsmComponentFile;
     private String filePath;
     private boolean requiresAck = false;
-    private long LSNByteOffset;
 
     public LSMIndexFileProperties() {
     }
 
     public LSMIndexFileProperties(String filePath, long fileSize, String nodeId, boolean lsmComponentFile,
-            long LSNByteOffset, boolean requiresAck) {
-        initialize(filePath, fileSize, nodeId, lsmComponentFile, LSNByteOffset, requiresAck);
+            boolean requiresAck) {
+        initialize(filePath, fileSize, nodeId, lsmComponentFile, requiresAck);
     }
 
     public LSMIndexFileProperties(LSMComponentProperties lsmComponentProperties) {
-        initialize(lsmComponentProperties.getComponentId(), -1, lsmComponentProperties.getNodeId(), false, -1L, false);
+        initialize(lsmComponentProperties.getComponentId(), -1, lsmComponentProperties.getNodeId(), false, false);
     }
 
-    public void initialize(String filePath, long fileSize, String nodeId, boolean lsmComponentFile, long LSNByteOffset,
+    public void initialize(String filePath, long fileSize, String nodeId, boolean lsmComponentFile,
             boolean requiresAck) {
         this.filePath = filePath;
         this.fileSize = fileSize;
         this.nodeId = nodeId;
         this.lsmComponentFile = lsmComponentFile;
-        this.LSNByteOffset = LSNByteOffset;
         this.requiresAck = requiresAck;
     }
 
@@ -61,7 +59,6 @@ public class LSMIndexFileProperties {
         dos.writeUTF(filePath);
         dos.writeLong(fileSize);
         dos.writeBoolean(lsmComponentFile);
-        dos.writeLong(LSNByteOffset);
         dos.writeBoolean(requiresAck);
     }
 
@@ -70,10 +67,9 @@ public class LSMIndexFileProperties {
         String filePath = input.readUTF();
         long fileSize = input.readLong();
         boolean lsmComponentFile = input.readBoolean();
-        long LSNByteOffset = input.readLong();
         boolean requiresAck = input.readBoolean();
-        LSMIndexFileProperties fileProp = new LSMIndexFileProperties(filePath, fileSize, nodeId, lsmComponentFile,
-                LSNByteOffset, requiresAck);
+        LSMIndexFileProperties fileProp =
+                new LSMIndexFileProperties(filePath, fileSize, nodeId, lsmComponentFile, requiresAck);
         return fileProp;
     }
 
@@ -108,11 +104,6 @@ public class LSMIndexFileProperties {
         sb.append("File Size: " + fileSize + "  ");
         sb.append("Node ID: " + nodeId + "  ");
         sb.append("isLSMComponentFile : " + lsmComponentFile + "  ");
-        sb.append("LSN Byte Offset: " + LSNByteOffset);
         return sb.toString();
     }
-
-    public long getLSNByteOffset() {
-        return LSNByteOffset;
-    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/ReplicaResourcesManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/ReplicaResourcesManager.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/ReplicaResourcesManager.java
index 7eea4a4..2ff74a8 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/ReplicaResourcesManager.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/storage/ReplicaResourcesManager.java
@@ -19,27 +19,26 @@
 package org.apache.asterix.replication.storage;
 
 import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
 import java.io.FilenameFilter;
 import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.logging.Level;
-import java.util.logging.Logger;
+import java.util.stream.Collectors;
 
 import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.config.MetadataProperties;
+import org.apache.asterix.common.dataflow.DatasetLocalResource;
 import org.apache.asterix.common.replication.IReplicaResourcesManager;
+import org.apache.asterix.common.storage.DatasetResourceReference;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
 import org.apache.asterix.common.utils.StorageConstants;
 import org.apache.asterix.common.utils.StoragePathUtil;
 import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
@@ -50,16 +49,15 @@ import org.apache.hyracks.storage.common.ILocalResourceRepository;
 import org.apache.hyracks.storage.common.LocalResource;
 
 public class ReplicaResourcesManager implements IReplicaResourcesManager {
-    private static final Logger LOGGER = Logger.getLogger(ReplicaResourcesManager.class.getName());
-    public final static String LSM_COMPONENT_MASK_SUFFIX = "_mask";
-    private final static String REPLICA_INDEX_LSN_MAP_NAME = ".LSN_MAP";
-    public static final long REPLICA_INDEX_CREATION_LSN = -1;
+    public static final String LSM_COMPONENT_MASK_SUFFIX = "_mask";
     private final PersistentLocalResourceRepository localRepository;
     private final Map<String, ClusterPartition[]> nodePartitions;
+    private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
 
-    public ReplicaResourcesManager(ILocalResourceRepository localRepository,
-            MetadataProperties metadataProperties) {
+    public ReplicaResourcesManager(ILocalResourceRepository localRepository, MetadataProperties metadataProperties,
+            IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
         this.localRepository = (PersistentLocalResourceRepository) localRepository;
+        this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
         nodePartitions = metadataProperties.getNodePartitions();
     }
 
@@ -86,12 +84,6 @@ public class ReplicaResourcesManager implements IReplicaResourcesManager {
         return indexPath.toString();
     }
 
-    public void initializeReplicaIndexLSNMap(String indexPath, long currentLSN) throws IOException {
-        HashMap<Long, Long> lsnMap = new HashMap<Long, Long>();
-        lsnMap.put(REPLICA_INDEX_CREATION_LSN, currentLSN);
-        updateReplicaIndexLSNMap(indexPath, lsnMap);
-    }
-
     public void createRemoteLSMComponentMask(LSMComponentProperties lsmComponentProperties) throws IOException {
         String maskPath = lsmComponentProperties.getMaskPath(this);
         Path path = Paths.get(maskPath);
@@ -106,13 +98,6 @@ public class ReplicaResourcesManager implements IReplicaResourcesManager {
         String maskPath = lsmComponentProperties.getMaskPath(this);
         Path path = Paths.get(maskPath);
         Files.deleteIfExists(path);
-
-        //add component LSN to the index LSNs map
-        Map<Long, Long> lsnMap = getReplicaIndexLSNMap(lsmComponentProperties.getReplicaComponentPath(this));
-        lsnMap.put(lsmComponentProperties.getOriginalLSN(), lsmComponentProperties.getReplicaLSN());
-
-        //update map on disk
-        updateReplicaIndexLSNMap(lsmComponentProperties.getReplicaComponentPath(this), lsnMap);
     }
 
     public Set<File> getReplicaIndexes(String replicaId) throws HyracksDataException {
@@ -128,58 +113,37 @@ public class ReplicaResourcesManager implements IReplicaResourcesManager {
     public long getPartitionsMinLSN(Set<Integer> partitions) throws HyracksDataException {
         long minRemoteLSN = Long.MAX_VALUE;
         for (Integer partition : partitions) {
-            //for every index in replica
-            Set<File> remoteIndexes = localRepository.getPartitionIndexes(partition);
-            for (File indexFolder : remoteIndexes) {
-                //read LSN map
-                try {
-                    //get max LSN per index
-                    long remoteIndexMaxLSN = getReplicaIndexMaxLSN(indexFolder);
-
-                    //get min of all maximums
-                    minRemoteLSN = Math.min(minRemoteLSN, remoteIndexMaxLSN);
-                } catch (IOException e) {
-                    LOGGER.log(Level.INFO,
-                            indexFolder.getAbsolutePath() + " Couldn't read LSN map for index " + indexFolder);
-                    continue;
-                }
+            final List<DatasetResourceReference> partitionResources = localRepository.getResources(resource -> {
+                DatasetLocalResource dsResource = (DatasetLocalResource) resource.getResource();
+                return dsResource.getPartition() == partition;
+            }).values().stream().map(DatasetResourceReference::of).collect(Collectors.toList());
+            for (DatasetResourceReference indexRef : partitionResources) {
+                long remoteIndexMaxLSN = indexCheckpointManagerProvider.get(indexRef).getLowWatermark();
+                minRemoteLSN = Math.min(minRemoteLSN, remoteIndexMaxLSN);
             }
         }
         return minRemoteLSN;
     }
 
-    public Map<Long, String> getLaggingReplicaIndexesId2PathMap(String replicaId, long targetLSN) throws IOException {
-        Map<Long, String> laggingReplicaIndexes = new HashMap<Long, String>();
-        try {
-            //for every index in replica
-            Set<File> remoteIndexes = getReplicaIndexes(replicaId);
-            for (File indexFolder : remoteIndexes) {
-                if (getReplicaIndexMaxLSN(indexFolder) < targetLSN) {
-                    File localResource = new File(
-                            indexFolder + File.separator + StorageConstants.METADATA_FILE_NAME);
-                    LocalResource resource = PersistentLocalResourceRepository.readLocalResource(localResource);
-                    laggingReplicaIndexes.put(resource.getId(), indexFolder.getAbsolutePath());
+    public Map<Long, DatasetResourceReference> getLaggingReplicaIndexesId2PathMap(String replicaId, long targetLSN)
+            throws HyracksDataException {
+        Map<Long, DatasetResourceReference> laggingReplicaIndexes = new HashMap<>();
+        final List<Integer> replicaPartitions =
+                Arrays.stream(nodePartitions.get(replicaId)).map(ClusterPartition::getPartitionId)
+                        .collect(Collectors.toList());
+        for (int patition : replicaPartitions) {
+            final Map<Long, LocalResource> partitionResources = localRepository.getPartitionResources(patition);
+            final List<DatasetResourceReference> indexesRefs =
+                    partitionResources.values().stream().map(DatasetResourceReference::of).collect(Collectors.toList());
+            for (DatasetResourceReference ref : indexesRefs) {
+                if (indexCheckpointManagerProvider.get(ref).getLowWatermark() < targetLSN) {
+                    laggingReplicaIndexes.put(ref.getResourceId(), ref);
                 }
             }
-        } catch (HyracksDataException e) {
-            e.printStackTrace();
         }
-
         return laggingReplicaIndexes;
     }
 
-    private long getReplicaIndexMaxLSN(File indexFolder) throws IOException {
-        long remoteIndexMaxLSN = 0;
-        //get max LSN per index
-        Map<Long, Long> lsnMap = getReplicaIndexLSNMap(indexFolder.getAbsolutePath());
-        if (lsnMap != null) {
-            for (Long lsn : lsnMap.values()) {
-                remoteIndexMaxLSN = Math.max(remoteIndexMaxLSN, lsn);
-            }
-        }
-        return remoteIndexMaxLSN;
-    }
-
     public void cleanInvalidLSMComponents(String replicaId) {
         //for every index in replica
         Set<File> remoteIndexes = null;
@@ -214,28 +178,6 @@ public class ReplicaResourcesManager implements IReplicaResourcesManager {
         }
     }
 
-    @SuppressWarnings({ "unchecked" })
-    public synchronized Map<Long, Long> getReplicaIndexLSNMap(String indexPath) throws IOException {
-        try (FileInputStream fis = new FileInputStream(indexPath + File.separator + REPLICA_INDEX_LSN_MAP_NAME);
-                ObjectInputStream oisFromFis = new ObjectInputStream(fis)) {
-            Map<Long, Long> lsnMap = null;
-            try {
-                lsnMap = (Map<Long, Long>) oisFromFis.readObject();
-            } catch (ClassNotFoundException e) {
-                e.printStackTrace();
-            }
-            return lsnMap;
-        }
-    }
-
-    public synchronized void updateReplicaIndexLSNMap(String indexPath, Map<Long, Long> lsnMap) throws IOException {
-        try (FileOutputStream fos = new FileOutputStream(indexPath + File.separator + REPLICA_INDEX_LSN_MAP_NAME);
-                ObjectOutputStream oosToFos = new ObjectOutputStream(fos)) {
-            oosToFos.writeObject(lsnMap);
-            oosToFos.flush();
-        }
-    }
-
     /**
      * @param partition
      * @return Absolute paths to all partition files

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index e87a39b..ba43fb7 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -29,6 +29,7 @@ import java.io.ObjectOutputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -38,8 +39,6 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Predicate;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.config.MetadataProperties;
@@ -48,10 +47,12 @@ import org.apache.asterix.common.exceptions.AsterixException;
 import org.apache.asterix.common.replication.IReplicationManager;
 import org.apache.asterix.common.replication.ReplicationJob;
 import org.apache.asterix.common.storage.DatasetResourceReference;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
 import org.apache.asterix.common.storage.ResourceReference;
 import org.apache.asterix.common.utils.StorageConstants;
 import org.apache.asterix.common.utils.StoragePathUtil;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.IOFileFilter;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.io.IIOManager;
@@ -72,7 +73,29 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     public static final Predicate<Path> INDEX_COMPONENTS = path -> !path.endsWith(StorageConstants.METADATA_FILE_NAME);
     // Private constants
     private static final int MAX_CACHED_RESOURCES = 1000;
-    private static final int RESOURCES_TREE_DEPTH_FROM_STORAGE_ROOT = 6;
+    private static final IOFileFilter METADATA_FILES_FILTER = new IOFileFilter() {
+        @Override
+        public boolean accept(File file) {
+            return file.getName().equals(StorageConstants.METADATA_FILE_NAME);
+        }
+
+        @Override
+        public boolean accept(File dir, String name) {
+            return false;
+        }
+    };
+
+    private static final IOFileFilter ALL_DIR_FILTER = new IOFileFilter() {
+        @Override
+        public boolean accept(File file) {
+            return true;
+        }
+
+        @Override
+        public boolean accept(File dir, String name) {
+            return true;
+        }
+    };
 
     // Finals
     private final IIOManager ioManager;
@@ -85,15 +108,17 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     private IReplicationManager replicationManager;
     private Set<Integer> nodeInactivePartitions;
     private final Path[] storageRoots;
+    private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
 
-    public PersistentLocalResourceRepository(IIOManager ioManager, String nodeId,
-            MetadataProperties metadataProperties) {
+    public PersistentLocalResourceRepository(IIOManager ioManager, String nodeId, MetadataProperties metadataProperties,
+            IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
         this.ioManager = ioManager;
+        this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
         storageRoots = new Path[ioManager.getIODevices().size()];
         final List<IODeviceHandle> ioDevices = ioManager.getIODevices();
         for (int i = 0; i < ioDevices.size(); i++) {
-            storageRoots[i] =
-                    Paths.get(ioDevices.get(i).getMount().getAbsolutePath(), StorageConstants.STORAGE_ROOT_DIR_NAME);
+            storageRoots[i] = Paths.get(ioDevices.get(i).getMount().getAbsolutePath(),
+                    StorageConstants.STORAGE_ROOT_DIR_NAME);
         }
         createStorageRoots();
         resourceCache = CacheBuilder.newBuilder().maximumSize(MAX_CACHED_RESOURCES).build();
@@ -119,7 +144,7 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     }
 
     @Override
-    public LocalResource get(String relativePath) throws HyracksDataException {
+    public synchronized LocalResource get(String relativePath) throws HyracksDataException {
         LocalResource resource = resourceCache.getIfPresent(relativePath);
         if (resource == null) {
             FileReference resourceFile = getLocalResourceFileByName(ioManager, relativePath);
@@ -153,7 +178,7 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         }
 
         resourceCache.put(resource.getPath(), resource);
-
+        indexCheckpointManagerProvider.get(DatasetResourceReference.of(resource)).init(0);
         //if replication enabled, send resource metadata info to remote nodes
         if (isReplicationEnabled) {
             createReplicationJob(ReplicationOperation.REPLICATE, resourceFile);
@@ -164,18 +189,16 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     public synchronized void delete(String relativePath) throws HyracksDataException {
         FileReference resourceFile = getLocalResourceFileByName(ioManager, relativePath);
         if (resourceFile.getFile().exists()) {
-            try {
-                // Invalidate before deleting the file just in case file deletion throws some exception.
-                // Since it's just a cache invalidation, it should not affect correctness.
-                resourceCache.invalidate(relativePath);
-                IoUtil.delete(resourceFile);
-            } finally {
-                // Regardless of successfully deleted or not, the operation should be replicated.
-                //if replication enabled, delete resource from remote replicas
-                if (isReplicationEnabled) {
-                    createReplicationJob(ReplicationOperation.DELETE, resourceFile);
-                }
+            if (isReplicationEnabled) {
+                createReplicationJob(ReplicationOperation.DELETE, resourceFile);
             }
+            // delete all checkpoints
+            final LocalResource localResource = readLocalResource(resourceFile.getFile());
+            indexCheckpointManagerProvider.get(DatasetResourceReference.of(localResource)).delete();
+            // Invalidate before deleting the file just in case file deletion throws some exception.
+            // Since it's just a cache invalidation, it should not affect correctness.
+            resourceCache.invalidate(relativePath);
+            IoUtil.delete(resourceFile);
         } else {
             throw HyracksDataException
                     .create(org.apache.hyracks.api.exceptions.ErrorCode.RESOURCE_DOES_NOT_EXIST, relativePath);
@@ -188,13 +211,13 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         return ioManager.resolve(fileName);
     }
 
-    public Map<Long, LocalResource> getResources(Predicate<LocalResource> filter) throws HyracksDataException {
+    public synchronized Map<Long, LocalResource> getResources(Predicate<LocalResource> filter)
+            throws HyracksDataException {
         Map<Long, LocalResource> resourcesMap = new HashMap<>();
         for (Path root : storageRoots) {
-            try (Stream<Path> stream = Files.find(root, RESOURCES_TREE_DEPTH_FROM_STORAGE_ROOT,
-                    (path, attr) -> path.getFileName().toString().equals(StorageConstants.METADATA_FILE_NAME))) {
-                final List<File> resourceMetadataFiles = stream.map(Path::toFile).collect(Collectors.toList());
-                for (File file : resourceMetadataFiles) {
+            final Collection<File> files = FileUtils.listFiles(root.toFile(), METADATA_FILES_FILTER, ALL_DIR_FILTER);
+            try {
+                for (File file : files) {
                     final LocalResource localResource = PersistentLocalResourceRepository.readLocalResource(file);
                     if (filter.test(localResource)) {
                         resourcesMap.put(localResource.getId(), localResource);
@@ -321,6 +344,13 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         return indexes;
     }
 
+    public Map<Long, LocalResource> getPartitionResources(int partition) throws HyracksDataException {
+        return getResources(resource -> {
+            DatasetLocalResource dsResource = (DatasetLocalResource) resource.getResource();
+            return dsResource.getPartition() == partition;
+        });
+    }
+
     /**
      * Given any index file, an absolute {@link FileReference} is returned which points to where the index of
      * {@code indexFile} is located.

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/929344e9/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
index 43024b6..33c6260 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
@@ -19,6 +19,7 @@
 package org.apache.asterix.transaction.management.resource;
 
 import org.apache.asterix.common.config.MetadataProperties;
+import org.apache.asterix.common.storage.IIndexCheckpointManagerProvider;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.IIOManager;
 import org.apache.hyracks.storage.common.ILocalResourceRepository;
@@ -28,16 +29,19 @@ public class PersistentLocalResourceRepositoryFactory implements ILocalResourceR
     private final IIOManager ioManager;
     private final String nodeId;
     private final MetadataProperties metadataProperties;
+    private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
 
     public PersistentLocalResourceRepositoryFactory(IIOManager ioManager, String nodeId,
-            MetadataProperties metadataProperties) {
+            MetadataProperties metadataProperties, IIndexCheckpointManagerProvider indexCheckpointManagerProvider) {
         this.ioManager = ioManager;
         this.nodeId = nodeId;
         this.metadataProperties = metadataProperties;
+        this.indexCheckpointManagerProvider = indexCheckpointManagerProvider;
     }
 
     @Override
     public ILocalResourceRepository createRepository() throws HyracksDataException {
-        return new PersistentLocalResourceRepository(ioManager, nodeId, metadataProperties);
+        return new PersistentLocalResourceRepository(ioManager, nodeId, metadataProperties,
+                indexCheckpointManagerProvider);
     }
 }


Mime
View raw message