diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
index 72a9b44..352a5f8 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -69,6 +69,8 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
+import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
+import org.apache.hyracks.algebricks.runtime.operators.std.AssignRuntimeFactory;
 import org.apache.hyracks.algebricks.runtime.operators.std.EmptyTupleSourceRuntimeFactory;
 import org.apache.hyracks.api.comm.IFrameWriter;
 import org.apache.hyracks.api.comm.VSizeFrame;
@@ -89,6 +91,7 @@
 import org.apache.hyracks.api.util.HyracksConstants;
 import org.apache.hyracks.dataflow.common.utils.TaskUtil;
 import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider;
+import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
 import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
 import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorNodePushable;
 import org.apache.hyracks.storage.am.common.api.IIndexBuilder;
@@ -166,33 +169,93 @@
     public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx,
             Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType, int[] filterFields,
             int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators,
-            StorageComponentProvider storageComponentProvider)
+            StorageComponentProvider storageComponentProvider, Index secondaryIndex)
             throws AlgebricksException, HyracksDataException, RemoteException, ACIDException {
-        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-        org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
-                DatasetUtil.getMergePolicyFactory(dataset, mdTxnCtx);
-        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-        PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
-                mergePolicy.first, mergePolicy.second, filterFields, primaryKeyIndexes, primaryKeyIndicators);
-        IndexOperation op = IndexOperation.INSERT;
-        IModificationOperationCallbackFactory modOpCallbackFactory =
-                new PrimaryIndexModificationOperationCallbackFactory(getTxnJobId(ctx), dataset.getDatasetId(),
-                        primaryIndexInfo.primaryKeyIndexes, TXN_SUBSYSTEM_PROVIDER, Operation.get(op),
-                        ResourceType.LSM_BTREE);
-        IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
-        IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
-                storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
-        LSMInsertDeleteOperatorNodePushable insertOp =
-                new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
-                        primaryIndexInfo.primaryIndexInsertFieldsPermutations,
-                        recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
-                        op, true, indexHelperFactory, modOpCallbackFactory, null);
-        CommitRuntime commitOp =
-                new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), primaryIndexInfo.primaryKeyIndexes,
-                        true, ctx.getTaskAttemptId().getTaskId().getPartition(), true);
-        insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
-        commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
-        return Pair.of(insertOp, commitOp);
+        CcApplicationContext appCtx =
+                (CcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext();
+        MetadataProvider mdProvider = new MetadataProvider(appCtx, null);
+        try {
+            MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+            org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
+                    DatasetUtil.getMergePolicyFactory(dataset, mdTxnCtx);
+            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+            PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+                    mergePolicy.first, mergePolicy.second, filterFields, primaryKeyIndexes, primaryKeyIndicators);
+            IndexOperation op = IndexOperation.INSERT;
+            IModificationOperationCallbackFactory modOpCallbackFactory =
+                    new PrimaryIndexModificationOperationCallbackFactory(getTxnJobId(ctx), dataset.getDatasetId(),
+                            primaryIndexInfo.primaryKeyIndexes, TXN_SUBSYSTEM_PROVIDER, Operation.get(op),
+                            ResourceType.LSM_BTREE);
+            IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
+            RecordDescriptor recordDesc =
+                    recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0);
+            IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
+                    storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+            LSMInsertDeleteOperatorNodePushable insertOp =
+                    new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                            primaryIndexInfo.primaryIndexInsertFieldsPermutations, recordDesc, op, true,
+                            indexHelperFactory, modOpCallbackFactory, null);
+
+            // For now, this assumes a single secondary index. recordDesc is always <pk-record-meta>
+            // for the index, we will have to create an assign operator that extract the sk
+            // then the secondary LSMInsertDeleteOperatorNodePushable
+            if (secondaryIndex != null) {
+                List<List<String>> skNames = secondaryIndex.getKeyFieldNames();
+                List<Integer> indicators = secondaryIndex.getKeyFieldSourceIndicators();
+                IScalarEvaluatorFactory[] secondaryFieldAccessEvalFactories =
+                        new IScalarEvaluatorFactory[skNames.size()];
+                for (int i = 0; i < skNames.size(); i++) {
+                    ARecordType sourceType = dataset.hasMetaPart()
+                            ? indicators.get(i).intValue() == Index.RECORD_INDICATOR ? recordType : metaType
+                            : recordType;
+                    int pos = skNames.get(i).size() > 1 ? -1 : sourceType.getFieldIndex(skNames.get(i).get(0));
+                    secondaryFieldAccessEvalFactories[i] = mdProvider.getDataFormat().getFieldAccessEvaluatorFactory(
+                            mdProvider.getFunctionManager(), sourceType, secondaryIndex.getKeyFieldNames().get(i), pos);
+                }
+                // outColumns are computed inside the assign runtime
+                int[] outColumns = new int[skNames.size()];
+                // projection list include old and new (primary and secondary keys)
+                int[] projectionList = new int[skNames.size() + primaryIndexInfo.index.getKeyFieldNames().size()];
+                for (int i = 0; i < secondaryFieldAccessEvalFactories.length; i++) {
+                    outColumns[i] = primaryIndexInfo.rDesc.getFieldCount() + i;
+                }
+                int projCount = 0;
+                for (int i = 0; i < secondaryFieldAccessEvalFactories.length; i++) {
+                    projectionList[projCount++] = primaryIndexInfo.rDesc.getFieldCount() + i;
+                }
+                for (int i = 0; i < primaryIndexInfo.index.getKeyFieldNames().size(); i++) {
+                    projectionList[projCount++] = i;
+                }
+                IPushRuntime assignOp =
+                        new AssignRuntimeFactory(outColumns, secondaryFieldAccessEvalFactories, projectionList, true)
+                                .createPushRuntime(ctx);
+                insertOp.setOutputFrameWriter(0, assignOp, primaryIndexInfo.rDesc);
+                assignOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
+                SecondaryIndexInfo secondaryIndexInfo = new SecondaryIndexInfo(primaryIndexInfo, secondaryIndex);
+                IIndexDataflowHelperFactory secondaryIndexHelperFactory = new IndexDataflowHelperFactory(
+                        storageComponentProvider.getStorageManager(), secondaryIndexInfo.fileSplitProvider);
+                LSMInsertDeleteOperatorNodePushable secondaryInsertOp =
+                        new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                                secondaryIndexInfo.insertFieldsPermutations, secondaryIndexInfo.rDesc, op, false,
+                                secondaryIndexHelperFactory, NoOpOperationCallbackFactory.INSTANCE, null);
+                assignOp.setOutputFrameWriter(0, secondaryInsertOp, secondaryIndexInfo.rDesc);
+                CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
+                        secondaryIndexInfo.primaryKeyIndexes, true, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                        true);
+                secondaryInsertOp.setOutputFrameWriter(0, commitOp, secondaryIndexInfo.rDesc);
+                commitOp.setInputRecordDescriptor(0, secondaryIndexInfo.rDesc);
+                return Pair.of(insertOp, commitOp);
+            } else {
+                CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
+                        primaryIndexInfo.primaryKeyIndexes, true, ctx.getTaskAttemptId().getTaskId().getPartition(),
+                        true);
+                insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
+                commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
+                return Pair.of(insertOp, commitOp);
+            }
+        } finally {
+            mdProvider.getLocks().unlock();
+        }
     }
 
     public IPushRuntime getFullScanPipeline(IFrameWriter countOp, IHyracksTaskContext ctx, Dataset dataset,
@@ -271,6 +334,34 @@
         return primaryIndexInfo;
     }
 
+    public SecondaryIndexInfo createSecondaryIndex(PrimaryIndexInfo primaryIndexInfo, Index secondaryIndex,
+            IStorageComponentProvider storageComponentProvider, int partition)
+            throws AlgebricksException, HyracksDataException, RemoteException, ACIDException {
+        MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+        org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
+                DatasetUtil.getMergePolicyFactory(primaryIndexInfo.dataset, mdTxnCtx);
+        MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+        Dataverse dataverse = new Dataverse(primaryIndexInfo.dataset.getDataverseName(),
+                NonTaggedDataFormat.class.getName(), MetadataUtil.PENDING_NO_OP);
+        MetadataProvider mdProvider = new MetadataProvider(
+                (ICcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext(), dataverse);
+        SecondaryIndexInfo secondaryIndexInfo = new SecondaryIndexInfo(primaryIndexInfo, secondaryIndex);
+        try {
+
+            IResourceFactory resourceFactory = primaryIndexInfo.dataset.getResourceFactory(mdProvider, secondaryIndex,
+                    primaryIndexInfo.recordType, primaryIndexInfo.metaType, mergePolicy.first, mergePolicy.second);
+            IndexBuilderFactory indexBuilderFactory =
+                    new IndexBuilderFactory(storageComponentProvider.getStorageManager(),
+                            secondaryIndexInfo.fileSplitProvider, resourceFactory, true);
+            IHyracksTaskContext ctx = createTestContext(newJobId(), partition, false);
+            IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, partition);
+            indexBuilder.build();
+        } finally {
+            mdProvider.getLocks().unlock();
+        }
+        return secondaryIndexInfo;
+    }
+
     public static ISerializerDeserializer<?>[] createPrimaryIndexSerdes(int primaryIndexNumOfTupleFields,
             IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType) {
         int i = 0;
@@ -286,6 +377,22 @@
         return primaryIndexSerdes;
     }
 
+    public static ISerializerDeserializer<?>[] createSecondaryIndexSerdes(ARecordType recordType, ARecordType metaType,
+            IAType[] primaryKeyTypes, IAType[] secondaryKeyTypes) {
+        ISerializerDeserializer<?>[] secondaryIndexSerdes =
+                new ISerializerDeserializer<?>[secondaryKeyTypes.length + primaryKeyTypes.length];
+        int i = 0;
+        for (; i < secondaryKeyTypes.length; i++) {
+            secondaryIndexSerdes[i] =
+                    SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(secondaryKeyTypes[i]);
+        }
+        for (; i < primaryKeyTypes.length; i++) {
+            secondaryIndexSerdes[i] =
+                    SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(primaryKeyTypes[i]);
+        }
+        return secondaryIndexSerdes;
+    }
+
     public static ITypeTraits[] createPrimaryIndexTypeTraits(int primaryIndexNumOfTupleFields, IAType[] primaryKeyTypes,
             ARecordType recordType, ARecordType metaType) {
         ITypeTraits[] primaryIndexTypeTraits = new ITypeTraits[primaryIndexNumOfTupleFields];
@@ -300,6 +407,19 @@
         return primaryIndexTypeTraits;
     }
 
+    public static ITypeTraits[] createSecondaryIndexTypeTraits(ARecordType recordType, ARecordType metaType,
+            IAType[] primaryKeyTypes, IAType[] secondaryKeyTypes) {
+        ITypeTraits[] secondaryIndexTypeTraits = new ITypeTraits[secondaryKeyTypes.length + primaryKeyTypes.length];
+        int i = 0;
+        for (; i < secondaryKeyTypes.length; i++) {
+            secondaryIndexTypeTraits[i] = TypeTraitProvider.INSTANCE.getTypeTrait(secondaryKeyTypes[i]);
+        }
+        for (; i < primaryKeyTypes.length; i++) {
+            secondaryIndexTypeTraits[i] = TypeTraitProvider.INSTANCE.getTypeTrait(primaryKeyTypes[i]);
+        }
+        return secondaryIndexTypeTraits;
+    }
+
     public IHyracksTaskContext createTestContext(JobId jobId, int partition, boolean withMessaging)
             throws HyracksDataException {
         IHyracksTaskContext ctx = TestUtils.create(KB32);
@@ -337,7 +457,47 @@
         return (DatasetLifecycleManager) getAppRuntimeContext().getDatasetLifecycleManager();
     }
 
+    public static class SecondaryIndexInfo {
+        private int[] primaryKeyIndexes;
+        private PrimaryIndexInfo primaryIndexInfo;
+        private Index secondaryIndex;
+        private ConstantFileSplitProvider fileSplitProvider;
+        private RecordDescriptor rDesc;
+        private int[] insertFieldsPermutations;
+
+        public SecondaryIndexInfo(PrimaryIndexInfo primaryIndexInfo, Index secondaryIndex) {
+            this.primaryIndexInfo = primaryIndexInfo;
+            this.secondaryIndex = secondaryIndex;
+            List<String> nodes = Collections.singletonList(ExecutionTestUtil.integrationUtil.ncs[0].getId());
+            CcApplicationContext appCtx =
+                    (CcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext();
+            FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(),
+                    primaryIndexInfo.dataset, secondaryIndex.getIndexName(), nodes);
+            fileSplitProvider = new ConstantFileSplitProvider(splits);
+            ITypeTraits[] secondaryIndexTypeTraits = createSecondaryIndexTypeTraits(primaryIndexInfo.recordType,
+                    primaryIndexInfo.metaType, primaryIndexInfo.primaryKeyTypes,
+                    secondaryIndex.getKeyFieldTypes().toArray(new IAType[secondaryIndex.getKeyFieldTypes().size()]));
+            ISerializerDeserializer<?>[] secondaryIndexSerdes = createSecondaryIndexSerdes(primaryIndexInfo.recordType,
+                    primaryIndexInfo.metaType, primaryIndexInfo.primaryKeyTypes,
+                    secondaryIndex.getKeyFieldTypes().toArray(new IAType[secondaryIndex.getKeyFieldTypes().size()]));
+            rDesc = new RecordDescriptor(secondaryIndexSerdes, secondaryIndexTypeTraits);
+            insertFieldsPermutations = new int[secondaryIndexTypeTraits.length];
+            for (int i = 0; i < insertFieldsPermutations.length; i++) {
+                insertFieldsPermutations[i] = i;
+            }
+            primaryKeyIndexes = new int[primaryIndexInfo.primaryKeyIndexes.length];
+            for (int i = 0; i < primaryKeyIndexes.length; i++) {
+                primaryKeyIndexes[i] = i + secondaryIndex.getKeyFieldNames().size();
+            }
+        }
+
+        public IFileSplitProvider getFileSplitProvider() {
+            return fileSplitProvider;
+        }
+    }
+
     public static class PrimaryIndexInfo {
+        private Dataset dataset;
         private IAType[] primaryKeyTypes;
         private ARecordType recordType;
         private ARecordType metaType;
@@ -356,6 +516,7 @@
                 ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
                 int[] filterFields, int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators)
                 throws AlgebricksException {
+            this.dataset = dataset;
             this.primaryKeyTypes = primaryKeyTypes;
             this.recordType = recordType;
             this.metaType = metaType;
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
index a239210..e923d93 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
@@ -18,6 +18,7 @@
  */
 package org.apache.asterix.test.dataflow;
 
+import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -27,6 +28,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
 import java.util.function.Predicate;
 
 import org.apache.asterix.app.bootstrap.TestNodeController;
@@ -37,6 +39,7 @@
 import org.apache.asterix.app.nc.NCAppRuntimeContext;
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
 import org.apache.asterix.common.transactions.ITransactionContext;
@@ -67,9 +70,11 @@
 import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
 import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
 import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
 import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
@@ -112,12 +117,24 @@
     private static IIndexDataflowHelper indexDataflowHelper;
     private static ITransactionContext txnCtx;
     private static LSMInsertDeleteOperatorNodePushable insertOp;
+    public static final ITestOpCallback<Semaphore> ALLOW_CALLBACK = new ITestOpCallback<Semaphore>() {
+        @Override
+        public void before(Semaphore smeaphore) {
+            smeaphore.release();
+        }
+
+        @Override
+        public void after() {
+        }
+    };
 
     @BeforeClass
     public static void setUp() throws Exception {
         System.out.println("SetUp: ");
         TestHelper.deleteExistingInstanceFiles();
-        nc = new TestNodeController(null, false);
+        String configPath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "test"
+                + File.separator + "resources" + File.separator + "multi-partition-test-configuration.xml";
+        nc = new TestNodeController(configPath, false);
         nc.init();
         ncAppCtx = nc.getAppRuntimeContext();
         dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
@@ -153,7 +170,7 @@
         txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                 new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
         insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
-                KEY_INDICATORS_LIST, storageManager).getLeft();
+                KEY_INDICATORS_LIST, storageManager, null).getLeft();
     }
 
     @After
@@ -162,10 +179,10 @@
     }
 
     static void allowAllOps(TestLsmBtree lsmBtree) {
-        lsmBtree.addModifyCallback(sem -> sem.release());
-        lsmBtree.addFlushCallback(sem -> sem.release());
-        lsmBtree.addSearchCallback(sem -> sem.release());
-        lsmBtree.addMergeCallback(sem -> sem.release());
+        lsmBtree.addModifyCallback(ALLOW_CALLBACK);
+        lsmBtree.addFlushCallback(ALLOW_CALLBACK);
+        lsmBtree.addSearchCallback(ALLOW_CALLBACK);
+        lsmBtree.addMergeCallback(ALLOW_CALLBACK);
     }
 
     @Test
@@ -184,7 +201,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -222,6 +239,16 @@
         }
     }
 
+    public void flush(boolean async) throws Exception {
+        flush(dsLifecycleMgr, lsmBtree, dataset, async);
+    }
+
+    public static void flush(IDatasetLifecycleManager dsLifecycleMgr, TestLsmBtree lsmBtree, Dataset dataset,
+            boolean async) throws Exception {
+        waitForOperations(lsmBtree);
+        dsLifecycleMgr.flushDataset(dataset.getDatasetId(), async);
+    }
+
     @Test
     public void testRollbackThenInsert() {
         try {
@@ -238,7 +265,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -267,7 +294,7 @@
             txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                     new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
             insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
-                    KEY_INDICATORS_LIST, storageManager).getLeft();
+                    KEY_INDICATORS_LIST, storageManager, null).getLeft();
             insertOp.open();
             for (int j = 0; j < RECORDS_PER_COMPONENT; j++) {
                 ITupleReference tuple = tupleGenerator.next();
@@ -311,7 +338,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -339,7 +366,7 @@
             lsmAccessor.deleteComponents(
                     c -> (c instanceof ILSMMemoryComponent && ((ILSMMemoryComponent) c).isModified()));
             // now that the rollback has completed, we will unblock the search
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             // search now and ensure
@@ -359,7 +386,7 @@
             DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
             lsmAccessor.deleteComponents(pred);
             // now that the rollback has completed, we will unblock the search
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(secondSearcher.result());
             searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
@@ -385,7 +412,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -404,7 +431,7 @@
             // disable flushes
             lsmBtree.clearFlushCallbacks();
             Flusher firstFlusher = new Flusher(lsmBtree);
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
+            flush(true);
             firstFlusher.waitUntilCount(1);
             // now that we enetered, we will rollback. This will not proceed since it is waiting for the flush to complete
             Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate);
@@ -441,7 +468,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -505,7 +532,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -526,7 +553,7 @@
             lsmBtree.clearFlushCallbacks();
             lsmBtree.clearSearchCallbacks();
             Flusher firstFlusher = new Flusher(lsmBtree);
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
+            flush(true);
             firstFlusher.waitUntilCount(1);
             Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
@@ -535,7 +562,7 @@
             Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate);
             //unblock the flush
             lsmBtree.allowFlush(1);
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             // ensure current mem component is not modified
@@ -565,7 +592,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -585,7 +612,7 @@
             // disable searches
             lsmBtree.clearFlushCallbacks();
             Flusher firstFlusher = new Flusher(lsmBtree);
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
+            flush(true);
             firstFlusher.waitUntilCount(1);
             lsmBtree.clearSearchCallbacks();
             Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
@@ -594,7 +621,7 @@
             // now that we enetered, we will rollback
             Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate);
             // The rollback will be waiting for the flush to complete
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             //unblock the flush
@@ -627,7 +654,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -666,7 +693,7 @@
             // unblock the merge
             lsmBtree.allowMerge(1);
             // unblock the search
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             rollerback.complete();
@@ -698,7 +725,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    flush(false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -734,7 +761,7 @@
             // now that we enetered, we will rollback
             Rollerback rollerBack = new Rollerback(lsmBtree, new DiskComponentLsnPredicate(lsn));
             // unblock the search
-            lsmBtree.addSearchCallback(sem -> sem.release());
+            lsmBtree.addSearchCallback(ALLOW_CALLBACK);
             lsmBtree.allowSearch(1);
             Assert.assertTrue(firstSearcher.result());
             // even though rollback has been called, it is still waiting for the merge to complete
@@ -790,10 +817,18 @@
 
         public Searcher(TestNodeController nc, int partition, Dataset dataset, StorageComponentProvider storageManager,
                 TestLsmBtree lsmBtree, int numOfRecords) {
-            lsmBtree.addSearchCallback(sem -> {
-                synchronized (Searcher.this) {
-                    entered = true;
-                    Searcher.this.notifyAll();
+            lsmBtree.addSearchCallback(new ITestOpCallback<Semaphore>() {
+
+                @Override
+                public void before(Semaphore sem) {
+                    synchronized (Searcher.this) {
+                        entered = true;
+                        Searcher.this.notifyAll();
+                    }
+                }
+
+                @Override
+                public void after() {
                 }
             });
             Callable<Boolean> callable = new Callable<Boolean>() {
@@ -821,10 +856,18 @@
         private volatile int count = 0;
 
         public Merger(TestLsmBtree lsmBtree) {
-            lsmBtree.addMergeCallback(sem -> {
-                synchronized (Merger.this) {
-                    count++;
-                    Merger.this.notifyAll();
+            lsmBtree.addMergeCallback(new ITestOpCallback<Semaphore>() {
+
+                @Override
+                public void before(Semaphore smeaphore) {
+                    synchronized (Merger.this) {
+                        count++;
+                        Merger.this.notifyAll();
+                    }
+                }
+
+                @Override
+                public void after() {
                 }
             });
         }
@@ -840,10 +883,18 @@
         private volatile int count = 0;
 
         public Flusher(TestLsmBtree lsmBtree) {
-            lsmBtree.addFlushCallback(sem -> {
-                synchronized (Flusher.this) {
-                    count++;
-                    Flusher.this.notifyAll();
+            lsmBtree.addFlushCallback(new ITestOpCallback<Semaphore>() {
+
+                @Override
+                public void before(Semaphore smeaphore) {
+                    synchronized (Flusher.this) {
+                        count++;
+                        Flusher.this.notifyAll();
+                    }
+                }
+
+                @Override
+                public void after() {
                 }
             });
         }
@@ -889,6 +940,20 @@
         Assert.assertEquals(numOfRecords, countOp.getCount());
     }
 
+    public static void waitForOperations(ILSMIndex index) throws InterruptedException {
+        // wait until number of activeOperation reaches 0
+        PrimaryIndexOperationTracker opTracker = (PrimaryIndexOperationTracker) index.getOperationTracker();
+        long maxWaitTime = 60000L; // 1 minute
+        long before = System.currentTimeMillis();
+        while (opTracker.getNumActiveOperations() > 0) {
+            Thread.sleep(5); // NOSONAR: Test code with a timeout
+            if (System.currentTimeMillis() - before > maxWaitTime) {
+                throw new IllegalStateException(
+                        (System.currentTimeMillis() - before) + "ms passed without completing the frame operation");
+            }
+        }
+    }
+
     public static TestTupleCounterFrameWriter create(RecordDescriptor recordDescriptor,
             Collection<FrameWriterOperation> exceptionThrowingOperations,
             Collection<FrameWriterOperation> errorThrowingOperations, boolean deepCopyInputFrames) {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
index 963cded..0a968c8 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
@@ -124,7 +124,7 @@
                 ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                         new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
-                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft();
+                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager, null).getLeft();
                 insertOp.open();
                 TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
                         RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
index 2c8141ce..c12f50c 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
@@ -18,18 +18,23 @@
  */
 package org.apache.asterix.test.dataflow;
 
+import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.asterix.app.bootstrap.TestNodeController;
 import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo;
+import org.apache.asterix.app.bootstrap.TestNodeController.SecondaryIndexInfo;
 import org.apache.asterix.app.data.gen.TupleGenerator;
 import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
 import org.apache.asterix.app.nc.NCAppRuntimeContext;
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
 import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.config.DatasetConfig.IndexType;
 import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.ITransactionManager;
@@ -44,14 +49,21 @@
 import org.apache.asterix.om.types.BuiltinType;
 import org.apache.asterix.om.types.IAType;
 import org.apache.asterix.test.common.TestHelper;
+import org.apache.commons.lang3.mutable.MutableBoolean;
 import org.apache.hyracks.api.comm.VSizeFrame;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.util.SingleThreadEventProcessor;
 import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
 import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
 import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
 import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
+import org.apache.hyracks.storage.am.lsm.btree.impl.IVirtualBufferCacheCallback;
 import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -59,8 +71,18 @@
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
+@RunWith(Parameterized.class)
 public class MultiPartitionLSMIndexTest {
+    static final int REPREAT_TEST_COUNT = 1;
+
+    @Parameterized.Parameters
+    public static List<Object[]> data() {
+        return Arrays.asList(new Object[REPREAT_TEST_COUNT][0]);
+    }
+
     private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 };
     private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" },
             new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false);
@@ -73,30 +95,42 @@
     private static final int[] KEY_INDEXES = { 0 };
     private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR };
     private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR });
-    private static final int TOTAL_NUM_OF_RECORDS = 10000;
-    private static final int RECORDS_PER_COMPONENT = 1000;
+    private static final int TOTAL_NUM_OF_RECORDS = 5000;
+    private static final int RECORDS_PER_COMPONENT = 500;
     private static final int DATASET_ID = 101;
     private static final String DATAVERSE_NAME = "TestDV";
     private static final String DATASET_NAME = "TestDS";
+    private static final String INDEX_NAME = "TestIdx";
     private static final String DATA_TYPE_NAME = "DUMMY";
     private static final String NODE_GROUP_NAME = "DEFAULT";
+    private static final IndexType INDEX_TYPE = IndexType.BTREE;
+    private static final List<List<String>> INDEX_FIELD_NAMES =
+            Arrays.asList(Arrays.asList(RECORD_TYPE.getFieldNames()[1]));
+    private static final List<Integer> INDEX_FIELD_INDICATORS = Arrays.asList(Index.RECORD_INDICATOR);
+    private static final List<IAType> INDEX_FIELD_TYPES = Arrays.asList(BuiltinType.AINT64);
     private static final StorageComponentProvider storageManager = new StorageComponentProvider();
     private static final int NUM_PARTITIONS = 2;
     private static TestNodeController nc;
     private static NCAppRuntimeContext ncAppCtx;
     private static IDatasetLifecycleManager dsLifecycleMgr;
     private static Dataset dataset;
+    private static Index secondaryIndex;
     private static ITransactionContext txnCtx;
-    private static TestLsmBtree[] primarylsmBtrees;
+    private static TestLsmBtree[] primaryLsmBtrees;
+    private static TestLsmBtree[] secondaryLsmBtrees;
     private static IHyracksTaskContext[] taskCtxs;
-    private static IIndexDataflowHelper[] indexDataflowHelpers;
+    private static IIndexDataflowHelper[] primaryIndexDataflowHelpers;
+    private static IIndexDataflowHelper[] secondaryIndexDataflowHelpers;
     private static LSMInsertDeleteOperatorNodePushable[] insertOps;
+    private static Actor[] actors;
 
     @BeforeClass
     public static void setUp() throws Exception {
         System.out.println("SetUp: ");
         TestHelper.deleteExistingInstanceFiles();
-        nc = new TestNodeController(null, false);
+        String configPath = System.getProperty("user.dir") + File.separator + "src" + File.separator + "test"
+                + File.separator + "resources" + File.separator + "multi-partition-test-configuration.xml";
+        nc = new TestNodeController(configPath, false);
         nc.init();
         ncAppCtx = nc.getAppRuntimeContext();
         dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
@@ -113,84 +147,515 @@
     public void createIndex() throws Exception {
         List<List<String>> partitioningKeys = new ArrayList<>();
         partitioningKeys.add(Collections.singletonList("key"));
-        dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
-                NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
-                        partitioningKeys, null, null, null, false, null),
-                null, DatasetType.INTERNAL, DATASET_ID, 0);
+        dataset =
+                new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+                        NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+                                partitioningKeys, null, null, null, false, null),
+                        null, DatasetType.INTERNAL, DATASET_ID, 0);
+        secondaryIndex = new Index(DATAVERSE_NAME, DATASET_NAME, INDEX_NAME, INDEX_TYPE, INDEX_FIELD_NAMES,
+                INDEX_FIELD_INDICATORS, INDEX_FIELD_TYPES, false, false, false, 0);
         taskCtxs = new IHyracksTaskContext[NUM_PARTITIONS];
-        indexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS];
-        primarylsmBtrees = new TestLsmBtree[NUM_PARTITIONS];
+        primaryIndexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS];
+        secondaryIndexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS];
+        primaryLsmBtrees = new TestLsmBtree[NUM_PARTITIONS];
+        secondaryLsmBtrees = new TestLsmBtree[NUM_PARTITIONS];
         insertOps = new LSMInsertDeleteOperatorNodePushable[NUM_PARTITIONS];
         JobId jobId = nc.newJobId();
         txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(jobId),
                 new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
+        actors = new Actor[NUM_PARTITIONS];
         for (int i = 0; i < taskCtxs.length; i++) {
             taskCtxs[i] = nc.createTestContext(jobId, i, false);
             PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
                     storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, i);
+            SecondaryIndexInfo secondaryIndexInfo =
+                    nc.createSecondaryIndex(primaryIndexInfo, secondaryIndex, storageManager, i);
             IndexDataflowHelperFactory iHelperFactory =
                     new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
-            indexDataflowHelpers[i] = iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i);
-            indexDataflowHelpers[i].open();
-            primarylsmBtrees[i] = (TestLsmBtree) indexDataflowHelpers[i].getIndexInstance();
-            indexDataflowHelpers[i].close();
+            primaryIndexDataflowHelpers[i] =
+                    iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i);
+            primaryIndexDataflowHelpers[i].open();
+            primaryLsmBtrees[i] = (TestLsmBtree) primaryIndexDataflowHelpers[i].getIndexInstance();
+            iHelperFactory =
+                    new IndexDataflowHelperFactory(nc.getStorageManager(), secondaryIndexInfo.getFileSplitProvider());
+            secondaryIndexDataflowHelpers[i] =
+                    iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i);
+            secondaryIndexDataflowHelpers[i].open();
+            secondaryLsmBtrees[i] = (TestLsmBtree) secondaryIndexDataflowHelpers[i].getIndexInstance();
+            secondaryIndexDataflowHelpers[i].close();
+            primaryIndexDataflowHelpers[i].close();
             insertOps[i] = nc.getInsertPipeline(taskCtxs[i], dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
-                    KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft();
+                    KEY_INDEXES, KEY_INDICATORS_LIST, storageManager, secondaryIndex).getLeft();
+            actors[i] = new Actor("player-" + i, i);
+        }
+        // allow all operations
+        for (int i = 0; i < NUM_PARTITIONS; i++) {
+            ComponentRollbackTest.allowAllOps(primaryLsmBtrees[i]);
+            ComponentRollbackTest.allowAllOps(secondaryLsmBtrees[i]);
+            actors[i].add(new Request(Request.Action.INSERT_OPEN));
         }
     }
 
     @After
     public void destroyIndex() throws Exception {
-        for (IIndexDataflowHelper indexDataflowHelper : indexDataflowHelpers) {
+        for (int i = 0; i < NUM_PARTITIONS; i++) {
+            Request close = new Request(Request.Action.INSERT_CLOSE);
+            actors[i].add(close);
+            close.await();
+        }
+        nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+        for (IIndexDataflowHelper indexDataflowHelper : secondaryIndexDataflowHelpers) {
             indexDataflowHelper.destroy();
         }
+        for (IIndexDataflowHelper indexDataflowHelper : primaryIndexDataflowHelpers) {
+            indexDataflowHelper.destroy();
+        }
+        for (Actor actor : actors) {
+            actor.stop();
+        }
     }
 
     @Test
     public void testFlushOneFullOneEmpty() {
         try {
-            // allow all operations
-            for (int i = 0; i < NUM_PARTITIONS; i++) {
-                ComponentRollbackTest.allowAllOps(primarylsmBtrees[i]);
+            int totalNumOfComponents = TOTAL_NUM_OF_RECORDS / RECORDS_PER_COMPONENT;
+            for (int j = 0; j < totalNumOfComponents; j++) {
+                actors[0].add(new Request(Request.Action.INSERT_PATCH));
+                actors[0].add(new Request(Request.Action.FLUSH_DATASET));
             }
-
-            insertOps[0].open();
-            TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
-                    RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
-            VSizeFrame frame = new VSizeFrame(taskCtxs[0]);
-            FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
-            int numFlushes = 0;
-            for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
-                // flush every RECORDS_PER_COMPONENT records
-                if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
-                    if (tupleAppender.getTupleCount() > 0) {
-                        tupleAppender.write(insertOps[0], true);
-                    }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
-                    numFlushes++;
-                }
-                ITupleReference tuple = tupleGenerator.next();
-                DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOps[0]);
-            }
-            if (tupleAppender.getTupleCount() > 0) {
-                tupleAppender.write(insertOps[0], true);
-            }
-            insertOps[0].close();
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
-            numFlushes++;
-            nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+            ensureDone(actors[0]);
             // search now and ensure partition 0 has all the records
             ComponentRollbackTest.searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
             // and that partition 1 has no records
             ComponentRollbackTest.searchAndAssertCount(nc, 1, dataset, storageManager, 0);
             // and that partition 0 has numFlushes disk components
-            Assert.assertEquals(numFlushes, primarylsmBtrees[0].getDiskComponents().size());
+            Assert.assertEquals(totalNumOfComponents, primaryLsmBtrees[0].getDiskComponents().size());
             // and that partition 1 has no disk components
-            Assert.assertEquals(0, primarylsmBtrees[1].getDiskComponents().size());
+            Assert.assertEquals(0, primaryLsmBtrees[1].getDiskComponents().size());
+            // and that in partition 0, all secondary components has a corresponding primary
+            List<ILSMDiskComponent> secondaryDiskComponents = secondaryLsmBtrees[0].getDiskComponents();
+            List<ILSMDiskComponent> primaryDiskComponents = primaryLsmBtrees[0].getDiskComponents();
+            for (int i = 0; i < secondaryDiskComponents.size(); i++) {
+                Assert.assertEquals(secondaryDiskComponents.get(i).getId(), primaryDiskComponents.get(i).getId());
+            }
         } catch (Throwable e) {
             e.printStackTrace();
             Assert.fail(e.getMessage());
         }
     }
 
+    private void ensureDone(Actor actor) throws InterruptedException {
+        Request req = new Request(Request.Action.DUMMY);
+        actor.add(req);
+        req.await();
+    }
+
+    /**
+     * This test update partition 0, schedule flush and modify partition 1
+     * Then ensure that in partition 1, primary and secondary have different component ids
+     */
+    @Test
+    public void testAllocateWhileFlushIsScheduled() {
+        try {
+            // when the vbc becomes full, we want to know
+            AtomicBoolean isFull = new AtomicBoolean(false);
+            MutableBoolean proceedToScheduleFlush = new MutableBoolean(false);
+            primaryLsmBtrees[0].addVirtuablBufferCacheCallback(new IVirtualBufferCacheCallback() {
+                @Override
+                public void isFullChanged(boolean newValue) {
+                    synchronized (isFull) {
+                        isFull.set(newValue);
+                        isFull.notifyAll();
+                    }
+                    synchronized (proceedToScheduleFlush) {
+                        while (!proceedToScheduleFlush.booleanValue()) {
+                            try {
+                                proceedToScheduleFlush.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+            Request insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[0].add(insertReq);
+            while (true) {
+                Thread.sleep(100);
+                if (insertReq.done) {
+                    // if done, then flush was not triggered, then we need to insert a new patch
+                    insertReq = new Request(Request.Action.INSERT_PATCH);
+                    actors[0].add(insertReq);
+                } else if (isFull.get()) {
+                    break;
+                }
+            }
+
+            // now, we need to do two things
+            // allocate primary in partition 1 but not proceed
+            MutableBoolean proceedToAllocateSecondaryIndex = new MutableBoolean(false);
+            MutableBoolean allocated = new MutableBoolean(false);
+            primaryLsmBtrees[1].addAllocateCallback(new ITestOpCallback<Void>() {
+                @Override
+                public void before(Void t) {
+                    // Nothing
+                }
+
+                @Override
+                public void after() {
+                    synchronized (allocated) {
+                        allocated.setValue(true);
+                        allocated.notifyAll();
+                    }
+                    synchronized (proceedToAllocateSecondaryIndex) {
+                        while (!proceedToAllocateSecondaryIndex.booleanValue()) {
+                            try {
+                                proceedToAllocateSecondaryIndex.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+            insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[1].add(insertReq);
+            // Wait for the allocation to happen
+            synchronized (allocated) {
+                while (!allocated.booleanValue()) {
+                    allocated.wait();
+                }
+            }
+            // The memory component has been allocated. now we allow the first actor to proceed to schedule flush
+            MutableBoolean flushStarted = new MutableBoolean(false);
+            primaryLsmBtrees[0].addFlushCallback(new ITestOpCallback<Semaphore>() {
+                @Override
+                public void before(Semaphore t) {
+                    synchronized (flushStarted) {
+                        flushStarted.setValue(true);
+                        flushStarted.notifyAll();
+                    }
+                }
+
+                @Override
+                public void after() {
+                }
+            });
+            synchronized (proceedToScheduleFlush) {
+                proceedToScheduleFlush.setValue(true);
+                proceedToScheduleFlush.notifyAll();
+            }
+            // Now we need to know that the flush has been scheduled
+            synchronized (flushStarted) {
+                while (!flushStarted.booleanValue()) {
+                    flushStarted.wait();
+                }
+            }
+
+            // we now allow the allocation to proceed
+            synchronized (proceedToAllocateSecondaryIndex) {
+                proceedToAllocateSecondaryIndex.setValue(true);
+                proceedToAllocateSecondaryIndex.notifyAll();
+            }
+            // ensure the insert patch has completed
+            insertReq.await();
+            primaryLsmBtrees[1].clearAllocateCallbacks();
+            // check the Ids of the memory components of partition 1
+            // This shows the bug
+            Assert.assertEquals(primaryLsmBtrees[1].getCurrentMemoryComponent().getId(),
+                    secondaryLsmBtrees[1].getCurrentMemoryComponent().getId());
+        } catch (Throwable e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    @Test
+    public void testRecycleWhileFlushIsScheduled() {
+        try {
+            Request insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[0].add(insertReq);
+            Request flushReq = new Request(Request.Action.FLUSH_DATASET);
+            actors[0].add(flushReq);
+            flushReq.await();
+            // ensure that index switched to second component
+            Assert.assertEquals(1, primaryLsmBtrees[0].getCurrentMemoryComponentIndex());
+            insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[0].add(insertReq);
+            flushReq = new Request(Request.Action.FLUSH_DATASET);
+            actors[0].add(flushReq);
+            flushReq.await();
+            // ensure we switched back to first component
+            Assert.assertEquals(0, primaryLsmBtrees[0].getCurrentMemoryComponentIndex());
+            // flush first component of partition 1
+            insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[1].add(insertReq);
+            flushReq = new Request(Request.Action.FLUSH_DATASET);
+            actors[1].add(flushReq);
+            flushReq.await();
+            // ensure partition 1 is now on second component
+            Assert.assertEquals(1, primaryLsmBtrees[1].getCurrentMemoryComponentIndex());
+            // now we want to control when schedule flush is executed
+            AtomicBoolean arrivedAtSchduleFlush = new AtomicBoolean(false);
+            AtomicBoolean finishedSchduleFlush = new AtomicBoolean(false);
+            MutableBoolean proceedToScheduleFlush = new MutableBoolean(false);
+            addOpTrackerCallback(primaryLsmBtrees[0], new ITestOpCallback<Void>() {
+                @Override
+                public void before(Void t) {
+                    synchronized (arrivedAtSchduleFlush) {
+                        arrivedAtSchduleFlush.set(true);
+                        arrivedAtSchduleFlush.notifyAll();
+                    }
+                    synchronized (proceedToScheduleFlush) {
+                        while (!proceedToScheduleFlush.booleanValue()) {
+                            try {
+                                proceedToScheduleFlush.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+
+                @Override
+                public void after() {
+                    synchronized (finishedSchduleFlush) {
+                        finishedSchduleFlush.set(true);
+                        finishedSchduleFlush.notifyAll();
+                    }
+                }
+            });
+            AtomicBoolean isFull = new AtomicBoolean(false);
+            MutableBoolean proceedAfterIsFullChanged = new MutableBoolean(false);
+            primaryLsmBtrees[1].addVirtuablBufferCacheCallback(new IVirtualBufferCacheCallback() {
+                @Override
+                public void isFullChanged(boolean newValue) {
+                    synchronized (isFull) {
+                        isFull.set(newValue);
+                        isFull.notifyAll();
+                    }
+                    synchronized (proceedAfterIsFullChanged) {
+                        while (!proceedAfterIsFullChanged.booleanValue()) {
+                            try {
+                                proceedAfterIsFullChanged.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+
+            // now we start adding records to partition 1 until flush is triggerred
+            insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[1].add(insertReq);
+            while (true) {
+                Thread.sleep(100);
+                if (insertReq.done) {
+                    // if done, then flush was not triggered, then we need to insert a new patch
+                    insertReq = new Request(Request.Action.INSERT_PATCH);
+                    actors[1].add(insertReq);
+                } else if (isFull.get()) {
+                    break;
+                }
+            }
+            // Now we know that vbc is full and flush will be scheduled, we allow this to proceed
+            synchronized (proceedAfterIsFullChanged) {
+                proceedAfterIsFullChanged.setValue(true);
+                proceedAfterIsFullChanged.notifyAll();
+            }
+
+            // now we want to control the recycling of components in partition 0
+            MutableBoolean recycledPrimary = new MutableBoolean(false);
+            MutableBoolean proceedAfterRecyclePrimary = new MutableBoolean(false);
+            ITestOpCallback<ILSMMemoryComponent> primaryRecycleCallback = new ITestOpCallback<ILSMMemoryComponent>() {
+                @Override
+                public void before(ILSMMemoryComponent t) {
+                }
+
+                @Override
+                public void after() {
+                    synchronized (recycledPrimary) {
+                        recycledPrimary.setValue(true);
+                        recycledPrimary.notifyAll();
+                    }
+                    synchronized (proceedAfterRecyclePrimary) {
+                        while (!proceedAfterRecyclePrimary.booleanValue()) {
+                            try {
+                                proceedAfterRecyclePrimary.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+            };
+            primaryLsmBtrees[0].addIoRecycleCallback(primaryRecycleCallback);
+
+            MutableBoolean arrivedToRecycleSecondary = new MutableBoolean(false);
+            MutableBoolean proceedToRecycleSecondary = new MutableBoolean(false);
+            ITestOpCallback<ILSMMemoryComponent> secondaryRecycleCallback = new ITestOpCallback<ILSMMemoryComponent>() {
+                @Override
+                public void before(ILSMMemoryComponent t) {
+                    synchronized (arrivedToRecycleSecondary) {
+                        arrivedToRecycleSecondary.setValue(true);
+                        arrivedToRecycleSecondary.notifyAll();
+                    }
+                    synchronized (proceedToRecycleSecondary) {
+                        while (!proceedToRecycleSecondary.booleanValue()) {
+                            try {
+                                proceedToRecycleSecondary.wait();
+                            } catch (InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                return;
+                            }
+                        }
+                    }
+                }
+
+                @Override
+                public void after() {
+                }
+            };
+            secondaryLsmBtrees[0].addIoRecycleCallback(secondaryRecycleCallback);
+            // we first ensure that schedule flush arrived
+            synchronized (arrivedAtSchduleFlush) {
+                while (!arrivedAtSchduleFlush.get()) {
+                    arrivedAtSchduleFlush.wait();
+                }
+            }
+
+            // we insert a single frame into partition 0
+            insertReq = new Request(Request.Action.INSERT_PATCH);
+            actors[0].add(insertReq);
+            // wait until component has been recycled
+            synchronized (recycledPrimary) {
+                while (!recycledPrimary.booleanValue()) {
+                    recycledPrimary.wait();
+                }
+            }
+            synchronized (proceedAfterRecyclePrimary) {
+                proceedAfterRecyclePrimary.setValue(true);
+                proceedAfterRecyclePrimary.notifyAll();
+            }
+            // now, we know that the primary has been recycled. we allow it to proceed
+            // we allow the scheduleFlush to proceed
+            synchronized (proceedToScheduleFlush) {
+                proceedToScheduleFlush.setValue(true);
+                proceedToScheduleFlush.notifyAll();
+            }
+            // wait until scheduleFlushCompletes
+            synchronized (finishedSchduleFlush) {
+                while (!finishedSchduleFlush.get()) {
+                    finishedSchduleFlush.wait();
+                }
+            }
+            // allow recycling of secondary
+            synchronized (proceedToRecycleSecondary) {
+                proceedToRecycleSecondary.setValue(true);
+                proceedToRecycleSecondary.notifyAll();
+            }
+            // ensure that the insert completes
+            insertReq.await();
+            // ensure the two memory components at partition 0 have different component ids
+            Assert.assertEquals(primaryLsmBtrees[0].getCurrentMemoryComponent().getId(),
+                    secondaryLsmBtrees[0].getCurrentMemoryComponent().getId());
+        } catch (Throwable e) {
+            e.printStackTrace();
+            Assert.fail(e.getMessage());
+        }
+    }
+
+    private static class Request {
+        enum Action {
+            DUMMY,
+            INSERT_OPEN,
+            INSERT_PATCH,
+            FLUSH_DATASET,
+            INSERT_CLOSE
+        }
+
+        private final Action action;
+        private volatile boolean done;
+
+        public Request(Action action) {
+            this.action = action;
+            done = false;
+        }
+
+        synchronized void complete() {
+            done = true;
+            notifyAll();
+        }
+
+        synchronized void await() throws InterruptedException {
+            while (!done) {
+                wait();
+            }
+        }
+
+    }
+
+    public class Actor extends SingleThreadEventProcessor<Request> {
+        private final int partition;
+        private final TupleGenerator tupleGenerator;
+        private final VSizeFrame frame;
+        private final FrameTupleAppender tupleAppender;
+
+        public Actor(String name, int partition) throws HyracksDataException {
+            super(name);
+            this.partition = partition;
+            tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+                    RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+            frame = new VSizeFrame(taskCtxs[partition]);
+            tupleAppender = new FrameTupleAppender(frame);
+        }
+
+        @Override
+        protected void handle(Request req) throws Exception {
+            try {
+                switch (req.action) {
+                    case FLUSH_DATASET:
+                        if (tupleAppender.getTupleCount() > 0) {
+                            tupleAppender.write(insertOps[partition], true);
+                        }
+                        dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                        break;
+                    case INSERT_CLOSE:
+                        insertOps[partition].close();
+                        break;
+                    case INSERT_OPEN:
+                        insertOps[partition].open();
+                        break;
+                    case INSERT_PATCH:
+                        for (int j = 0; j < RECORDS_PER_COMPONENT; j++) {
+                            ITupleReference tuple = tupleGenerator.next();
+                            DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOps[partition]);
+                        }
+                        if (tupleAppender.getTupleCount() > 0) {
+                            tupleAppender.write(insertOps[partition], true);
+                        }
+                        ComponentRollbackTest.waitForOperations(primaryLsmBtrees[partition]);
+                        break;
+                    default:
+                        break;
+                }
+            } finally {
+                req.complete();
+            }
+        }
+    }
+
+    private static void addOpTrackerCallback(TestLsmBtree lsmBtree, ITestOpCallback<Void> callback) {
+        if (!lsmBtree.isPrimaryIndex()) {
+            throw new IllegalArgumentException("Can only add callbacks to primary opTracker");
+        }
+        TestPrimaryIndexOperationTracker opTracker = (TestPrimaryIndexOperationTracker) lsmBtree.getOperationTracker();
+        opTracker.addCallback(callback);
+    }
 }
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
index b65ba03..652616a 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
@@ -121,10 +121,11 @@
     public void createIndex() throws Exception {
         List<List<String>> partitioningKeys = new ArrayList<>();
         partitioningKeys.add(Collections.singletonList("key"));
-        dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
-                NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
-                        partitioningKeys, null, null, null, false, null),
-                null, DatasetType.INTERNAL, DATASET_ID, 0);
+        dataset =
+                new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+                        NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+                                partitioningKeys, null, null, null, false, null),
+                        null, DatasetType.INTERNAL, DATASET_ID, 0);
         PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
                 storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
         IndexDataflowHelperFactory iHelperFactory =
@@ -138,7 +139,7 @@
         txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
                 new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
         insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
-                KEY_INDICATORS_LIST, storageManager).getLeft();
+                KEY_INDICATORS_LIST, storageManager, null).getLeft();
     }
 
     @After
@@ -147,7 +148,7 @@
     }
 
     void unblockSearch(TestLsmBtree lsmBtree) {
-        lsmBtree.addSearchCallback(sem -> sem.release());
+        lsmBtree.addSearchCallback(ComponentRollbackTest.ALLOW_CALLBACK);
         lsmBtree.allowSearch(1);
     }
 
@@ -170,7 +171,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    ComponentRollbackTest.flush(dsLifecycleMgr, lsmBtree, dataset, false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -183,7 +184,7 @@
             firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+            ComponentRollbackTest.flush(dsLifecycleMgr, lsmBtree, dataset, false);
             nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // unblock the search
             unblockSearch(lsmBtree);
@@ -216,7 +217,7 @@
                     if (tupleAppender.getTupleCount() > 0) {
                         tupleAppender.write(insertOp, true);
                     }
-                    dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+                    ComponentRollbackTest.flush(dsLifecycleMgr, lsmBtree, dataset, false);
                 }
                 ITupleReference tuple = tupleGenerator.next();
                 DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
@@ -229,7 +230,7 @@
             firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
             // wait till firstSearcher enter the components
             firstSearcher.waitUntilEntered();
-            dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+            ComponentRollbackTest.flush(dsLifecycleMgr, lsmBtree, dataset, false);
             nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
             // merge all components
             ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBTreeResourceFactoryProvider.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBTreeResourceFactoryProvider.java
index 7268296..8d37878 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBTreeResourceFactoryProvider.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBTreeResourceFactoryProvider.java
@@ -33,6 +33,7 @@
 import org.apache.asterix.metadata.utils.IndexUtil;
 import org.apache.asterix.om.types.ARecordType;
 import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexOperationTrackerFactory;
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.common.utils.Pair;
 import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
@@ -68,6 +69,9 @@
         int[] bloomFilterFields = getBloomFilterFields(dataset, index);
         double bloomFilterFalsePositiveRate = mdProvider.getStorageProperties().getBloomFilterFalsePositiveRate();
         ILSMOperationTrackerFactory opTrackerFactory = dataset.getIndexOperationTrackerFactory(index);
+        if (opTrackerFactory instanceof PrimaryIndexOperationTrackerFactory) {
+            opTrackerFactory = new TestPrimaryIndexOperationTrackerFactory(dataset.getDatasetId());
+        }
         ILSMIOOperationCallbackFactory ioOpCallbackFactory = dataset.getIoOperationCallbackFactory(index);
         IStorageManager storageManager = storageComponentProvider.getStorageManager();
         IMetadataPageManagerFactory metadataPageManagerFactory =
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
index ddcb5b5..fea6cd8 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
@@ -18,14 +18,20 @@
  */
 package org.apache.asterix.test.dataflow;
 
+import java.util.List;
+
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent;
 
 public class TestLsmBtreeIoOpCallbackFactory extends LSMBTreeIOOperationCallbackFactory {
@@ -92,12 +98,31 @@
     }
 
     public class TestLsmBtreeIoOpCallback extends LSMBTreeIOOperationCallback {
+        private final TestLsmBtree lsmBtree;
+
         public TestLsmBtreeIoOpCallback(ILSMIndex index, ILSMComponentIdGenerator idGenerator) {
             super(index, idGenerator);
+            lsmBtree = (TestLsmBtree) index;
+        }
+
+        @Override
+        public void beforeOperation(LSMIOOperationType opType) throws HyracksDataException {
+            lsmBtree.beforeIoOperationCalled();
+            super.beforeOperation(opType);
+            lsmBtree.beforeIoOperationReturned();
+        }
+
+        @Override
+        public void afterOperation(LSMIOOperationType opType, List<ILSMComponent> oldComponents,
+                ILSMDiskComponent newComponent) throws HyracksDataException {
+            lsmBtree.afterIoOperationCalled();
+            super.afterOperation(opType, oldComponents, newComponent);
+            lsmBtree.afterIoOperationReturned();
         }
 
         @Override
         public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) {
+            lsmBtree.afterIoFinalizeCalled();
             super.afterFinalize(opType, newComponent);
             synchronized (TestLsmBtreeIoOpCallbackFactory.this) {
                 if (newComponent != null) {
@@ -119,6 +144,21 @@
                 }
                 TestLsmBtreeIoOpCallbackFactory.this.notifyAll();
             }
+            lsmBtree.afterIoFinalizeReturned();
+        }
+
+        @Override
+        public void recycled(ILSMMemoryComponent component, boolean advance) throws HyracksDataException {
+            lsmBtree.recycledCalled(component);
+            super.recycled(component, advance);
+            lsmBtree.recycledReturned(component);
+        }
+
+        @Override
+        public void allocated(ILSMMemoryComponent component) throws HyracksDataException {
+            lsmBtree.allocatedCalled(component);
+            super.allocated(component);
+            lsmBtree.allocatedReturned(component);
         }
 
         private void recordFailure(LSMIOOperationType opType) {
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTracker.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTracker.java
new file mode 100644
index 0000000..e376ff9
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTracker.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.dataflow;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.asterix.common.context.DatasetInfo;
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.asterix.common.transactions.ILogManager;
+import org.apache.asterix.common.transactions.LogRecord;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
+
+public class TestPrimaryIndexOperationTracker extends PrimaryIndexOperationTracker {
+
+    private final List<ITestOpCallback<Void>> callbacks = new ArrayList<>();
+
+    public TestPrimaryIndexOperationTracker(int datasetID, ILogManager logManager, DatasetInfo dsInfo,
+            ILSMComponentIdGenerator idGenerator) {
+        super(datasetID, logManager, dsInfo, idGenerator);
+    }
+
+    public void addCallback(ITestOpCallback<Void> callback) {
+        synchronized (callbacks) {
+            callbacks.add(callback);
+        }
+    }
+
+    public void clearCallbacks() {
+        synchronized (callbacks) {
+            callbacks.clear();
+        }
+    }
+
+    @Override
+    public void triggerScheduleFlush(LogRecord logRecord) throws HyracksDataException {
+        synchronized (callbacks) {
+            for (ITestOpCallback<Void> callback : callbacks) {
+                callback.before(null);
+            }
+        }
+        super.triggerScheduleFlush(logRecord);
+        synchronized (callbacks) {
+            for (ITestOpCallback<Void> callback : callbacks) {
+                callback.after();
+            }
+        }
+    }
+
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTrackerFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTrackerFactory.java
new file mode 100644
index 0000000..5d7a7c6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestPrimaryIndexOperationTrackerFactory.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.test.dataflow;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.common.context.DatasetLifecycleManager;
+import org.apache.asterix.common.context.DatasetResource;
+import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexOperationTrackerFactory;
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
+
+public class TestPrimaryIndexOperationTrackerFactory extends PrimaryIndexOperationTrackerFactory {
+
+    private static final long serialVersionUID = 1L;
+    private int datasetId;
+
+    public TestPrimaryIndexOperationTrackerFactory(int datasetId) {
+        super(datasetId);
+        this.datasetId = datasetId;
+    }
+
+    @Override
+    public ILSMOperationTracker getOperationTracker(INCServiceContext ctx) {
+        try {
+            INcApplicationContext appCtx = (INcApplicationContext) ctx.getApplicationContext();
+            DatasetLifecycleManager dslcManager = (DatasetLifecycleManager) appCtx.getDatasetLifecycleManager();
+            DatasetResource dsr = dslcManager.getDatasetLifecycle(datasetId);
+            PrimaryIndexOperationTracker opTracker = dsr.getOpTracker();
+            if (!(opTracker instanceof TestPrimaryIndexOperationTracker)) {
+                Field opTrackerField = DatasetResource.class.getDeclaredField("datasetPrimaryOpTracker");
+                opTracker = new TestPrimaryIndexOperationTracker(datasetId,
+                        appCtx.getTransactionSubsystem().getLogManager(), dsr.getDatasetInfo(), dsr.getIdGenerator());
+                setFinal(opTrackerField, dsr, opTracker);
+            }
+            return opTracker;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    static void setFinal(Field field, Object obj, Object newValue) throws Exception {
+        field.setAccessible(true);
+        Field modifiersField = Field.class.getDeclaredField("modifiers");
+        modifiersField.setAccessible(true);
+        modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);
+        field.set(obj, newValue);
+    }
+}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
index 30cfb4f..7543bf6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
@@ -134,7 +134,7 @@
                         new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
                 // Prepare insert operation
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
-                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
+                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager, null).getLeft();
                 insertOp.open();
                 TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATOR,
                         RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
index dabf9d3..8897169 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
@@ -130,7 +130,7 @@
                 IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false);
                 // Prepare insert operation
                 LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
-                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
+                        RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager, null).getLeft();
                 insertOp.open();
                 TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATOR,
                         RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
diff --git a/asterixdb/asterix-app/src/test/resources/multi-partition-test-configuration.xml b/asterixdb/asterix-app/src/test/resources/multi-partition-test-configuration.xml
new file mode 100644
index 0000000..aaeb244
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/multi-partition-test-configuration.xml
@@ -0,0 +1,112 @@
+<!--
+ ! Licensed to the Apache Software Foundation (ASF) under one
+ ! or more contributor license agreements.  See the NOTICE file
+ ! distributed with this work for additional information
+ ! regarding copyright ownership.  The ASF licenses this file
+ ! to you under the Apache License, Version 2.0 (the
+ ! "License"); you may not use this file except in compliance
+ ! with the License.  You may obtain a copy of the License at
+ !
+ !   http://www.apache.org/licenses/LICENSE-2.0
+ !
+ ! Unless required by applicable law or agreed to in writing,
+ ! software distributed under the License is distributed on an
+ ! "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ! KIND, either express or implied.  See the License for the
+ ! specific language governing permissions and limitations
+ ! under the License.
+ !-->
+<asterixConfiguration xmlns="asterixconf">
+  <metadataNode>asterix_nc1</metadataNode>
+  <store>
+    <ncId>asterix_nc1</ncId>
+    <storeDirs>iodevice0,iodevice1</storeDirs>
+  </store>
+  <store>
+    <ncId>asterix_nc2</ncId>
+    <storeDirs>iodevice0,iodevice1</storeDirs>
+  </store>
+  <transactionLogDir>
+    <ncId>asterix_nc1</ncId>
+    <txnLogDirPath>target/txnLogDir/asterix_nc1</txnLogDirPath>
+  </transactionLogDir>
+  <transactionLogDir>
+    <ncId>asterix_nc2</ncId>
+    <txnLogDirPath>target/txnLogDir/asterix_nc2</txnLogDirPath>
+  </transactionLogDir>
+
+  <property>
+    <name>max.wait.active.cluster</name>
+    <value>60</value>
+    <description>Maximum wait (in seconds) for a cluster to be ACTIVE (all
+      nodes are available)
+      before a submitted query/statement can be
+      executed. (Default = 60 seconds)
+    </description>
+  </property>
+
+  <property>
+    <name>compiler.framesize</name>
+    <value>32KB</value>
+  </property>
+  <property>
+    <name>compiler.sortmemory</name>
+    <value>320KB</value>
+  </property>
+  <property>
+    <name>compiler.groupmemory</name>
+    <value>160KB</value>
+  </property>
+  <property>
+    <name>compiler.joinmemory</name>
+    <value>256KB</value>
+  </property>
+  <property>
+    <name>storage.buffercache.pagesize</name>
+    <value>32KB</value>
+    <description>The page size in bytes for pages in the buffer cache.
+      (Default = "128KB")
+    </description>
+  </property>
+  <property>
+    <name>storage.buffercache.size</name>
+    <value>48MB</value>
+    <description>The size of memory allocated to the disk buffer cache.
+      The value should be a multiple of the buffer cache page size.
+    </description>
+  </property>
+  <property>
+    <name>storage.memorycomponent.numpages</name>
+    <value>32</value>
+    <description>The number of pages to allocate for a memory component.
+      This budget is shared by all the memory components of the primary
+      index and all its secondary indexes across all I/O devices on a node.
+      Note: in-memory components usually has fill factor of 75% since
+      the pages are 75% full and the remaining 25% is un-utilized.
+    </description>
+  </property>
+  <property>
+    <name>storage.memorycomponent.globalbudget</name>
+    <value>512MB</value>
+    <description>The size of memory allocated to the memory components.
+      The value should be a multiple of the memory component page size.
+    </description>
+  </property>
+  <property>
+    <name>messaging.frame.size</name>
+    <value>4096</value>
+    <description>The frame size to be used for NC to NC messaging. (Default = 4kb)
+    </description>
+  </property>
+  <property>
+    <name>messaging.frame.count</name>
+    <value>512</value>
+    <description>Number of reusable frames for NC to NC messaging. (Default = 512)
+    </description>
+  </property>
+  <property>
+    <name>log.level</name>
+    <value>INFO</value>
+    <description>foo</description>
+  </property>
+</asterixConfiguration>
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
index 8002895..ce43bca 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
@@ -207,7 +207,7 @@
             if (opTracker != null && opTracker.getNumActiveOperations() == 0
                     && dsr.getDatasetInfo().getReferenceCount() == 0 && dsr.getDatasetInfo().isOpen()
                     && !dsr.isMetadataDataset()) {
-                closeDataset(dsr.getDatasetInfo());
+                closeDataset(dsr);
                 LOGGER.info(() -> "Evicted Dataset" + dsr.getDatasetID());
                 return true;
             }
@@ -341,7 +341,7 @@
     @Override
     public synchronized void flushAllDatasets() throws HyracksDataException {
         for (DatasetResource dsr : datasets.values()) {
-            flushDatasetOpenIndexes(dsr.getDatasetInfo(), false);
+            flushDatasetOpenIndexes(dsr, false);
         }
     }
 
@@ -349,7 +349,7 @@
     public synchronized void flushDataset(int datasetId, boolean asyncFlush) throws HyracksDataException {
         DatasetResource dsr = datasets.get(datasetId);
         if (dsr != null) {
-            flushDatasetOpenIndexes(dsr.getDatasetInfo(), asyncFlush);
+            flushDatasetOpenIndexes(dsr, asyncFlush);
         }
     }
 
@@ -385,11 +385,17 @@
     /*
      * This method can only be called asynchronously safely if we're sure no modify operation will take place until the flush is scheduled
      */
-    private void flushDatasetOpenIndexes(DatasetInfo dsInfo, boolean asyncFlush) throws HyracksDataException {
+    private void flushDatasetOpenIndexes(DatasetResource dsr, boolean asyncFlush) throws HyracksDataException {
+        DatasetInfo dsInfo = dsr.getDatasetInfo();
         if (dsInfo.isExternal()) {
             // no memory components for external dataset
             return;
         }
+        PrimaryIndexOperationTracker primaryOpTracker = dsr.getOpTracker();
+        if (primaryOpTracker.getNumActiveOperations() > 0) {
+            throw new IllegalStateException(
+                    "flushDatasetOpenIndexes is called on a dataset with currently active operations");
+        }
 
         ILSMComponentIdGenerator idGenerator = getComponentIdGenerator(dsInfo.getDatasetID());
         idGenerator.refresh();
@@ -435,11 +441,12 @@
         }
     }
 
-    private void closeDataset(DatasetInfo dsInfo) throws HyracksDataException {
+    private void closeDataset(DatasetResource dsr) throws HyracksDataException {
         // First wait for any ongoing IO operations
+        DatasetInfo dsInfo = dsr.getDatasetInfo();
         dsInfo.waitForIO();
         try {
-            flushDatasetOpenIndexes(dsInfo, false);
+            flushDatasetOpenIndexes(dsr, false);
         } catch (Exception e) {
             throw HyracksDataException.create(e);
         }
@@ -460,7 +467,7 @@
     public synchronized void closeAllDatasets() throws HyracksDataException {
         ArrayList<DatasetResource> openDatasets = new ArrayList<>(datasets.values());
         for (DatasetResource dsr : openDatasets) {
-            closeDataset(dsr.getDatasetInfo());
+            closeDataset(dsr);
         }
     }
 
@@ -469,7 +476,7 @@
         ArrayList<DatasetResource> openDatasets = new ArrayList<>(datasets.values());
         for (DatasetResource dsr : openDatasets) {
             if (!dsr.isMetadataDataset()) {
-                closeDataset(dsr.getDatasetInfo());
+                closeDataset(dsr);
             }
         }
     }
@@ -568,7 +575,7 @@
     public void flushDataset(IReplicationStrategy replicationStrategy) throws HyracksDataException {
         for (DatasetResource dsr : datasets.values()) {
             if (replicationStrategy.isMatch(dsr.getDatasetID())) {
-                flushDatasetOpenIndexes(dsr.getDatasetInfo(), false);
+                flushDatasetOpenIndexes(dsr, false);
             }
         }
     }
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
index 7bb12c4..8df872b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperationCallback.java
@@ -61,6 +61,7 @@
      *
      * @param component
      * @param componentSwitched
+     *            true if the component index was advanced for this recycle, false otherwise
      */
     void recycled(ILSMMemoryComponent component, boolean componentSwitched) throws HyracksDataException;
 
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
index 2989af9..acc3347 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
@@ -18,8 +18,8 @@
  */
 package org.apache.hyracks.storage.am.lsm.btree.impl;
 
-import java.util.concurrent.Semaphore;
+public interface ITestOpCallback<T> {
+    void before(T t);
 
-public interface ITestOpCallback {
-    void callback(Semaphore smeaphore);
+    void after();
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/IVirtualBufferCacheCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/IVirtualBufferCacheCallback.java
new file mode 100644
index 0000000..d5f3fb2
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/IVirtualBufferCacheCallback.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.lsm.btree.impl;
+
+public interface IVirtualBufferCacheCallback {
+    void isFullChanged(boolean newValue);
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
index 8782565..bf3bb31 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
@@ -40,6 +40,7 @@
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
@@ -56,11 +57,17 @@
     private final Semaphore searchSemaphore = new Semaphore(0);
     private final Semaphore flushSemaphore = new Semaphore(0);
     private final Semaphore mergeSemaphore = new Semaphore(0);
-    private final List<ITestOpCallback> modifyCallbacks = new ArrayList<>();
-    private final List<ITestOpCallback> searchCallbacks = new ArrayList<>();
-    private final List<ITestOpCallback> flushCallbacks = new ArrayList<>();
-    private final List<ITestOpCallback> mergeCallbacks = new ArrayList<>();
-    private final List<ITestOpCallback> allocateComponentCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Semaphore>> modifyCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Semaphore>> searchCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Semaphore>> flushCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Semaphore>> mergeCallbacks = new ArrayList<>();
+
+    private final List<ITestOpCallback<ILSMMemoryComponent>> ioAllocateCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<ILSMMemoryComponent>> ioRecycleCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Void>> ioBeforeCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Void>> ioAfterOpCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Void>> ioAfterFinalizeCallbacks = new ArrayList<>();
+    private final List<ITestOpCallback<Void>> allocateComponentCallbacks = new ArrayList<>();
 
     private volatile int numScheduledFlushes;
     private volatile int numStartedFlushes;
@@ -89,17 +96,22 @@
     @Override
     public void modify(IIndexOperationContext ictx, ITupleReference tuple) throws HyracksDataException {
         synchronized (modifyCallbacks) {
-            for (ITestOpCallback callback : modifyCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : modifyCallbacks) {
                 callback(callback, modifySemaphore);
             }
         }
         acquire(modifySemaphore);
         super.modify(ictx, tuple);
+        synchronized (modifyCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : modifyCallbacks) {
+                callback.after();
+            }
+        }
     }
 
-    public static void callback(ITestOpCallback callback, Semaphore semaphore) {
+    public static <T> void callback(ITestOpCallback<T> callback, T t) {
         if (callback != null) {
-            callback.callback(semaphore);
+            callback.before(t);
         }
     }
 
@@ -121,13 +133,18 @@
     public ILSMDiskComponent doFlush(ILSMIOOperation operation) throws HyracksDataException {
         numStartedFlushes++;
         synchronized (flushCallbacks) {
-            for (ITestOpCallback callback : flushCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : flushCallbacks) {
                 callback(callback, flushSemaphore);
             }
         }
         acquire(flushSemaphore);
         ILSMDiskComponent c = super.doFlush(operation);
         numFinishedFlushes++;
+        synchronized (flushCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : flushCallbacks) {
+                callback.after();
+            }
+        }
         return c;
     }
 
@@ -135,13 +152,18 @@
     public ILSMDiskComponent doMerge(ILSMIOOperation operation) throws HyracksDataException {
         numStartedMerges++;
         synchronized (mergeCallbacks) {
-            for (ITestOpCallback callback : mergeCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : mergeCallbacks) {
                 callback(callback, mergeSemaphore);
             }
         }
         acquire(mergeSemaphore);
         ILSMDiskComponent c = super.doMerge(operation);
         numFinishedMerges++;
+        synchronized (mergeCallbacks) {
+            for (ITestOpCallback<Semaphore> callback : mergeCallbacks) {
+                callback.after();
+            }
+        }
         return c;
     }
 
@@ -199,11 +221,11 @@
         return numFinishedMerges;
     }
 
-    public List<ITestOpCallback> getModifyCallbacks() {
+    public List<ITestOpCallback<Semaphore>> getModifyCallbacks() {
         return modifyCallbacks;
     }
 
-    public void addModifyCallback(ITestOpCallback modifyCallback) {
+    public void addModifyCallback(ITestOpCallback<Semaphore> modifyCallback) {
         synchronized (mergeCallbacks) {
             modifyCallbacks.add(modifyCallback);
         }
@@ -215,7 +237,31 @@
         }
     }
 
-    public List<ITestOpCallback> getSearchCallbacks() {
+    public void addIoRecycleCallback(ITestOpCallback<ILSMMemoryComponent> callback) {
+        synchronized (ioRecycleCallbacks) {
+            ioRecycleCallbacks.add(callback);
+        }
+    }
+
+    public void clearIoRecycleCallback() {
+        synchronized (ioRecycleCallbacks) {
+            ioRecycleCallbacks.clear();
+        }
+    }
+
+    public void addIoAllocateCallback(ITestOpCallback<ILSMMemoryComponent> callback) {
+        synchronized (ioAllocateCallbacks) {
+            ioAllocateCallbacks.add(callback);
+        }
+    }
+
+    public void clearIoAllocateCallback() {
+        synchronized (ioAllocateCallbacks) {
+            ioAllocateCallbacks.clear();
+        }
+    }
+
+    public List<ITestOpCallback<Semaphore>> getSearchCallbacks() {
         return searchCallbacks;
     }
 
@@ -225,13 +271,13 @@
         }
     }
 
-    public void addSearchCallback(ITestOpCallback searchCallback) {
+    public void addSearchCallback(ITestOpCallback<Semaphore> searchCallback) {
         synchronized (searchCallbacks) {
             searchCallbacks.add(searchCallback);
         }
     }
 
-    public void addFlushCallback(ITestOpCallback flushCallback) {
+    public void addFlushCallback(ITestOpCallback<Semaphore> flushCallback) {
         synchronized (flushCallbacks) {
             flushCallbacks.add(flushCallback);
         }
@@ -243,7 +289,7 @@
         }
     }
 
-    public void addMergeCallback(ITestOpCallback mergeCallback) {
+    public void addMergeCallback(ITestOpCallback<Semaphore> mergeCallback) {
         synchronized (mergeCallbacks) {
             mergeCallbacks.add(mergeCallback);
         }
@@ -259,7 +305,7 @@
         return searchSemaphore;
     }
 
-    public void addAllocateCallback(ITestOpCallback callback) {
+    public void addAllocateCallback(ITestOpCallback<Void> callback) {
         synchronized (allocateComponentCallbacks) {
             allocateComponentCallbacks.add(callback);
         }
@@ -271,13 +317,111 @@
         }
     }
 
+    public void addVirtuablBufferCacheCallback(IVirtualBufferCacheCallback callback) {
+        for (IVirtualBufferCache vbc : virtualBufferCaches) {
+            ((TestVirtualBufferCache) vbc).addCallback(callback);
+        }
+    }
+
+    public void clearVirtuablBufferCacheCallbacks() {
+        for (IVirtualBufferCache vbc : virtualBufferCaches) {
+            ((TestVirtualBufferCache) vbc).clearCallbacks();
+        }
+    }
+
     @Override
     public void allocateMemoryComponents() throws HyracksDataException {
-        super.allocateMemoryComponents();
         synchronized (allocateComponentCallbacks) {
-            for (ITestOpCallback callback : allocateComponentCallbacks) {
+            for (ITestOpCallback<Void> callback : allocateComponentCallbacks) {
                 callback(callback, null);
             }
         }
+        super.allocateMemoryComponents();
+        synchronized (allocateComponentCallbacks) {
+            for (ITestOpCallback<Void> callback : allocateComponentCallbacks) {
+                callback.after();
+            }
+        }
     }
+
+    public void beforeIoOperationCalled() {
+        synchronized (ioBeforeCallbacks) {
+            for (ITestOpCallback<Void> callback : ioBeforeCallbacks) {
+                callback.before(null);
+            }
+        }
+    }
+
+    public void beforeIoOperationReturned() {
+        synchronized (ioBeforeCallbacks) {
+            for (ITestOpCallback<Void> callback : ioBeforeCallbacks) {
+                callback.after();
+            }
+        }
+    }
+
+    public void afterIoOperationCalled() {
+        synchronized (ioAfterOpCallbacks) {
+            for (ITestOpCallback<Void> callback : ioAfterOpCallbacks) {
+                callback.before(null);
+            }
+        }
+    }
+
+    public void afterIoOperationReturned() {
+        synchronized (ioAfterOpCallbacks) {
+            for (ITestOpCallback<Void> callback : ioAfterOpCallbacks) {
+                callback.after();
+            }
+        }
+    }
+
+    public void afterIoFinalizeCalled() {
+        synchronized (ioAfterFinalizeCallbacks) {
+            for (ITestOpCallback<Void> callback : ioAfterFinalizeCallbacks) {
+                callback.before(null);
+            }
+        }
+    }
+
+    public void afterIoFinalizeReturned() {
+        synchronized (ioAfterFinalizeCallbacks) {
+            for (ITestOpCallback<Void> callback : ioAfterFinalizeCallbacks) {
+                callback.after();
+            }
+        }
+    }
+
+    public void recycledCalled(ILSMMemoryComponent component) {
+        synchronized (ioRecycleCallbacks) {
+            for (ITestOpCallback<ILSMMemoryComponent> callback : ioRecycleCallbacks) {
+                callback.before(component);
+            }
+        }
+    }
+
+    public void recycledReturned(ILSMMemoryComponent component) {
+        synchronized (ioRecycleCallbacks) {
+            for (ITestOpCallback<ILSMMemoryComponent> callback : ioRecycleCallbacks) {
+                callback.after();
+            }
+        }
+    }
+
+    public void allocatedCalled(ILSMMemoryComponent component) {
+        synchronized (ioAllocateCallbacks) {
+            for (ITestOpCallback<ILSMMemoryComponent> callback : ioAllocateCallbacks) {
+                callback.before(component);
+            }
+        }
+    }
+
+    public void allocatedReturned(ILSMMemoryComponent component) {
+        synchronized (ioAllocateCallbacks) {
+            for (ITestOpCallback<ILSMMemoryComponent> callback : ioAllocateCallbacks) {
+                callback.after();
+            }
+        }
+    }
+
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeLocalResource.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeLocalResource.java
index 47a8046..9b53120 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeLocalResource.java
@@ -60,6 +60,13 @@
         IIOManager ioManager = serviceCtx.getIoManager();
         FileReference file = ioManager.resolve(path);
         List<IVirtualBufferCache> vbcs = vbcProvider.getVirtualBufferCaches(serviceCtx, file);
+        for (int i = 0; i < vbcs.size(); i++) {
+            IVirtualBufferCache vbc = vbcs.get(i);
+            if (!(vbc instanceof TestVirtualBufferCache)) {
+                vbcs.remove(i);
+                vbcs.add(i, new TestVirtualBufferCache(vbc));
+            }
+        }
         ioOpCallbackFactory.initialize(serviceCtx);
         return TestLsmBtreeUtil.createLSMTree(ioManager, vbcs, file, storageManager.getBufferCache(serviceCtx),
                 typeTraits, cmpFactories, bloomFilterKeyFields, bloomFilterFalsePositiveRate,
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeSearchCursor.java
index d3504ae..45e39aad 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtreeSearchCursor.java
@@ -19,6 +19,7 @@
 package org.apache.hyracks.storage.am.lsm.btree.impl;
 
 import java.util.List;
+import java.util.concurrent.Semaphore;
 
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTreeSearchCursor;
@@ -36,9 +37,9 @@
     @Override
     public void next() throws HyracksDataException {
         try {
-            List<ITestOpCallback> callbacks = lsmBtree.getSearchCallbacks();
+            List<ITestOpCallback<Semaphore>> callbacks = lsmBtree.getSearchCallbacks();
             synchronized (callbacks) {
-                for (ITestOpCallback cb : callbacks) {
+                for (ITestOpCallback<Semaphore> cb : callbacks) {
                     TestLsmBtree.callback(cb, lsmBtree.getSearchSemaphore());
                 }
             }
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java
new file mode 100644
index 0000000..c7e064f
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestVirtualBufferCache.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.lsm.btree.impl;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.replication.IIOReplicationManager;
+import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.common.buffercache.ICachedPage;
+import org.apache.hyracks.storage.common.buffercache.IExtraPageBlockHelper;
+import org.apache.hyracks.storage.common.buffercache.IFIFOPageQueue;
+import org.apache.hyracks.storage.common.file.IFileMapManager;
+
+public class TestVirtualBufferCache implements IVirtualBufferCache {
+    private final IVirtualBufferCache vbc;
+    private final AtomicBoolean isFull = new AtomicBoolean(false);
+    private final List<IVirtualBufferCacheCallback> callbacks;
+
+    public TestVirtualBufferCache(IVirtualBufferCache vbc) {
+        this.vbc = vbc;
+        callbacks = new ArrayList<>();
+    }
+
+    public void addCallback(IVirtualBufferCacheCallback callback) {
+        synchronized (callbacks) {
+            callbacks.add(callback);
+        }
+    }
+
+    public void clearCallbacks() {
+        synchronized (callbacks) {
+            callbacks.clear();
+        }
+    }
+
+    @Override
+    public int createFile(FileReference fileRef) throws HyracksDataException {
+        return vbc.createFile(fileRef);
+    }
+
+    @Override
+    public int openFile(FileReference fileRef) throws HyracksDataException {
+        return vbc.openFile(fileRef);
+    }
+
+    @Override
+    public void openFile(int fileId) throws HyracksDataException {
+        vbc.openFile(fileId);
+    }
+
+    @Override
+    public void closeFile(int fileId) throws HyracksDataException {
+        vbc.closeFile(fileId);
+    }
+
+    @Override
+    public void deleteFile(int fileId) throws HyracksDataException {
+        vbc.deleteFile(fileId);
+    }
+
+    @Override
+    public void deleteFile(FileReference file) throws HyracksDataException {
+        vbc.deleteFile(file);
+    }
+
+    @Override
+    public ICachedPage pin(long dpid, boolean newPage) throws HyracksDataException {
+        return vbc.pin(dpid, newPage);
+    }
+
+    @Override
+    public void unpin(ICachedPage page) throws HyracksDataException {
+        vbc.unpin(page);
+    }
+
+    @Override
+    public void flush(ICachedPage page) throws HyracksDataException {
+        vbc.flush(page);
+    }
+
+    @Override
+    public void force(int fileId, boolean metadata) throws HyracksDataException {
+        vbc.force(fileId, metadata);
+    }
+
+    @Override
+    public ICachedPage confiscatePage(long dpid) throws HyracksDataException {
+        return vbc.confiscatePage(dpid);
+    }
+
+    @Override
+    public ICachedPage confiscateLargePage(long dpid, int multiplier, int extraBlockPageId)
+            throws HyracksDataException {
+        return vbc.confiscateLargePage(dpid, multiplier, extraBlockPageId);
+    }
+
+    @Override
+    public void returnPage(ICachedPage page) {
+        vbc.returnPage(page);
+    }
+
+    @Override
+    public void returnPage(ICachedPage page, boolean reinsert) {
+        vbc.returnPage(page, reinsert);
+    }
+
+    @Override
+    public int getPageSize() {
+        return vbc.getPageSize();
+    }
+
+    @Override
+    public int getPageSizeWithHeader() {
+        return vbc.getPageSizeWithHeader();
+    }
+
+    @Override
+    public int getPageBudget() {
+        return vbc.getPageBudget();
+    }
+
+    @Override
+    public int getNumPagesOfFile(int fileId) throws HyracksDataException {
+        return vbc.getNumPagesOfFile(fileId);
+    }
+
+    @Override
+    public int getFileReferenceCount(int fileId) {
+        return vbc.getFileReferenceCount(fileId);
+    }
+
+    @Override
+    public void close() throws HyracksDataException {
+        vbc.close();
+    }
+
+    @Override
+    public IFIFOPageQueue createFIFOQueue() {
+        return vbc.createFIFOQueue();
+    }
+
+    @Override
+    public void finishQueue() throws HyracksDataException {
+        vbc.finishQueue();
+    }
+
+    @Override
+    public boolean isReplicationEnabled() {
+        return vbc.isReplicationEnabled();
+    }
+
+    @Override
+    public IIOReplicationManager getIOReplicationManager() {
+        return vbc.getIOReplicationManager();
+    }
+
+    @Override
+    public void purgeHandle(int fileId) throws HyracksDataException {
+        vbc.purgeHandle(fileId);
+    }
+
+    @Override
+    public void resizePage(ICachedPage page, int multiplier, IExtraPageBlockHelper extraPageBlockHelper)
+            throws HyracksDataException {
+        vbc.resizePage(page, multiplier, extraPageBlockHelper);
+    }
+
+    @Override
+    public void open() throws HyracksDataException {
+        vbc.open();
+    }
+
+    @Override
+    public boolean isFull() {
+        boolean newValue = vbc.isFull();
+        if (isFull.compareAndSet(!newValue, newValue)) {
+            synchronized (callbacks) {
+                for (int i = 0; i < callbacks.size(); i++) {
+                    callbacks.get(i).isFullChanged(newValue);
+                }
+            }
+        }
+        return newValue;
+    }
+
+    @Override
+    public void reset() {
+        vbc.reset();
+    }
+
+    @Override
+    public IFileMapManager getFileMapProvider() {
+        return vbc.getFileMapProvider();
+    }
+}