diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
index 6327e5d..737106f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/DeleteIT.java
@@ -47,12 +47,15 @@
 import org.apache.hadoop.hbase.RawCell;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.Tag;
+import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.RegionScanner;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.compile.DeleteCompiler;
 import org.apache.phoenix.compile.MutationPlan;
+import org.apache.phoenix.end2end.index.IndexTestUtil;
+import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.jdbc.PhoenixStatement;
 import org.apache.phoenix.parse.DeleteStatement;
 import org.apache.phoenix.parse.SQLParser;
@@ -978,7 +981,7 @@
         // Add tag "customer-delete" to delete marker.
         props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
 
-        createAndUpsertTable(tableName, indexName, props);
+        createAndUpsertTable(tableName, indexName, props, false);
         // Make sure that the plan creates is of ClientSelectDeleteMutationPlan
         verifyDeletePlan(delete, DeleteCompiler.ClientSelectDeleteMutationPlan.class, props);
         executeDelete(delete, props, 1);
@@ -1003,7 +1006,7 @@
         Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
         props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
 
-        createAndUpsertTable(tableName, indexName, props);
+        createAndUpsertTable(tableName, indexName, props, false);
         // Make sure that the plan creates is of ServerSelectDeleteMutationPlan
         verifyDeletePlan(delete, DeleteCompiler.ServerSelectDeleteMutationPlan.class, props);
         executeDelete(delete, props, 2);
@@ -1029,7 +1032,7 @@
         props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
         // Don't create index table. We will use MultiRowDeleteMutationPlan
         // if there is no index present for a table.
-        createAndUpsertTable(tableName, null, props);
+        createAndUpsertTable(tableName, null, props, false);
         // Make sure that the plan creates is of MultiRowDeleteMutationPlan
         verifyDeletePlan(delete, DeleteCompiler.MultiRowDeleteMutationPlan.class, props);
         executeDelete(delete, props, 1);
@@ -1054,8 +1057,8 @@
             assertEquals(plan.getClass(), planClass);
         }
     }
-    private void createAndUpsertTable(String tableName, String indexName, Properties props)
-            throws SQLException {
+    private void createAndUpsertTable(String tableName, String indexName, Properties props,
+            boolean useOldCoproc) throws Exception {
         String ddl = "CREATE TABLE " + tableName +
                 " (k INTEGER NOT NULL PRIMARY KEY, v1 VARCHAR, v2 VARCHAR)";
         try(Connection conn = DriverManager.getConnection(getUrl(), props)) {
@@ -1066,6 +1069,11 @@
                     String indexDdl1 = "CREATE INDEX " + indexName + " ON " + tableName + "(v1,v2)";
                     statement.execute(indexDdl1);
                 }
+                if (useOldCoproc) {
+                    Admin admin = ((PhoenixConnection) conn).getQueryServices().getAdmin();
+                    IndexTestUtil.downgradeCoprocs(tableName, indexName, admin);
+                }
+
                 statement.execute(
                         "upsert into " + tableName + " values (1, 'foo', 'foo1')");
                 statement.execute(
@@ -1122,4 +1130,23 @@
             assertEquals(0, tags.size());
         }
     }
+
+    /*
+        Test whether source of operation tags are added to Delete mutations if we are using
+        old index coproc.
+     */
+    @Test
+    public void testDeleteTagsWithOldIndexCoproc() throws Exception {
+        String tableName = generateUniqueName();
+        String tagValue = "customer-delete";
+        String delete = "DELETE FROM " + tableName + " WHERE k = 1";
+        Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+        props.setProperty(ConnectionQueryServices.SOURCE_OPERATION_ATTRIB, tagValue);
+        // The new table will always have new index coproc. Downgrade it to use older one.
+        createAndUpsertTable(tableName, null, props, true);
+        executeDelete(delete, props, 1);
+        String startRowKeyForBaseTable = "1";
+        // Make sure that Delete Marker has cell tag for base table.
+        checkTagPresentInDeleteMarker(tableName, startRowKeyForBaseTable, true, tagValue);
+    }
 }
\ No newline at end of file
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java
index ba8be11..a7c6c6d 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParameterizedIndexUpgradeToolIT.java
@@ -17,13 +17,12 @@
  */
 package org.apache.phoenix.end2end;
 
+import org.apache.phoenix.end2end.index.IndexTestUtil;
 import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.TableDescriptor;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.end2end.index.IndexCoprocIT;
 import org.apache.phoenix.hbase.index.IndexRegionObserver;
 import org.apache.phoenix.hbase.index.Indexer;
@@ -400,7 +399,7 @@
                         indexDesc.hasCoprocessor(IndexRegionObserver.class.getName()));
                     Assert.assertFalse("Found Indexer on " + table,
                         indexDesc.hasCoprocessor(Indexer.class.getName()));
-                    IndexCoprocIT.assertCoprocConfig(indexDesc, IndexRegionObserver.class.getName(),
+                    IndexTestUtil.assertCoprocConfig(indexDesc, IndexRegionObserver.class.getName(),
                         IndexCoprocIT.INDEX_REGION_OBSERVER_CONFIG);
                 }
             }
@@ -410,7 +409,7 @@
                 TableDescriptor indexDesc = admin.getDescriptor(TableName.valueOf(index));
                 Assert.assertTrue("Couldn't find GlobalIndexChecker on " + index,
                     indexDesc.hasCoprocessor(GlobalIndexChecker.class.getName()));
-                IndexCoprocIT.assertCoprocConfig(indexDesc, GlobalIndexChecker.class.getName(),
+                IndexTestUtil.assertCoprocConfig(indexDesc, GlobalIndexChecker.class.getName(),
                     IndexCoprocIT.GLOBAL_INDEX_CHECKER_CONFIG);
             }
         }
@@ -440,7 +439,7 @@
                     indexDesc.hasCoprocessor(Indexer.class.getName()));
                 Assert.assertFalse("Found IndexRegionObserver on " + table,
                     indexDesc.hasCoprocessor(IndexRegionObserver.class.getName()));
-                IndexCoprocIT.assertCoprocConfig(indexDesc, Indexer.class.getName(),
+                IndexTestUtil.assertCoprocConfig(indexDesc, Indexer.class.getName(),
                     IndexCoprocIT.INDEXER_CONFIG);
             }
         }
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexCoprocIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexCoprocIT.java
index c0a11ff..c71129b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexCoprocIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexCoprocIT.java
@@ -23,20 +23,14 @@
 import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
 import org.apache.phoenix.hbase.index.IndexRegionObserver;
 import org.apache.phoenix.hbase.index.Indexer;
-import org.apache.phoenix.hbase.index.covered.NonTxIndexBuilder;
 import org.apache.phoenix.index.GlobalIndexChecker;
-import org.apache.phoenix.index.PhoenixIndexBuilder;
-import org.apache.phoenix.index.PhoenixIndexCodec;
 import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.query.QueryServices;
-import org.apache.phoenix.query.QueryServicesOptions;
 import org.apache.phoenix.util.SchemaUtil;
-import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -46,10 +40,13 @@
 import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Properties;
 
+import static org.apache.phoenix.end2end.index.IndexTestUtil.assertCoprocConfig;
+import static org.apache.phoenix.end2end.index.IndexTestUtil.assertCoprocsContains;
+import static org.apache.phoenix.end2end.index.IndexTestUtil.assertCoprocsNotContains;
+import static org.apache.phoenix.end2end.index.IndexTestUtil.assertUsingOldCoprocs;
+
 @RunWith(Parameterized.class)
 public class IndexCoprocIT extends ParallelStatsDisabledIT {
     private boolean isNamespaceMapped = false;
@@ -82,35 +79,15 @@
             isNamespaceMapped).getString();
         String physicalIndexName = SchemaUtil.getPhysicalHBaseTableName(schemaName,
             indexName, isNamespaceMapped).getString();
-        Admin admin = ((PhoenixConnection) getConnection()).getQueryServices().getAdmin();
+        Admin admin = getConnection().getQueryServices().getAdmin();
 
         createBaseTable(schemaName, tableName, isMultiTenant, 0, null);
         createIndexTable(schemaName, tableName, indexName);
 
-        TableDescriptor baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
-        TableDescriptorBuilder baseDescBuilder = TableDescriptorBuilder.newBuilder(baseDescriptor);
-        TableDescriptor indexDescriptor = admin.getDescriptor(TableName.valueOf(physicalIndexName));
-        TableDescriptorBuilder indexDescBuilder = TableDescriptorBuilder.newBuilder(indexDescriptor);
-
-        assertCoprocsContains(IndexRegionObserver.class, baseDescriptor);
-        assertCoprocsContains(GlobalIndexChecker.class, indexDescriptor);
-
-        removeCoproc(IndexRegionObserver.class, baseDescBuilder, admin);
-        removeCoproc(IndexRegionObserver.class, indexDescBuilder, admin);
-        removeCoproc(GlobalIndexChecker.class, indexDescBuilder, admin);
-
-        Map<String, String> props = new HashMap<String, String>();
-        props.put(NonTxIndexBuilder.CODEC_CLASS_NAME_KEY, PhoenixIndexCodec.class.getName());
-        Indexer.enableIndexing(baseDescBuilder, PhoenixIndexBuilder.class,
-            props, QueryServicesOptions.DEFAULT_COPROCESSOR_PRIORITY);
-        admin.modifyTable(baseDescBuilder.build());
-        baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
-        indexDescriptor = admin.getDescriptor(TableName.valueOf(physicalIndexName));
-        assertUsingOldCoprocs(baseDescriptor, indexDescriptor);
-
+        IndexTestUtil.downgradeCoprocs(physicalTableName, physicalIndexName, admin);
         createBaseTable(schemaName, tableName, true, 0, null);
-        baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
-        indexDescriptor = admin.getDescriptor(TableName.valueOf(physicalIndexName));
+        TableDescriptor baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
+        TableDescriptor indexDescriptor = admin.getDescriptor(TableName.valueOf(physicalIndexName));
         assertUsingOldCoprocs(baseDescriptor, indexDescriptor);
     }
 
@@ -150,7 +127,7 @@
             isNamespaceMapped).getString();
         String physicalIndexName = SchemaUtil.getPhysicalHBaseTableName(schemaName,
             indexName, isNamespaceMapped).getString();
-        Admin admin = ((PhoenixConnection) getConnection()).getQueryServices().getAdmin();
+        Admin admin = getConnection().getQueryServices().getAdmin();
 
         createBaseTable(schemaName, tableName, isMultiTenant, 0, null);
         createIndexTable(schemaName, tableName, indexName);
@@ -167,15 +144,6 @@
         assertCoprocsContains(IndexRegionObserver.class, baseDescriptor);
         assertCoprocsContains(GlobalIndexChecker.class, indexDescriptor);
     }
-    private void assertUsingOldCoprocs(TableDescriptor baseDescriptor,
-                                       TableDescriptor indexDescriptor) {
-        assertCoprocsContains(Indexer.class, baseDescriptor);
-        assertCoprocConfig(baseDescriptor, Indexer.class.getName(),
-            INDEXER_CONFIG);
-        assertCoprocsNotContains(IndexRegionObserver.class, baseDescriptor);
-        assertCoprocsNotContains(IndexRegionObserver.class, indexDescriptor);
-        assertCoprocsNotContains(GlobalIndexChecker.class, indexDescriptor);
-    }
 
     private void assertUsingNewCoprocs(TableDescriptor baseDescriptor) {
         assertCoprocsContains(IndexRegionObserver.class, baseDescriptor);
@@ -194,58 +162,6 @@
             GLOBAL_INDEX_CHECKER_CONFIG);
     }
 
-    private void assertCoprocsContains(Class clazz, TableDescriptor descriptor) {
-        String expectedCoprocName = clazz.getName();
-        boolean foundCoproc = isCoprocPresent(descriptor, expectedCoprocName);
-        Assert.assertTrue("Could not find coproc " + expectedCoprocName +
-            " in descriptor " + descriptor,foundCoproc);
-    }
-
-    private void assertCoprocsNotContains(Class clazz, TableDescriptor descriptor) {
-        String expectedCoprocName = clazz.getName();
-        boolean foundCoproc = isCoprocPresent(descriptor, expectedCoprocName);
-        Assert.assertFalse("Could find coproc " + expectedCoprocName +
-            " in descriptor " + descriptor,foundCoproc);
-    }
-
-    private boolean isCoprocPresent(TableDescriptor descriptor, String expectedCoprocName) {
-        boolean foundCoproc = false;
-        for (String coprocName : descriptor.getCoprocessors()){
-            if (coprocName.equals(expectedCoprocName)){
-                foundCoproc = true;
-                break;
-            }
-        }
-        return foundCoproc;
-    }
-
-    private void removeCoproc(Class clazz, TableDescriptorBuilder descBuilder, Admin admin) throws Exception {
-        descBuilder.removeCoprocessor(clazz.getName());
-        admin.modifyTable(descBuilder.build());
-    }
-
-    public static void assertCoprocConfig(TableDescriptor indexDesc,
-                                   String className, String expectedConfigValue){
-        boolean foundConfig = false;
-        for (Map.Entry<Bytes, Bytes> entry :
-            indexDesc.getValues().entrySet()){
-            String propKey = Bytes.toString(entry.getKey().get());
-            String propValue = Bytes.toString(entry.getValue().get());
-            //Unfortunately, a good API to read coproc properties didn't show up until
-            //HBase 2.0. Doing this the painful String-matching way to be compatible with 1.x
-            if (propKey.contains("coprocessor")){
-                if (propValue.contains(className)){
-                    Assert.assertEquals(className + " is configured incorrectly",
-                        expectedConfigValue,
-                        propValue);
-                    foundConfig = true;
-                    break;
-                }
-            }
-        }
-        Assert.assertTrue("Couldn't find config for " + className, foundConfig);
-    }
-
     private void createIndexTable(String schemaName, String tableName, String indexName)
         throws SQLException {
         Connection conn = getConnection();
@@ -300,5 +216,4 @@
         props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(isNamespaceMapped));
         return (PhoenixConnection) DriverManager.getConnection(getUrl(),props);
     }
-
-}
+}
\ No newline at end of file
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexTestUtil.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
index c97847d..bfa9ed6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexTestUtil.java
@@ -17,6 +17,7 @@
  */
 package org.apache.phoenix.end2end.index;
 
+import static org.apache.phoenix.end2end.index.IndexCoprocIT.INDEXER_CONFIG;
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
 import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
@@ -30,16 +31,28 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.BitSet;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.hbase.index.IndexRegionObserver;
+import org.apache.phoenix.hbase.index.Indexer;
+import org.apache.phoenix.hbase.index.covered.NonTxIndexBuilder;
 import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
+import org.apache.phoenix.index.GlobalIndexChecker;
+import org.apache.phoenix.index.PhoenixIndexBuilder;
+import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.query.QueryServicesOptions;
 import org.apache.phoenix.schema.ColumnNotFoundException;
 import org.apache.phoenix.schema.PColumn;
 import org.apache.phoenix.schema.PColumnFamily;
@@ -54,6 +67,7 @@
 import org.apache.phoenix.util.SchemaUtil;
 
 import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
+import org.junit.Assert;
 
 public class IndexTestUtil {
 
@@ -165,4 +179,102 @@
         return row.toRowMutations();
     }
 
-}
+    public static void downgradeCoprocs(String physicalTableName,
+            String physicalIndexName, Admin admin) throws Exception {
+        TableDescriptor baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
+        TableDescriptorBuilder baseDescBuilder = TableDescriptorBuilder.newBuilder(baseDescriptor);
+
+        assertCoprocsContains(IndexRegionObserver.class, baseDescriptor);
+        removeCoproc(IndexRegionObserver.class, baseDescBuilder, admin);
+
+        if (physicalIndexName != null) {
+            TableDescriptor indexDescriptor =
+                    admin.getDescriptor(TableName.valueOf(physicalIndexName));
+            assertCoprocsContains(GlobalIndexChecker.class, indexDescriptor);
+
+            TableDescriptorBuilder indexDescBuilder =
+                    TableDescriptorBuilder.newBuilder(indexDescriptor);
+            removeCoproc(IndexRegionObserver.class, indexDescBuilder, admin);
+            removeCoproc(GlobalIndexChecker.class, indexDescBuilder, admin);
+        }
+
+        Map<String, String> props = new HashMap<>();
+        props.put(NonTxIndexBuilder.CODEC_CLASS_NAME_KEY, PhoenixIndexCodec.class.getName());
+        Indexer.enableIndexing(baseDescBuilder, PhoenixIndexBuilder.class,
+                props, QueryServicesOptions.DEFAULT_COPROCESSOR_PRIORITY);
+        admin.modifyTable(baseDescBuilder.build());
+        baseDescriptor = admin.getDescriptor(TableName.valueOf(physicalTableName));
+        TableDescriptor indexDescriptor = null;
+        if (physicalIndexName != null) {
+            indexDescriptor = admin.getDescriptor(TableName.valueOf(physicalIndexName));
+        }
+        assertUsingOldCoprocs(baseDescriptor, indexDescriptor);
+    }
+
+
+    public static void assertCoprocsContains(Class clazz, TableDescriptor descriptor) {
+        String expectedCoprocName = clazz.getName();
+        boolean foundCoproc = isCoprocPresent(descriptor, expectedCoprocName);
+        Assert.assertTrue("Could not find coproc " + expectedCoprocName +
+                " in descriptor " + descriptor,foundCoproc);
+    }
+
+
+    public static boolean isCoprocPresent(TableDescriptor descriptor, String expectedCoprocName) {
+        boolean foundCoproc = false;
+        for (String coprocName : descriptor.getCoprocessors()){
+            if (coprocName.equals(expectedCoprocName)){
+                foundCoproc = true;
+                break;
+            }
+        }
+        return foundCoproc;
+    }
+
+    public static void removeCoproc(Class clazz, TableDescriptorBuilder descBuilder, Admin admin)
+            throws Exception {
+        descBuilder.removeCoprocessor(clazz.getName());
+        admin.modifyTable(descBuilder.build());
+    }
+
+    public static void assertUsingOldCoprocs(TableDescriptor baseDescriptor,
+            TableDescriptor indexDescriptor) {
+        assertCoprocsContains(Indexer.class, baseDescriptor);
+        assertCoprocConfig(baseDescriptor, Indexer.class.getName(),
+                INDEXER_CONFIG);
+        assertCoprocsNotContains(IndexRegionObserver.class, baseDescriptor);
+        if (indexDescriptor != null) {
+            assertCoprocsNotContains(IndexRegionObserver.class, indexDescriptor);
+            assertCoprocsNotContains(GlobalIndexChecker.class, indexDescriptor);
+        }
+    }
+
+    public static void assertCoprocConfig(TableDescriptor indexDesc,
+            String className, String expectedConfigValue){
+        boolean foundConfig = false;
+        for (Map.Entry<Bytes, Bytes> entry :
+                indexDesc.getValues().entrySet()){
+            String propKey = Bytes.toString(entry.getKey().get());
+            String propValue = Bytes.toString(entry.getValue().get());
+            //Unfortunately, a good API to read coproc properties didn't show up until
+            //HBase 2.0. Doing this the painful String-matching way to be compatible with 1.x
+            if (propKey.contains("coprocessor")){
+                if (propValue.contains(className)){
+                    Assert.assertEquals(className + " is configured incorrectly",
+                            expectedConfigValue,
+                            propValue);
+                    foundConfig = true;
+                    break;
+                }
+            }
+        }
+        Assert.assertTrue("Couldn't find config for " + className, foundConfig);
+    }
+
+    public static void assertCoprocsNotContains(Class clazz, TableDescriptor descriptor) {
+        String expectedCoprocName = clazz.getName();
+        boolean foundCoproc = isCoprocPresent(descriptor, expectedCoprocName);
+        Assert.assertFalse("Could find coproc " + expectedCoprocName +
+                " in descriptor " + descriptor,foundCoproc);
+    }
+}
\ No newline at end of file
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
index 98edf41..628d294 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java
@@ -30,14 +30,6 @@
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.hadoop.hbase.ArrayBackedTag;
-import org.apache.hadoop.hbase.CellScanner;
-import org.apache.hadoop.hbase.PhoenixTagType;
-import org.apache.hadoop.hbase.PrivateCellUtil;
-import org.apache.hadoop.hbase.RawCell;
-import org.apache.hadoop.hbase.Tag;
-import org.apache.hadoop.hbase.TagUtil;
-import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.thirdparty.com.google.common.collect.ArrayListMultimap;
 import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
 import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
@@ -101,6 +93,7 @@
 import org.apache.phoenix.trace.TracingUtils;
 import org.apache.phoenix.trace.util.NullSpan;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.SchemaUtil;
 import org.apache.phoenix.util.ServerUtil;
 import org.apache.phoenix.util.ServerUtil.ConnectionType;
@@ -733,7 +726,7 @@
         scan.setFilter(skipScanFilter);
         try (RegionScanner scanner = c.getEnvironment().getRegion().getScanner(scan)) {
             boolean more = true;
-            while(more) {
+            while (more) {
                 List<Cell> cells = new ArrayList<Cell>();
                 more = scanner.next(cells);
                 if (cells.isEmpty()) {
@@ -974,7 +967,7 @@
         context.populateOriginalMutations(miniBatchOp);
         // Need to add cell tags to Delete Marker before we do any index processing
         // since we add tags to tables which doesn't have indexes also.
-        setDeleteAttributes(miniBatchOp);
+        IndexUtil.setDeleteAttributes(miniBatchOp);
 
         /*
          * Exclusively lock all rows so we get a consistent read
@@ -1031,49 +1024,6 @@
         }
     }
 
-    /**
-     * Set Cell Tags to delete markers with source of operation attribute.
-     * @param miniBatchOp
-     * @throws IOException
-     */
-    private void setDeleteAttributes(MiniBatchOperationInProgress<Mutation> miniBatchOp)
-            throws IOException {
-        for (int i = 0; i < miniBatchOp.size(); i++) {
-            Mutation m = miniBatchOp.getOperation(i);
-            if (!(m instanceof  Delete)) {
-                // Ignore if it is not Delete type.
-                continue;
-            }
-            byte[] sourceOpAttr = m.getAttribute(QueryServices.SOURCE_OPERATION_ATTRIB);
-            if (sourceOpAttr == null) {
-                continue;
-            }
-            Tag sourceOpTag = new ArrayBackedTag(PhoenixTagType.SOURCE_OPERATION_TAG_TYPE,
-                    sourceOpAttr);
-            List<Cell> updatedCells = new ArrayList<>();
-            for (CellScanner cellScanner = m.cellScanner(); cellScanner.advance();) {
-                Cell cell = cellScanner.current();
-                RawCell rawCell = (RawCell)cell;
-                List<Tag> tags = new ArrayList<>();
-                Iterator<Tag> tagsIterator = rawCell.getTags();
-                while (tagsIterator.hasNext()) {
-                    tags.add(tagsIterator.next());
-                }
-                tags.add(sourceOpTag);
-                // TODO: PrivateCellUtil's IA is Private. HBASE-25328 adds a builder method
-                // TODO: for creating Tag which will be LP with IA.coproc
-                Cell updatedCell = PrivateCellUtil.createCell(cell, tags);
-                updatedCells.add(updatedCell);
-            }
-            m.getFamilyCellMap().clear();
-            // Clear and add new Cells to the Mutation.
-            for (Cell cell : updatedCells) {
-                Delete d = (Delete) m;
-                d.addDeleteMarker(cell);
-            }
-        }
-    }
-
     private void setBatchMutateContext(ObserverContext<RegionCoprocessorEnvironment> c, BatchMutateContext context) {
       this.batchMutateContext.set(context);
     }
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index 8b04faf..0501f28 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -81,6 +81,7 @@
 import org.apache.phoenix.trace.TracingUtils;
 import org.apache.phoenix.trace.util.NullSpan;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.ScanUtil;
 import org.apache.phoenix.util.ServerUtil;
 import org.apache.phoenix.util.ServerUtil.ConnectionType;
@@ -367,12 +368,15 @@
   public void preBatchMutateWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c,
           MiniBatchOperationInProgress<Mutation> miniBatchOp) throws Throwable {
 
+    // Need to add cell tags to Delete Marker before we do any index processing
+    // since we add tags to tables which doesn't have indexes also.
+    IndexUtil.setDeleteAttributes(miniBatchOp);
       // first group all the updates for a single row into a single update to be processed
       Map<ImmutableBytesPtr, MultiMutation> mutationsMap =
               new HashMap<ImmutableBytesPtr, MultiMutation>();
           
       Durability defaultDurability = Durability.SYNC_WAL;
-      if(c.getEnvironment().getRegion() != null) {
+      if (c.getEnvironment().getRegion() != null) {
           defaultDurability = c.getEnvironment().getRegion().getTableDescriptor().getDurability();
           defaultDurability = (defaultDurability == Durability.USE_DEFAULT) ? 
                   Durability.SYNC_WAL : defaultDurability;
@@ -512,7 +516,7 @@
           byte[] tableName = c.getEnvironment().getRegion().getTableDescriptor().getTableName().getName();
           Iterator<Pair<Mutation, byte[]>> indexUpdatesItr = indexUpdates.iterator();
           List<Mutation> localUpdates = new ArrayList<Mutation>(indexUpdates.size());
-          while(indexUpdatesItr.hasNext()) {
+          while (indexUpdatesItr.hasNext()) {
               Pair<Mutation, byte[]> next = indexUpdatesItr.next();
               if (Bytes.compareTo(next.getSecond(), tableName) == 0) {
                   localUpdates.add(next.getFirst());
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
index c40b5ea..0449c00 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/IndexUtil.java
@@ -39,6 +39,15 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hadoop.hbase.ArrayBackedTag;
+import org.apache.hadoop.hbase.CellScanner;
+import org.apache.hadoop.hbase.PhoenixTagType;
+import org.apache.hadoop.hbase.PrivateCellUtil;
+import org.apache.hadoop.hbase.RawCell;
+import org.apache.hadoop.hbase.Tag;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
+import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.thirdparty.com.google.common.cache.Cache;
 import org.apache.phoenix.thirdparty.com.google.common.cache.CacheBuilder;
 import org.apache.hadoop.hbase.Cell;
@@ -818,4 +827,50 @@
         }
     }
 
+    /**
+     * Set Cell Tags to delete markers with source of operation attribute.
+     * @param miniBatchOp miniBatchOp
+     * @throws IOException IOException
+     */
+    public static void setDeleteAttributes(
+            MiniBatchOperationInProgress<Mutation> miniBatchOp)
+            throws IOException {
+        for (int i = 0; i < miniBatchOp.size(); i++) {
+            Mutation m = miniBatchOp.getOperation(i);
+            if (!(m instanceof Delete)) {
+                // Ignore if it is not Delete type.
+                continue;
+            }
+            byte[] sourceOpAttr =
+                    m.getAttribute(QueryServices.SOURCE_OPERATION_ATTRIB);
+            if (sourceOpAttr == null) {
+                continue;
+            }
+            Tag sourceOpTag = new ArrayBackedTag(
+                    PhoenixTagType.SOURCE_OPERATION_TAG_TYPE, sourceOpAttr);
+            List<Cell> updatedCells = new ArrayList<>();
+            for (CellScanner cellScanner = m.cellScanner();
+                 cellScanner.advance();) {
+                Cell cell = cellScanner.current();
+                RawCell rawCell = (RawCell) cell;
+                List<Tag> tags = new ArrayList<>();
+                Iterator<Tag> tagsIterator = rawCell.getTags();
+                while (tagsIterator.hasNext()) {
+                    tags.add(tagsIterator.next());
+                }
+                tags.add(sourceOpTag);
+                // TODO: PrivateCellUtil's IA is Private.
+                // HBASE-25328 adds a builder methodfor creating Tag which
+                // will be LP with IA.coproc
+                Cell updatedCell = PrivateCellUtil.createCell(cell, tags);
+                updatedCells.add(updatedCell);
+            }
+            m.getFamilyCellMap().clear();
+            // Clear and add new Cells to the Mutation.
+            for (Cell cell : updatedCells) {
+                Delete d = (Delete) m;
+                d.addDeleteMarker(cell);
+            }
+        }
+    }
 }