diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
index 4acc6d3..2d40ec8 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
@@ -56,7 +56,7 @@
     }
 
     @Override
-    public synchronized void init(long validComponentSequence, long lsn, long validComponentId)
+    public synchronized void init(long validComponentSequence, long lsn, long validComponentId, String masterNodeId)
             throws HyracksDataException {
         List<IndexCheckpoint> checkpoints;
         try {
@@ -68,26 +68,34 @@
             LOGGER.warn(() -> "Checkpoints found on initializing: " + indexPath);
             delete();
         }
-        IndexCheckpoint firstCheckpoint = IndexCheckpoint.first(validComponentSequence, lsn, validComponentId);
+        IndexCheckpoint firstCheckpoint =
+                IndexCheckpoint.first(validComponentSequence, lsn, validComponentId, masterNodeId);
         persist(firstCheckpoint);
     }
 
     @Override
-    public synchronized void replicated(long componentSequence, long masterLsn, long componentId)
+    public synchronized void replicated(long componentSequence, long masterLsn, long componentId, String masterNodeId)
             throws HyracksDataException {
         final Long localLsn = getLatest().getMasterNodeFlushMap().get(masterLsn);
         if (localLsn == null) {
-            throw new IllegalStateException("Component flushed before lsn mapping was received");
+            throw new IllegalStateException("Component replicated before lsn mapping was received");
         }
-        flushed(componentSequence, localLsn, componentId);
+        flushed(componentSequence, localLsn, componentId, masterNodeId);
+    }
+
+    @Override
+    public synchronized void flushed(long componentSequence, long lsn, long componentId, String masterNodeId)
+            throws HyracksDataException {
+        final IndexCheckpoint latest = getLatest();
+        IndexCheckpoint nextCheckpoint =
+                IndexCheckpoint.next(latest, lsn, componentSequence, componentId, masterNodeId);
+        persist(nextCheckpoint);
+        deleteHistory(nextCheckpoint.getId(), HISTORY_CHECKPOINTS);
     }
 
     @Override
     public synchronized void flushed(long componentSequence, long lsn, long componentId) throws HyracksDataException {
-        final IndexCheckpoint latest = getLatest();
-        IndexCheckpoint nextCheckpoint = IndexCheckpoint.next(latest, lsn, componentSequence, componentId);
-        persist(nextCheckpoint);
-        deleteHistory(nextCheckpoint.getId(), HISTORY_CHECKPOINTS);
+        flushed(componentSequence, lsn, componentId, null);
     }
 
     @Override
@@ -95,7 +103,7 @@
         final IndexCheckpoint latest = getLatest();
         latest.getMasterNodeFlushMap().put(masterLsn, localLsn);
         final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(),
-                latest.getValidComponentSequence(), latest.getLastComponentId());
+                latest.getValidComponentSequence(), latest.getLastComponentId(), null);
         persist(next);
         notifyAll();
     }
@@ -155,8 +163,8 @@
     @Override
     public synchronized void setLastComponentId(long componentId) throws HyracksDataException {
         final IndexCheckpoint latest = getLatest();
-        final IndexCheckpoint next =
-                IndexCheckpoint.next(latest, latest.getLowWatermark(), latest.getValidComponentSequence(), componentId);
+        final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(),
+                latest.getValidComponentSequence(), componentId, null);
         persist(next);
     }
 
@@ -165,7 +173,7 @@
         final IndexCheckpoint latest = getLatest();
         if (componentSequence > latest.getValidComponentSequence()) {
             final IndexCheckpoint next = IndexCheckpoint.next(latest, latest.getLowWatermark(), componentSequence,
-                    latest.getLastComponentId());
+                    latest.getLastComponentId(), null);
             persist(next);
         }
     }
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/ReplicaManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/ReplicaManager.java
index 66a69ef..f6de92d 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/ReplicaManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/ReplicaManager.java
@@ -19,6 +19,7 @@
 package org.apache.asterix.app.nc;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -30,6 +31,7 @@
 
 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.replication.IPartitionReplica;
 import org.apache.asterix.common.storage.IReplicaManager;
 import org.apache.asterix.common.storage.ReplicaIdentifier;
@@ -58,10 +60,12 @@
      */
     private final Map<ReplicaIdentifier, PartitionReplica> replicas = new HashMap<>();
     private final Object replicaSyncLock = new Object();
+    private final Set<Integer> nodeOwnedPartitions = new HashSet<>();
 
     public ReplicaManager(INcApplicationContext appCtx, Set<Integer> partitions) {
         this.appCtx = appCtx;
         this.partitions.addAll(partitions);
+        setNodeOwnedPartitions(appCtx);
     }
 
     @Override
@@ -154,6 +158,11 @@
         return new ArrayList<>(replicas.values());
     }
 
+    @Override
+    public boolean isPartitionOwner(int partition) {
+        return nodeOwnedPartitions.contains(partition);
+    }
+
     public void closePartitionResources(int partition) throws HyracksDataException {
         final IDatasetLifecycleManager datasetLifecycleManager = appCtx.getDatasetLifecycleManager();
         //TODO(mhubail) we can flush only datasets of the requested partition
@@ -171,4 +180,13 @@
         String nodeId = appCtx.getServiceContext().getNodeId();
         return id.getNodeId().equals(nodeId);
     }
+
+    private void setNodeOwnedPartitions(INcApplicationContext appCtx) {
+        ClusterPartition[] clusterPartitions =
+                appCtx.getMetadataProperties().getNodePartitions().get(appCtx.getServiceContext().getNodeId());
+        if (clusterPartitions != null) {
+            nodeOwnedPartitions.addAll(Arrays.stream(clusterPartitions).map(ClusterPartition::getPartitionId)
+                    .collect(Collectors.toList()));
+        }
+    }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java
index 801ca0b..f9230dd 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IIndexCheckpointManager.java
@@ -28,9 +28,23 @@
      * @param validComponentSequence
      * @param lsn
      * @param validComponentId
+     * @param masterNodeId
      * @throws HyracksDataException
      */
-    void init(long validComponentSequence, long lsn, long validComponentId) throws HyracksDataException;
+    void init(long validComponentSequence, long lsn, long validComponentId, String masterNodeId)
+            throws HyracksDataException;
+
+    /**
+     * Called when a new LSM disk component is flushed due to a replicated component.
+     * When called, the index checkpoint is updated with the latest valid {@code componentSequence}
+     * and low watermark {@code lsn}
+     *
+     * @param componentSequence
+     * @param lsn
+     * @param masterNodeId
+     * @throws HyracksDataException
+     */
+    void flushed(long componentSequence, long lsn, long componentId, String masterNodeId) throws HyracksDataException;
 
     /**
      * Called when a new LSM disk component is flushed. When called, the index checkpoint is updated
@@ -50,9 +64,11 @@
      * @param componentSequence
      * @param masterLsn
      * @param componentId
+     * @param masterNodeId
      * @throws HyracksDataException
      */
-    void replicated(long componentSequence, long masterLsn, long componentId) throws HyracksDataException;
+    void replicated(long componentSequence, long masterLsn, long componentId, String masterNodeId)
+            throws HyracksDataException;
 
     /**
      * Called when a flush log is received and replicated from master. The mapping between
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IReplicaManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IReplicaManager.java
index 4f46227..88d3113 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IReplicaManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IReplicaManager.java
@@ -100,4 +100,12 @@
      * @return the list of replicas
      */
     List<IPartitionReplica> getReplicas();
+
+    /**
+     * Returns true if {@code partition} is owned by this node, otherwise false.
+     *
+     * @param partition
+     * @return true if the partition is owned by this node, otherwise false.
+     */
+    boolean isPartitionOwner(int partition);
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java
index 24b9ae69..a2cf531 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexCheckpoint.java
@@ -44,8 +44,11 @@
     private long lowWatermark;
     private long lastComponentId;
     private Map<Long, Long> masterNodeFlushMap;
+    private String masterNodeId;
+    private long masterValidSeq;
 
-    public static IndexCheckpoint first(long lastComponentSequence, long lowWatermark, long validComponentId) {
+    public static IndexCheckpoint first(long lastComponentSequence, long lowWatermark, long validComponentId,
+            String masterNodeId) {
         IndexCheckpoint firstCheckpoint = new IndexCheckpoint();
         firstCheckpoint.id = INITIAL_CHECKPOINT_ID;
         firstCheckpoint.lowWatermark = lowWatermark;
@@ -53,11 +56,13 @@
         firstCheckpoint.lastComponentId = validComponentId;
         firstCheckpoint.masterNodeFlushMap = new HashMap<>();
         firstCheckpoint.masterNodeFlushMap.put(HAS_NULL_MISSING_VALUES_FIX, HAS_NULL_MISSING_VALUES_FIX);
+        firstCheckpoint.masterNodeId = masterNodeId;
+        firstCheckpoint.masterValidSeq = lastComponentSequence;
         return firstCheckpoint;
     }
 
     public static IndexCheckpoint next(IndexCheckpoint latest, long lowWatermark, long validComponentSequence,
-            long lastComponentId) {
+            long lastComponentId, String masterNodeId) {
         if (lowWatermark < latest.getLowWatermark()) {
             if (LOGGER.isErrorEnabled()) {
                 LOGGER.error("low watermark {} less than the latest checkpoint low watermark {}", lowWatermark, latest);
@@ -70,6 +75,13 @@
         next.lastComponentId = lastComponentId;
         next.validComponentSequence = validComponentSequence;
         next.masterNodeFlushMap = latest.getMasterNodeFlushMap();
+        if (masterNodeId != null) {
+            next.masterNodeId = masterNodeId;
+            next.masterValidSeq = validComponentSequence;
+        } else {
+            next.masterNodeId = latest.getMasterNodeId();
+            next.masterValidSeq = latest.getMasterValidSeq();
+        }
         // remove any lsn from the map that wont be used anymore
         next.masterNodeFlushMap.values().removeIf(lsn -> lsn < lowWatermark && lsn != HAS_NULL_MISSING_VALUES_FIX);
         return next;
@@ -111,6 +123,14 @@
         }
     }
 
+    public String getMasterNodeId() {
+        return masterNodeId;
+    }
+
+    public long getMasterValidSeq() {
+        return masterValidSeq;
+    }
+
     public static IndexCheckpoint fromJson(String json) throws HyracksDataException {
         try {
             return OBJECT_MAPPER.readValue(json, IndexCheckpoint.class);
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/ResourceReference.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/ResourceReference.java
index ebf212b..7065767 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/ResourceReference.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/ResourceReference.java
@@ -147,6 +147,10 @@
         return ResourceReference.ofIndex(relativePath.getParent().resolve(dataset).toFile().getPath());
     }
 
+    public boolean isMetadataResource() {
+        return getName().equals(StorageConstants.METADATA_FILE_NAME);
+    }
+
     public Path getFileRelativePath() {
         return relativePath.resolve(name);
     }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
index 6a080bd..2d231d3 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StorageConstants.java
@@ -37,6 +37,7 @@
      * begin with ".". Otherwise {@link AbstractLSMIndexFileManager} will try to
      * use them as index files.
      */
+    public static final String INDEX_NON_DATA_FILES_PREFIX = ".";
     public static final String INDEX_CHECKPOINT_FILE_PREFIX = ".idx_checkpoint_";
     public static final String METADATA_FILE_NAME = ".metadata";
     public static final String MASK_FILE_PREFIX = ".mask_";
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/CheckpointPartitionIndexesTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/CheckpointPartitionIndexesTask.java
index dac4a70..97b6556 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/CheckpointPartitionIndexesTask.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/CheckpointPartitionIndexesTask.java
@@ -48,10 +48,12 @@
 
     private final int partition;
     private final long maxComponentId;
+    private final String masterNodeId;
 
-    public CheckpointPartitionIndexesTask(int partition, long maxComponentId) {
+    public CheckpointPartitionIndexesTask(int partition, long maxComponentId, String masterNodeId) {
         this.partition = partition;
         this.maxComponentId = maxComponentId;
+        this.masterNodeId = masterNodeId;
     }
 
     @Override
@@ -66,7 +68,6 @@
         for (LocalResource ls : partitionResources) {
             DatasetResourceReference ref = DatasetResourceReference.of(ls);
             final IIndexCheckpointManager indexCheckpointManager = indexCheckpointManagerProvider.get(ref);
-            indexCheckpointManager.delete();
             // Get most recent sequence of existing files to avoid deletion
             Path indexPath = StoragePathUtil.getIndexPath(ioManager, ref);
             String[] files = indexPath.toFile().list(AbstractLSMIndexFileManager.COMPONENT_FILES_FILTER);
@@ -79,7 +80,11 @@
                 maxComponentSequence =
                         Math.max(maxComponentSequence, IndexComponentFileReference.of(file).getSequenceEnd());
             }
-            indexCheckpointManager.init(maxComponentSequence, currentLSN, maxComponentId);
+            if (indexCheckpointManager.getCheckpointCount() > 0) {
+                indexCheckpointManager.flushed(maxComponentSequence, currentLSN, maxComponentId, masterNodeId);
+            } else {
+                indexCheckpointManager.init(maxComponentSequence, currentLSN, maxComponentId, masterNodeId);
+            }
         }
         ReplicationProtocol.sendAck(worker.getChannel(), worker.getReusableBuffer());
     }
@@ -95,6 +100,11 @@
             DataOutputStream dos = new DataOutputStream(out);
             dos.writeInt(partition);
             dos.writeLong(maxComponentId);
+            boolean hasMaster = masterNodeId != null;
+            dos.writeBoolean(hasMaster);
+            if (hasMaster) {
+                dos.writeUTF(masterNodeId);
+            }
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
@@ -104,7 +114,9 @@
         try {
             int partition = input.readInt();
             long maxComponentId = input.readLong();
-            return new CheckpointPartitionIndexesTask(partition, maxComponentId);
+            final boolean hasMaster = input.readBoolean();
+            final String masterNodeId = hasMaster ? input.readUTF() : null;
+            return new CheckpointPartitionIndexesTask(partition, maxComponentId, masterNodeId);
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java
index b8f61d0..c92a527 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/MarkComponentValidTask.java
@@ -46,11 +46,13 @@
     private final long masterLsn;
     private final long lastComponentId;
     private final String file;
+    private final String masterNodeId;
 
-    public MarkComponentValidTask(String file, long masterLsn, long lastComponentId) {
+    public MarkComponentValidTask(String file, long masterLsn, long lastComponentId, String masterNodeId) {
         this.file = file;
         this.lastComponentId = lastComponentId;
         this.masterLsn = masterLsn;
+        this.masterNodeId = masterNodeId;
     }
 
     @Override
@@ -95,7 +97,7 @@
                 replicationTimeOut -= TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
             }
             final long componentSequence = IndexComponentFileReference.of(indexRef.getName()).getSequenceEnd();
-            indexCheckpointManager.replicated(componentSequence, masterLsn, lastComponentId);
+            indexCheckpointManager.replicated(componentSequence, masterLsn, lastComponentId, masterNodeId);
         }
     }
 
@@ -111,6 +113,11 @@
             dos.writeUTF(file);
             dos.writeLong(masterLsn);
             dos.writeLong(lastComponentId);
+            boolean hasMaster = masterNodeId != null;
+            dos.writeBoolean(hasMaster);
+            if (hasMaster) {
+                dos.writeUTF(masterNodeId);
+            }
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
@@ -120,6 +127,8 @@
         final String indexFile = input.readUTF();
         final long lsn = input.readLong();
         final long lastComponentId = input.readLong();
-        return new MarkComponentValidTask(indexFile, lsn, lastComponentId);
+        final boolean hasMaster = input.readBoolean();
+        final String masterNodeId = hasMaster ? input.readUTF() : null;
+        return new MarkComponentValidTask(indexFile, lsn, lastComponentId, masterNodeId);
     }
 }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListResponse.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListResponse.java
index 9c3902e..a9921c6 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListResponse.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListResponse.java
@@ -23,7 +23,9 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.asterix.replication.api.IReplicationMessage;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -31,11 +33,16 @@
 public class PartitionResourcesListResponse implements IReplicationMessage {
 
     private final int partition;
-    private final List<String> resources;
+    private Map<String, Long> partitionReplicatedResources;
+    private final List<String> files;
+    private final boolean owner;
 
-    public PartitionResourcesListResponse(int partition, List<String> resources) {
+    public PartitionResourcesListResponse(int partition, Map<String, Long> partitionReplicatedResources,
+            List<String> files, boolean owner) {
         this.partition = partition;
-        this.resources = resources;
+        this.partitionReplicatedResources = partitionReplicatedResources;
+        this.files = files;
+        this.owner = owner;
     }
 
     @Override
@@ -48,17 +55,23 @@
         try {
             DataOutputStream dos = new DataOutputStream(out);
             dos.writeInt(partition);
-            dos.writeInt(resources.size());
-            for (String file : resources) {
+            dos.writeInt(files.size());
+            for (String file : files) {
                 dos.writeUTF(file);
             }
+            dos.writeBoolean(owner);
+            dos.writeInt(partitionReplicatedResources.size());
+            for (Map.Entry<String, Long> stringLongEntry : partitionReplicatedResources.entrySet()) {
+                dos.writeUTF(stringLongEntry.getKey());
+                dos.writeLong(stringLongEntry.getValue());
+            }
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
     }
 
-    public List<String> getResources() {
-        return resources;
+    public List<String> getFiles() {
+        return files;
     }
 
     public static PartitionResourcesListResponse create(DataInput input) throws IOException {
@@ -68,6 +81,20 @@
         for (int i = 0; i < size; i++) {
             resources.add(input.readUTF());
         }
-        return new PartitionResourcesListResponse(partition, resources);
+        boolean owner = input.readBoolean();
+        int resourceSize = input.readInt();
+        Map<String, Long> partitionReplicatedResources = new HashMap<>();
+        for (int i = 0; i < resourceSize; i++) {
+            partitionReplicatedResources.put(input.readUTF(), input.readLong());
+        }
+        return new PartitionResourcesListResponse(partition, partitionReplicatedResources, resources, owner);
+    }
+
+    public boolean isOwner() {
+        return owner;
+    }
+
+    public Map<String, Long> getPartitionReplicatedResources() {
+        return partitionReplicatedResources;
     }
 }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListTask.java
index cff12de..3ea252f 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListTask.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/PartitionResourcesListTask.java
@@ -23,6 +23,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import org.apache.asterix.common.api.INcApplicationContext;
@@ -50,11 +51,15 @@
                 (PersistentLocalResourceRepository) appCtx.getLocalResourceRepository();
         localResourceRepository.cleanup(partition);
         final IReplicationStrategy replicationStrategy = appCtx.getReplicationManager().getReplicationStrategy();
-        final List<String> partitionResources =
+        // .metadata file -> resource id
+        Map<String, Long> partitionReplicatedResources =
+                localResourceRepository.getPartitionReplicatedResources(partition, replicationStrategy);
+        // all data files in partitions + .metadata files
+        final List<String> partitionFiles =
                 localResourceRepository.getPartitionReplicatedFiles(partition, replicationStrategy).stream()
                         .map(StoragePathUtil::getFileRelativePath).collect(Collectors.toList());
-        final PartitionResourcesListResponse response =
-                new PartitionResourcesListResponse(partition, partitionResources);
+        final PartitionResourcesListResponse response = new PartitionResourcesListResponse(partition,
+                partitionReplicatedResources, partitionFiles, appCtx.getReplicaManager().isPartitionOwner(partition));
         ReplicationProtocol.sendTo(worker.getChannel(), response, worker.getReusableBuffer());
     }
 
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/ReplicateFileTask.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/ReplicateFileTask.java
index 44c7bea..3ee3094 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/ReplicateFileTask.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/messaging/ReplicateFileTask.java
@@ -55,11 +55,13 @@
     private final String file;
     private final long size;
     private final boolean indexMetadata;
+    private final String masterNodeId;
 
-    public ReplicateFileTask(String file, long size, boolean indexMetadata) {
+    public ReplicateFileTask(String file, long size, boolean indexMetadata, String masterNodeId) {
         this.file = file;
         this.size = size;
         this.indexMetadata = indexMetadata;
+        this.masterNodeId = masterNodeId;
     }
 
     @Override
@@ -103,7 +105,7 @@
         final long currentLSN = appCtx.getTransactionSubsystem().getLogManager().getAppendLSN();
         indexCheckpointManager.delete();
         indexCheckpointManager.init(UNINITIALIZED_COMPONENT_SEQ, currentLSN,
-                LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID.getMaxId());
+                LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID.getMaxId(), masterNodeId);
         LOGGER.info(() -> "Checkpoint index: " + indexRef);
     }
 
@@ -119,6 +121,11 @@
             dos.writeUTF(file);
             dos.writeLong(size);
             dos.writeBoolean(indexMetadata);
+            boolean hasMaster = masterNodeId != null;
+            dos.writeBoolean(hasMaster);
+            if (hasMaster) {
+                dos.writeUTF(masterNodeId);
+            }
         } catch (IOException e) {
             throw HyracksDataException.create(e);
         }
@@ -128,7 +135,9 @@
         final String s = input.readUTF();
         final long i = input.readLong();
         final boolean isMetadata = input.readBoolean();
-        return new ReplicateFileTask(s, i, isMetadata);
+        final boolean hasMaster = input.readBoolean();
+        final String masterNodeId = hasMaster ? input.readUTF() : null;
+        return new ReplicateFileTask(s, i, isMetadata, masterNodeId);
     }
 
     @Override
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/FileSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/FileSynchronizer.java
index 7bb2858..813b293 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/FileSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/FileSynchronizer.java
@@ -55,7 +55,9 @@
             final IIOManager ioManager = appCtx.getIoManager();
             final ISocketChannel channel = replica.getChannel();
             final FileReference filePath = ioManager.resolve(file);
-            ReplicateFileTask task = new ReplicateFileTask(file, filePath.getFile().length(), metadata);
+            String masterNode = appCtx.getReplicaManager().isPartitionOwner(replica.getIdentifier().getPartition())
+                    ? appCtx.getServiceContext().getNodeId() : null;
+            ReplicateFileTask task = new ReplicateFileTask(file, filePath.getFile().length(), metadata, masterNode);
             LOGGER.info("attempting to replicate {} to replica {}", task, replica);
             ReplicationProtocol.sendTo(replica, task);
             // send the file itself
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java
index 47f872c..48ad5a5 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/IndexSynchronizer.java
@@ -94,8 +94,10 @@
         final FileSynchronizer fileSynchronizer = new FileSynchronizer(appCtx, replica);
         job.getJobFiles().stream().map(StoragePathUtil::getFileRelativePath).forEach(fileSynchronizer::replicate);
         // send mark component valid
-        MarkComponentValidTask markValidTask =
-                new MarkComponentValidTask(indexFile, getReplicatedComponentLsn(), getReplicatedComponentId());
+        String masterNode = appCtx.getReplicaManager().isPartitionOwner(replica.getIdentifier().getPartition())
+                ? appCtx.getServiceContext().getNodeId() : null;
+        MarkComponentValidTask markValidTask = new MarkComponentValidTask(indexFile, getReplicatedComponentLsn(),
+                getReplicatedComponentId(), masterNode);
         ReplicationProtocol.sendTo(replica, markValidTask);
         ReplicationProtocol.waitForAck(replica);
         LOGGER.debug("Replicated component ({}) to replica {}", indexFile, replica);
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaFilesSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaFilesSynchronizer.java
index b47fd39..3c93d17 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaFilesSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaFilesSynchronizer.java
@@ -20,26 +20,39 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 import org.apache.asterix.common.api.INcApplicationContext;
 import org.apache.asterix.common.replication.IReplicationStrategy;
+import org.apache.asterix.common.storage.IIndexCheckpointManager;
+import org.apache.asterix.common.storage.IndexCheckpoint;
+import org.apache.asterix.common.storage.ResourceReference;
 import org.apache.asterix.common.utils.StoragePathUtil;
 import org.apache.asterix.replication.api.PartitionReplica;
 import org.apache.asterix.replication.messaging.PartitionResourcesListResponse;
 import org.apache.asterix.replication.messaging.PartitionResourcesListTask;
 import org.apache.asterix.replication.messaging.ReplicationProtocol;
 import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.network.ISocketChannel;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager;
+import org.apache.hyracks.storage.am.lsm.common.impls.IndexComponentFileReference;
+import org.apache.hyracks.storage.common.LocalResource;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 /**
  * Ensures that the files between master and a replica are synchronized
  */
 public class ReplicaFilesSynchronizer {
 
+    private static final Logger LOGGER = LogManager.getLogger();
     private final PartitionReplica replica;
     private final INcApplicationContext appCtx;
 
@@ -50,31 +63,42 @@
 
     public void sync() throws IOException {
         final int partition = replica.getIdentifier().getPartition();
-        final Set<String> replicaFiles = getReplicaFiles(partition);
+        PartitionResourcesListResponse replicaResourceResponse = getReplicaFiles(partition);
+        Map<ResourceReference, Long> resourceReferenceLongMap = getValidReplicaResources(
+                replicaResourceResponse.getPartitionReplicatedResources(), replicaResourceResponse.isOwner());
+        // clean up files for invalid resources (deleted or recreated while the replica was down)
+        Set<String> deletedReplicaFiles =
+                cleanupReplicaInvalidResources(replicaResourceResponse, resourceReferenceLongMap);
         final PersistentLocalResourceRepository localResourceRepository =
                 (PersistentLocalResourceRepository) appCtx.getLocalResourceRepository();
         final IReplicationStrategy replicationStrategy = appCtx.getReplicationManager().getReplicationStrategy();
         final Set<String> masterFiles =
                 localResourceRepository.getPartitionReplicatedFiles(partition, replicationStrategy).stream()
                         .map(StoragePathUtil::getFileRelativePath).collect(Collectors.toSet());
-        // find files on master and not on replica
-        final List<String> replicaMissingFiles =
-                masterFiles.stream().filter(file -> !replicaFiles.contains(file)).collect(Collectors.toList());
-        replicateMissingFiles(replicaMissingFiles);
-        // find files on replica and not on master
-        final List<String> replicaInvalidFiles =
-                replicaFiles.stream().filter(file -> !masterFiles.contains(file)).collect(Collectors.toList());
-        deleteInvalidFiles(replicaInvalidFiles);
+        // exclude from the replica files the list of invalid deleted files
+        final Set<String> replicaFiles = new HashSet<>(replicaResourceResponse.getFiles());
+        replicaFiles.removeAll(deletedReplicaFiles);
+        syncMissingFiles(replicaFiles, masterFiles);
+        deleteReplicaExtraFiles(replicaFiles, masterFiles);
     }
 
-    private Set<String> getReplicaFiles(int partition) throws IOException {
-        final PartitionResourcesListTask replicaFilesRequest = new PartitionResourcesListTask(partition);
-        final ISocketChannel channel = replica.getChannel();
-        final ByteBuffer reusableBuffer = replica.getReusableBuffer();
-        ReplicationProtocol.sendTo(replica, replicaFilesRequest);
-        final PartitionResourcesListResponse response =
-                (PartitionResourcesListResponse) ReplicationProtocol.read(channel, reusableBuffer);
-        return new HashSet<>(response.getResources());
+    private void deleteReplicaExtraFiles(Set<String> replicaFiles, Set<String> masterFiles) {
+        final List<String> replicaInvalidFiles =
+                replicaFiles.stream().filter(file -> !masterFiles.contains(file)).collect(Collectors.toList());
+        if (!replicaInvalidFiles.isEmpty()) {
+            LOGGER.debug("deleting files not on current master {} on replica {}", replicaInvalidFiles,
+                    replica.getIdentifier());
+            deleteInvalidFiles(replicaInvalidFiles);
+        }
+    }
+
+    private void syncMissingFiles(Set<String> replicaFiles, Set<String> masterFiles) {
+        final List<String> replicaMissingFiles =
+                masterFiles.stream().filter(file -> !replicaFiles.contains(file)).collect(Collectors.toList());
+        if (!replicaMissingFiles.isEmpty()) {
+            LOGGER.debug("replicating missing files {} on replica {}", replicaMissingFiles, replica.getIdentifier());
+            replicateMissingFiles(replicaMissingFiles);
+        }
     }
 
     private void replicateMissingFiles(List<String> files) {
@@ -88,4 +112,73 @@
         final FileSynchronizer sync = new FileSynchronizer(appCtx, replica);
         files.forEach(sync::delete);
     }
+
+    private long getResourceMasterValidSeq(ResourceReference rr) throws HyracksDataException {
+        IIndexCheckpointManager iIndexCheckpointManager = appCtx.getIndexCheckpointManagerProvider().get(rr);
+        int checkpointCount = iIndexCheckpointManager.getCheckpointCount();
+        if (checkpointCount > 0) {
+            IndexCheckpoint latest = iIndexCheckpointManager.getLatest();
+            long masterValidSeq = latest.getMasterValidSeq();
+            LOGGER.info("setting resource {} valid component seq to {}", rr, masterValidSeq);
+            return masterValidSeq;
+        }
+        return AbstractLSMIndexFileManager.UNINITIALIZED_COMPONENT_SEQ;
+    }
+
+    private Set<String> cleanupReplicaInvalidResources(PartitionResourcesListResponse replicaResourceResponse,
+            Map<ResourceReference, Long> validReplicaResources) {
+        Set<String> invalidFiles = new HashSet<>();
+        for (String replicaResPath : replicaResourceResponse.getFiles()) {
+            ResourceReference replicaRes = ResourceReference.of(replicaResPath);
+            if (!validReplicaResources.containsKey(replicaRes)) {
+                LOGGER.debug("replica invalid file {} to be deleted", replicaRes.getFileRelativePath());
+                invalidFiles.add(replicaResPath);
+            } else if (replicaResourceResponse.isOwner() && !replicaRes.isMetadataResource()) {
+                // find files where the owner generated and failed before replicating
+                Long masterValidSeq = validReplicaResources.get(replicaRes);
+                IndexComponentFileReference componentFileReference =
+                        IndexComponentFileReference.of(replicaRes.getName());
+                if (componentFileReference.getSequenceStart() > masterValidSeq
+                        || componentFileReference.getSequenceEnd() > masterValidSeq) {
+                    LOGGER.debug("will ask replica {} to delete file {} based on valid master valid seq {}",
+                            replica.getIdentifier(), replicaResPath, masterValidSeq);
+                    invalidFiles.add(replicaResPath);
+                }
+            }
+        }
+        if (!invalidFiles.isEmpty()) {
+            LOGGER.info("will delete the following files from replica {}", invalidFiles);
+            deleteInvalidFiles(new ArrayList<>(invalidFiles));
+        }
+        return invalidFiles;
+    }
+
+    private PartitionResourcesListResponse getReplicaFiles(int partition) throws IOException {
+        final PartitionResourcesListTask replicaFilesRequest = new PartitionResourcesListTask(partition);
+        final ISocketChannel channel = replica.getChannel();
+        final ByteBuffer reusableBuffer = replica.getReusableBuffer();
+        ReplicationProtocol.sendTo(replica, replicaFilesRequest);
+        return (PartitionResourcesListResponse) ReplicationProtocol.read(channel, reusableBuffer);
+    }
+
+    private Map<ResourceReference, Long> getValidReplicaResources(Map<String, Long> partitionReplicatedResources,
+            boolean owner) throws HyracksDataException {
+        Map<ResourceReference, Long> resource2ValidSeqMap = new HashMap<>();
+        for (Map.Entry<String, Long> resourceEntry : partitionReplicatedResources.entrySet()) {
+            ResourceReference rr = ResourceReference.of(resourceEntry.getKey());
+            final PersistentLocalResourceRepository localResourceRepository =
+                    (PersistentLocalResourceRepository) appCtx.getLocalResourceRepository();
+            LocalResource localResource = localResourceRepository.get(rr.getRelativePath().toString());
+            if (localResource != null) {
+                if (localResource.getId() != resourceEntry.getValue()) {
+                    LOGGER.info("replica has resource {} but with different resource id; ours {}, theirs {}", rr,
+                            localResource.getId(), resourceEntry.getValue());
+                } else {
+                    long resourceMasterValidSeq = owner ? getResourceMasterValidSeq(rr) : Integer.MAX_VALUE;
+                    resource2ValidSeqMap.put(rr, resourceMasterValidSeq);
+                }
+            }
+        }
+        return resource2ValidSeqMap;
+    }
 }
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
index 3209a98..6030245 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/sync/ReplicaSynchronizer.java
@@ -71,8 +71,10 @@
 
     private void checkpointReplicaIndexes() throws IOException {
         final int partition = replica.getIdentifier().getPartition();
+        String masterNode =
+                appCtx.getReplicaManager().isPartitionOwner(partition) ? appCtx.getServiceContext().getNodeId() : null;
         CheckpointPartitionIndexesTask task =
-                new CheckpointPartitionIndexesTask(partition, getPartitionMaxComponentId(partition));
+                new CheckpointPartitionIndexesTask(partition, getPartitionMaxComponentId(partition), masterNode);
         ReplicationProtocol.sendTo(replica, task);
         ReplicationProtocol.waitForAck(replica);
     }
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 a73b71a..2494893 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
@@ -19,7 +19,7 @@
 package org.apache.asterix.transaction.management.resource;
 
 import static org.apache.asterix.common.storage.ResourceReference.getComponentSequence;
-import static org.apache.asterix.common.utils.StorageConstants.INDEX_CHECKPOINT_FILE_PREFIX;
+import static org.apache.asterix.common.utils.StorageConstants.INDEX_NON_DATA_FILES_PREFIX;
 import static org.apache.asterix.common.utils.StorageConstants.METADATA_FILE_NAME;
 import static org.apache.hyracks.api.exceptions.ErrorCode.CANNOT_CREATE_FILE;
 import static org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexFileManager.COMPONENT_FILES_FILTER;
@@ -90,7 +90,7 @@
     private static final String METADATA_FILE_MASK_NAME =
             StorageConstants.MASK_FILE_PREFIX + StorageConstants.METADATA_FILE_NAME;
     private static final FilenameFilter LSM_INDEX_FILES_FILTER =
-            (dir, name) -> !name.startsWith(INDEX_CHECKPOINT_FILE_PREFIX);
+            (dir, name) -> name.startsWith(METADATA_FILE_NAME) || !name.startsWith(INDEX_NON_DATA_FILES_PREFIX);
     private static final FilenameFilter MASK_FILES_FILTER =
             (dir, name) -> name.startsWith(StorageConstants.MASK_FILE_PREFIX);
     private static final int MAX_CACHED_RESOURCES = 1000;
@@ -200,7 +200,7 @@
             byte[] bytes = OBJECT_MAPPER.writeValueAsBytes(resource.toJson(persistedResourceRegistry));
             FileUtil.writeAndForce(Paths.get(resourceFile.getAbsolutePath()), bytes);
             indexCheckpointManagerProvider.get(DatasetResourceReference.of(resource)).init(UNINITIALIZED_COMPONENT_SEQ,
-                    0, LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID.getMaxId());
+                    0, LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID.getMaxId(), null);
             deleteResourceFileMask(resourceFile);
         } catch (Exception e) {
             cleanup(resourceFile);
@@ -400,6 +400,20 @@
         });
     }
 
+    public Map<String, Long> getPartitionReplicatedResources(int partition, IReplicationStrategy strategy)
+            throws HyracksDataException {
+        final Map<String, Long> partitionReplicatedResources = new HashMap<>();
+        final Map<Long, LocalResource> partitionResources = getPartitionResources(partition);
+        for (LocalResource lr : partitionResources.values()) {
+            DatasetLocalResource datasetLocalResource = (DatasetLocalResource) lr.getResource();
+            if (strategy.isMatch(datasetLocalResource.getDatasetId())) {
+                DatasetResourceReference drr = DatasetResourceReference.of(lr);
+                partitionReplicatedResources.put(drr.getFileRelativePath().toString(), lr.getId());
+            }
+        }
+        return partitionReplicatedResources;
+    }
+
     public List<String> getPartitionReplicatedFiles(int partition, IReplicationStrategy strategy)
             throws HyracksDataException {
         final List<String> partitionReplicatedFiles = new ArrayList<>();