feat: flag a put in future timestamp and record a metrics (#3534)

* add future delete mutation metrics

* feat: add test

* flag puts at future timestamp

* fix formatting

* add thorough test

* address comments:

Author: sushanb

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/adapters/IncompatibleMutationAdapter.java

@@ -18,6 +18,7 @@
 
 import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY;
 import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.INCOMPATIBLE_MUTATION_METRIC_KEY;
+import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.PUTS_IN_FUTURE_METRIC_KEY;
 
 import com.google.cloud.bigtable.hbase.adapters.DeleteAdapter;
 import com.google.cloud.bigtable.hbase.replication.metrics.MetricsExporter;
@@ -56,6 +57,10 @@ private void incrementIncompatibleMutations() {
     metricsExporter.incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 1);
   }
 
+  private void incrementPutsInFutureMutations() {
+    metricsExporter.incCounters(PUTS_IN_FUTURE_METRIC_KEY, 1);
+  }
+
   /**
    * Creates an IncompatibleMutationAdapter with HBase configuration, MetricSource, and CBT
    * connection.
@@ -101,6 +106,11 @@ public final List<Cell> adaptIncompatibleMutations(BigtableWALEntry walEntry) {
       Cell cell = cellsToAdapt.get(index);
       // All puts are valid.
       if (cell.getTypeByte() == KeyValue.Type.Put.getCode()) {
+        // flag if put is issued for future timestamp
+        // do not log as we might fill up disk space due condition being true from clock skew
+        if (cell.getTimestamp() > walEntry.getWalWriteTime()) {
+          incrementPutsInFutureMutations();
+        }
         returnedCells.add(cell);
         continue;
       }

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/metrics/HBaseToCloudBigtableReplicationMetrics.java

@@ -24,4 +24,5 @@ private HBaseToCloudBigtableReplicationMetrics() {}
   public static final String INCOMPATIBLE_MUTATION_METRIC_KEY = "bigtableIncompatibleMutations";
   public static final String DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY =
       "bigtableDroppedIncompatibleMutations";
+  public static final String PUTS_IN_FUTURE_METRIC_KEY = "bigtablePutsInFutureMutations";
 }

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/adapters/IncompatibleMutationAdapterTest.java

@@ -18,10 +18,12 @@
 
 import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY;
 import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.INCOMPATIBLE_MUTATION_METRIC_KEY;
+import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.PUTS_IN_FUTURE_METRIC_KEY;
 import static org.mockito.Mockito.reset;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 import com.google.cloud.bigtable.hbase.replication.metrics.MetricsExporter;
 import java.nio.charset.StandardCharsets;
@@ -35,6 +37,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.Type;
 import org.apache.hadoop.hbase.client.Connection;
 import org.junit.After;
 import org.junit.Assert;
@@ -226,4 +229,23 @@ public void testIncompatibleDeletesAreDropped() {
     verify(metricsExporter, times(1)).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 1);
     verify(metricsExporter, times(1)).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 1);
   }
+
+  @Test
+  public void testFuturePutAreFlagged() {
+    ArrayList<Cell> walEntryCells = new ArrayList<>();
+    Cell put1 = new KeyValue(rowKey, cf, qual, 900, Type.Put);
+    Cell put2 = new KeyValue(rowKey, cf, qual, 1005L, Type.Put);
+    walEntryCells.add(put1);
+    walEntryCells.add(put2);
+    BigtableWALEntry walEntry = new BigtableWALEntry(1000L, walEntryCells, tableName);
+
+    Assert.assertEquals(
+        Arrays.asList(put1, put2),
+        incompatibleMutationAdapter.adaptIncompatibleMutations(walEntry));
+
+    verify(metricsExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0);
+    verify(metricsExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0);
+    verify(metricsExporter, times(1)).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 1);
+    verifyNoMoreInteractions(metricsExporter);
+  }
 }