HBASE-12065 Import tool is not restoring multiple DeleteFamily markers of a row (Maddineni Sukumar)

hbase-server/src/main/java/org/apache/hadoop/hbase/mapreduce/Import.java

@@ -173,7 +173,22 @@ public class Import extends Configured implements Tool {
 
         kv = convertKv(kv, cfRenameMap);
         // Deletes and Puts are gathered and written when finished
-        if (CellUtil.isDelete(kv)) {
+        /*
+         * If there are sequence of mutations and tombstones in an Export, and after Import the same
+         * sequence should be restored as it is. If we combine all Delete tombstones into single
+         * request then there is chance of ignoring few DeleteFamily tombstones, because if we
+         * submit multiple DeleteFamily tombstones in single Delete request then we are maintaining
+         * only newest in hbase table and ignoring other. Check - HBASE-12065
+         */
+        if (CellUtil.isDeleteFamily(kv)) {
+          Delete deleteFamily = new Delete(key.get());
+          deleteFamily.addDeleteMarker(kv);
+          if (durability != null) {
+            deleteFamily.setDurability(durability);
+          }
+          deleteFamily.setClusterIds(clusterIds);
+          context.write(key, deleteFamily);
+        } else if (CellUtil.isDelete(kv)) {
           if (delete == null) {
             delete = new Delete(key.get());
           }

hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestImportExport.java

@@ -339,6 +339,83 @@ public class TestImportExport {
     assertEquals(now, res[6].getTimestamp());
     t.close();
   }
+  
+  
+  @Test
+  public void testWithMultipleDeleteFamilyMarkersOfSameRowSameFamily() throws Exception {
+    String EXPORT_TABLE = "exportWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
+    HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(EXPORT_TABLE));
+    desc.addFamily(new HColumnDescriptor(FAMILYA)
+        .setMaxVersions(5)
+        .setKeepDeletedCells(true)
+    );
+    UTIL.getHBaseAdmin().createTable(desc);
+    HTable exportT = new HTable(UTIL.getConfiguration(), EXPORT_TABLE);
+
+    //Add first version of QUAL
+    Put p = new Put(ROW1);
+    p.add(FAMILYA, QUAL, now, QUAL);
+    exportT.put(p);
+
+    //Add Delete family marker
+    Delete d = new Delete(ROW1, now+3);
+    exportT.delete(d);
+
+    //Add second version of QUAL
+    p = new Put(ROW1);
+    p.add(FAMILYA, QUAL, now+5, "s".getBytes());
+    exportT.put(p);
+
+    //Add second Delete family marker
+    d = new Delete(ROW1, now+7);
+    exportT.delete(d);
+    
+    
+    String[] args = new String[] {
+        "-D" + Export.RAW_SCAN + "=true",
+        EXPORT_TABLE,
+        FQ_OUTPUT_DIR,
+        "1000", // max number of key versions per key to export
+    };
+    assertTrue(runExport(args));
+
+    String IMPORT_TABLE = "importWithMultipleDeleteFamilyMarkersOfSameRowSameFamily";
+    desc = new HTableDescriptor(TableName.valueOf(IMPORT_TABLE));
+    desc.addFamily(new HColumnDescriptor(FAMILYA)
+        .setMaxVersions(5)
+        .setKeepDeletedCells(true)
+    );
+    UTIL.getHBaseAdmin().createTable(desc);
+    
+    HTable importT = new HTable(UTIL.getConfiguration(), IMPORT_TABLE);
+    args = new String[] {
+        IMPORT_TABLE,
+        FQ_OUTPUT_DIR
+    };
+    assertTrue(runImport(args));
+
+    Scan s = new Scan();
+    s.setMaxVersions();
+    s.setRaw(true);
+    
+    ResultScanner importedTScanner = importT.getScanner(s);
+    Result importedTResult = importedTScanner.next();
+    
+    ResultScanner exportedTScanner = exportT.getScanner(s);
+    Result  exportedTResult =  exportedTScanner.next();
+    try
+    {
+      Result.compareResults(exportedTResult, importedTResult);
+    }
+    catch (Exception e) {
+      fail("Original and imported tables data comparision failed with error:"+e.getMessage());
+    }
+    finally
+    {
+      exportT.close();
+      importT.close();
+    }
+  }
 
   /**
    * Create a simple table, run an Export Job on it, Import with filtering on,  verify counts,