HBASE-21475 Put mutation (having TTL set) added via co-processor is retrieved even after TTL expires

Signed-off-by: Andrew Purtell <apurtell@apache.org>

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -3822,6 +3822,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
           // Returned mutations from coprocessor correspond to the Mutation at index i. We can
           // directly add the cells from those mutations to the familyMaps of this mutation.
           Map<byte[], List<Cell>> cpFamilyMap = cpMutation.getFamilyCellMap();
+          region.rewriteCellTags(cpFamilyMap, mutation);
           // will get added to the memStore later
           mergeFamilyMaps(familyCellMaps[i], cpFamilyMap);
 

hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverForAddingMutationsFromCoprocessors.java

@@ -200,6 +200,39 @@ public class TestRegionObserverForAddingMutationsFromCoprocessors {
     }
   }
 
+  @Test
+  public void testPutWithTTL() throws Exception {
+    createTable(TestPutWithTTLCoprocessor.class.getName());
+
+    try (Table t = util.getConnection().getTable(tableName)) {
+      t.put(new Put(row1).addColumn(test, dummy, dummy).setTTL(3000));
+      assertRowCount(t, 2);
+      // wait long enough for the TTL to expire
+      Thread.sleep(5000);
+      assertRowCount(t, 0);
+    }
+  }
+
+  public static class TestPutWithTTLCoprocessor implements RegionCoprocessor, RegionObserver {
+    @Override
+    public Optional<RegionObserver> getRegionObserver() {
+      return Optional.of(this);
+    }
+
+    @Override
+    public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
+        MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
+      Mutation mut = miniBatchOp.getOperation(0);
+      List<Cell> cells = mut.getFamilyCellMap().get(test);
+      Put[] puts = new Put[] {
+          new Put(Bytes.toBytes("cpPut")).addColumn(test, dummy, cells.get(0).getTimestamp(),
+            Bytes.toBytes("cpdummy")).setTTL(mut.getTTL())
+          };
+      LOG.info("Putting:" + Arrays.toString(puts));
+      miniBatchOp.addOperationsFromCP(0, puts);
+    }
+  }
+
   public static class TestMultiMutationCoprocessor implements RegionCoprocessor, RegionObserver {
     @Override
     public Optional<RegionObserver> getRegionObserver() {