feat: Client-side timestamping

Unoperate · Jan 21, 2022 · daf5f22 · daf5f22
1 parent 1f08c09
commit daf5f22
Show file tree

Hide file tree

Showing 38 changed files with 547 additions and 44 deletions.
diff --git a/...on-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/TestBufferedMutator.java b/...on-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/TestBufferedMutator.java
@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.hbase.mirroring;
 
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_CONCURRENT_WRITES;
+import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_FLOW_CONTROLLER_STRATEGY_MAX_OUTSTANDING_REQUESTS;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_SYNCHRONOUS_WRITES;
 import static com.google.common.truth.Truth.assertThat;
@@ -104,6 +105,7 @@ private Configuration createConfiguration() {
     // testing the other case here anyways) and set it to true to meet the requirements otherwise.
     configuration.set(MIRRORING_CONCURRENT_WRITES, String.valueOf(this.mutateConcurrently));
     configuration.set(MIRRORING_SYNCHRONOUS_WRITES, String.valueOf(this.mutateConcurrently));
+    configuration.setBoolean(MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS, true);
     return configuration;
   }
 

diff --git a/...ion-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/TestMirroringTable.java b/...ion-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/TestMirroringTable.java
diff --git a/...-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/utils/DatabaseHelpers.java b/...-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/utils/DatabaseHelpers.java
@@ -15,6 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.mirroring.utils;
 
+import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS;
 import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
@@ -184,6 +185,12 @@ public MirroringConnection createConnection() throws IOException {
     return connectionRule.createConnection(this.executorServiceRule.executorService);
   }
 
+  public MirroringConnection createTimestampedConnection() throws IOException {
+    Configuration conf = ConfigurationHelper.newConfiguration();
+    conf.setBoolean(MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS, true);
+    return connectionRule.createConnection(this.executorServiceRule.executorService, conf);
+  }
+
   public MirroringConnection createConnection(Configuration configuration) throws IOException {
     return connectionRule.createConnection(this.executorServiceRule.executorService, configuration);
   }

diff --git a/...egration-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/utils/Helpers.java b/...egration-tests/src/test/java/com/google/cloud/bigtable/hbase/mirroring/utils/Helpers.java
@@ -43,7 +43,7 @@ public static Put createPut(
 
   public static Put createPut(int id, byte[] family, byte[] qualifier) {
     byte[] rowAndValue = Longs.toByteArray(id);
-    return createPut(rowAndValue, family, qualifier, id, rowAndValue);
+    return createPut(rowAndValue, family, qualifier, rowAndValue);
   }
 
   public static Get createGet(byte[] row, byte[] family, byte[] qualifier) {

diff --git a/...t-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringConnection.java b/...t-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringConnection.java
@@ -26,6 +26,8 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringSpanConstants.HBaseOperation;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.referencecounting.ListenableReferenceCounter;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.TimestamperFactory;
 import com.google.cloud.bigtable.mirroring.hbase1_x.verification.MismatchDetector;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -90,6 +92,7 @@ public class MirroringConnection implements Connection {
   protected final boolean performWritesConcurrently;
 
   protected final boolean waitForSecondaryWrites;
+  protected final Timestamper timestamper;
 
   /**
    * The constructor called from {@link
@@ -162,6 +165,9 @@ public MirroringConnection(Configuration conf, boolean managed, ExecutorService
     this.readSampler = new ReadSampler(this.configuration.mirroringOptions.readSamplingRate);
     this.performWritesConcurrently = this.configuration.mirroringOptions.performWritesConcurrently;
     this.waitForSecondaryWrites = this.configuration.mirroringOptions.waitForSecondaryWrites;
+    this.timestamper =
+        TimestamperFactory.create(
+            this.configuration.mirroringOptions.enableDefaultClientSideTimestamps);
   }
 
   @Override
@@ -198,6 +204,7 @@ public Table call() throws IOException {
           this.flowController,
           this.secondaryWriteErrorConsumer,
           this.readSampler,
+          this.timestamper,
           this.performWritesConcurrently,
           this.waitForSecondaryWrites,
           this.mirroringTracer,
@@ -223,6 +230,7 @@ public BufferedMutator getBufferedMutator(BufferedMutatorParams bufferedMutatorP
         flowController,
         executorService,
         secondaryWriteErrorConsumer,
+        timestamper,
         mirroringTracer);
   }
 

diff --git a/...ient-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringOptions.java b/...ient-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringOptions.java
@@ -18,6 +18,7 @@
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_BUFFERED_MUTATOR_BYTES_TO_FLUSH;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_CONCURRENT_WRITES;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_CONNECTION_CONNECTION_TERMINATION_TIMEOUT;
+import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_FAILLOG_DROP_ON_OVERFLOW_KEY;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_FAILLOG_MAX_BUFFER_SIZE_KEY;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_FAILLOG_PREFIX_PATH_KEY;
@@ -106,6 +107,7 @@ public static class Faillog {
   public final Faillog faillog;
 
   public final int resultScannerBufferedMismatchedResults;
+  public final boolean enableDefaultClientSideTimestamps;
 
   public MirroringOptions(Configuration configuration) {
     this.mismatchDetectorFactoryClass =
@@ -153,5 +155,8 @@ public MirroringOptions(Configuration configuration) {
         "Performing writes concurrently and not waiting for writes is forbidden. "
             + "It has no advantage over performing writes asynchronously and not waiting for them.");
     this.faillog = new Faillog(configuration);
+
+    this.enableDefaultClientSideTimestamps =
+        configuration.getBoolean(MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS, false);
   }
 }
diff --git a/...client-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringTable.java b/...client-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringTable.java
@@ -36,6 +36,7 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.referencecounting.HierarchicalReferenceCounter;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.referencecounting.ReferenceCounter;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
 import com.google.cloud.bigtable.mirroring.hbase1_x.verification.MismatchDetector;
 import com.google.cloud.bigtable.mirroring.hbase1_x.verification.VerificationContinuationFactory;
 import com.google.common.annotations.VisibleForTesting;
@@ -114,6 +115,7 @@ public boolean apply(@NullableDecl Object o) {
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final SettableFuture<Void> closedFuture = SettableFuture.create();
   private final int resultScannerBufferedMismatchedResults;
+  private final Timestamper timestamper;
   /**
    * @param executorService ExecutorService is used to perform operations on secondaryTable and
    *     verification tasks.
@@ -129,6 +131,7 @@ public MirroringTable(
       FlowController flowController,
       SecondaryWriteErrorConsumer secondaryWriteErrorConsumer,
       ReadSampler readSampler,
+      Timestamper timestamper,
       boolean performWritesConcurrently,
       boolean waitForSecondaryWrites,
       MirroringTracer mirroringTracer,
@@ -148,6 +151,7 @@ public MirroringTable(
     this.mirroringTracer = mirroringTracer;
     this.requestScheduler =
         new RequestScheduler(flowController, this.mirroringTracer, this.referenceCounter);
+    this.timestamper = timestamper;
     this.batcher =
         new Batcher(
             this.primaryTable,
@@ -156,6 +160,7 @@ public MirroringTable(
             this.secondaryWriteErrorConsumer,
             this.verificationContinuationFactory,
             this.readSampler,
+            this.timestamper,
             resultIsFaultyPredicate,
             waitForSecondaryWrites,
             performWritesConcurrently,
@@ -545,6 +550,7 @@ private boolean checkAndMutateWithSpan(
       final byte[] value,
       final RowMutations rowMutations)
       throws IOException {
+    this.timestamper.fillTimestamp(rowMutations);
     boolean wereMutationsApplied =
         this.mirroringTracer.spanFactory.wrapPrimaryOperation(
             new CallableThrowingIOException<Boolean>() {

diff --git a/...cloud/bigtable/mirroring/hbase1_x/bufferedmutator/ConcurrentMirroringBufferedMutator.java b/...cloud/bigtable/mirroring/hbase1_x/bufferedmutator/ConcurrentMirroringBufferedMutator.java
@@ -22,6 +22,7 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.flowcontrol.RequestResourcesDescription;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringSpanConstants.HBaseOperation;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.MapMaker;
 import com.google.common.util.concurrent.FutureCallback;
@@ -78,6 +79,7 @@ public ConcurrentMirroringBufferedMutator(
       BufferedMutatorParams bufferedMutatorParams,
       MirroringConfiguration configuration,
       ExecutorService executorService,
+      Timestamper timestamper,
       MirroringTracer mirroringTracer)
       throws IOException {
     super(
@@ -86,6 +88,7 @@ public ConcurrentMirroringBufferedMutator(
         bufferedMutatorParams,
         configuration,
         executorService,
+        timestamper,
         mirroringTracer);
   }
 

diff --git a/...om/google/cloud/bigtable/mirroring/hbase1_x/bufferedmutator/MirroringBufferedMutator.java b/...om/google/cloud/bigtable/mirroring/hbase1_x/bufferedmutator/MirroringBufferedMutator.java
@@ -24,6 +24,7 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringSpanConstants.HBaseOperation;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.referencecounting.ListenableReferenceCounter;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -60,6 +61,9 @@
  */
 @InternalApi("For internal usage only")
 public abstract class MirroringBufferedMutator<BufferEntryType> implements BufferedMutator {
+
+  private final Timestamper timestamper;
+
   public static BufferedMutator create(
       boolean concurrent,
       Connection primaryConnection,
@@ -69,6 +73,7 @@ public static BufferedMutator create(
       FlowController flowController,
       ExecutorService executorService,
       SecondaryWriteErrorConsumer secondaryWriteErrorConsumer,
+      Timestamper timestamper,
       MirroringTracer mirroringTracer)
       throws IOException {
     if (concurrent) {
@@ -78,6 +83,7 @@ public static BufferedMutator create(
           bufferedMutatorParams,
           configuration,
           executorService,
+          timestamper,
           mirroringTracer);
     } else {
       return new SequentialMirroringBufferedMutator(
@@ -88,6 +94,7 @@ public static BufferedMutator create(
           flowController,
           executorService,
           secondaryWriteErrorConsumer,
+          timestamper,
           mirroringTracer);
     }
   }
@@ -121,6 +128,7 @@ public MirroringBufferedMutator(
       BufferedMutatorParams bufferedMutatorParams,
       MirroringConfiguration configuration,
       ExecutorService executorService,
+      Timestamper timestamper,
       MirroringTracer mirroringTracer)
       throws IOException {
     this.userListener = bufferedMutatorParams.getListener();
@@ -167,13 +175,16 @@ public void onException(
             this.mutationsBufferFlushThresholdBytes,
             this.ongoingFlushesCounter,
             this.mirroringTracer);
+    this.timestamper = timestamper;
   }
 
   @Override
   public void mutate(Mutation mutation) throws IOException {
     try (Scope scope =
         this.mirroringTracer.spanFactory.operationScope(HBaseOperation.BUFFERED_MUTATOR_MUTATE)) {
-      mutateScoped(Collections.singletonList(mutation));
+      List<Mutation> m = Collections.singletonList(mutation);
+      timestamper.fillTimestamp(m);
+      mutateScoped(m);
     }
   }
 
@@ -182,6 +193,7 @@ public void mutate(final List<? extends Mutation> list) throws IOException {
     try (Scope scope =
         this.mirroringTracer.spanFactory.operationScope(
             HBaseOperation.BUFFERED_MUTATOR_MUTATE_LIST)) {
+      timestamper.fillTimestamp(list);
       mutateScoped(list);
     }
   }

diff --git a/...cloud/bigtable/mirroring/hbase1_x/bufferedmutator/SequentialMirroringBufferedMutator.java b/...cloud/bigtable/mirroring/hbase1_x/bufferedmutator/SequentialMirroringBufferedMutator.java
@@ -26,6 +26,7 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.flowcontrol.RequestResourcesDescription;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringSpanConstants.HBaseOperation;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
 import com.google.common.collect.MapMaker;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
@@ -137,6 +138,7 @@ public SequentialMirroringBufferedMutator(
       FlowController flowController,
       ExecutorService executorService,
       SecondaryWriteErrorConsumer secondaryWriteErrorConsumer,
+      Timestamper timestamper,
       MirroringTracer mirroringTracer)
       throws IOException {
     super(
@@ -145,6 +147,7 @@ public SequentialMirroringBufferedMutator(
         bufferedMutatorParams,
         configuration,
         executorService,
+        timestamper,
         mirroringTracer);
     this.secondaryWriteErrorConsumer = secondaryWriteErrorConsumer;
     this.flowController = flowController;

diff --git a/...-client-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/utils/Batcher.java b/...-client-1.x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/utils/Batcher.java
@@ -29,6 +29,7 @@
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.flowcontrol.RequestResourcesDescription;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringSpanConstants.HBaseOperation;
 import com.google.cloud.bigtable.mirroring.hbase1_x.utils.mirroringmetrics.MirroringTracer;
+import com.google.cloud.bigtable.mirroring.hbase1_x.utils.timestamper.Timestamper;
 import com.google.cloud.bigtable.mirroring.hbase1_x.verification.VerificationContinuationFactory;
 import com.google.common.base.Function;
 import com.google.common.base.Preconditions;
@@ -70,6 +71,7 @@ public class Batcher {
   private final boolean waitForSecondaryWrites;
   private final boolean performWritesConcurrently;
   private final MirroringTracer mirroringTracer;
+  private final Timestamper timestamper;
 
   public Batcher(
       Table primaryTable,
@@ -78,6 +80,7 @@ public Batcher(
       SecondaryWriteErrorConsumer secondaryWriteErrorConsumer,
       VerificationContinuationFactory verificationContinuationFactory,
       ReadSampler readSampler,
+      Timestamper timestamper,
       Predicate<Object> resultIsFaultyPredicate,
       boolean waitForSecondaryWrites,
       boolean performWritesConcurrently,
@@ -92,6 +95,7 @@ public Batcher(
     this.waitForSecondaryWrites = waitForSecondaryWrites;
     this.performWritesConcurrently = performWritesConcurrently;
     this.mirroringTracer = mirroringTracer;
+    this.timestamper = timestamper;
   }
 
   public void batchSingleWriteOperation(Row operation) throws IOException {
@@ -129,6 +133,7 @@ public <R> void batch(
       final Object[] results,
       @Nullable final Callback<R> callback)
       throws IOException, InterruptedException {
+    timestamper.fillTimestamp(inputOperations);
     final RewrittenIncrementAndAppendIndicesInfo<? extends Row> actions =
         new RewrittenIncrementAndAppendIndicesInfo<>(inputOperations);
     Log.trace(

diff --git a/...java/com/google/cloud/bigtable/mirroring/hbase1_x/utils/MirroringConfigurationHelper.java b/...java/com/google/cloud/bigtable/mirroring/hbase1_x/utils/MirroringConfigurationHelper.java
@@ -30,6 +30,8 @@
 import java.util.Objects;
 import java.util.regex.Pattern;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.BufferedMutator;
+import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.client.Table;
@@ -280,6 +282,21 @@ public class MirroringConfigurationHelper {
   public static final String MIRRORING_SCANNER_BUFFERED_MISMATCHED_READS =
       "google.bigtable.mirroring.result-scanner.buffered-mismatched-reads";
 
+  /**
+   * If set to {@code true} the Mirroring Client will automatically add timestamps to {@link
+   * org.apache.hadoop.hbase.client.Put}s without timestamp set based on client's host local time.
+   * Client-side timestamps assigned by {@link Table}s and {@link BufferedMutator}`s created by one
+   * {@link Connection} are always increasing, even if system clock is moved backwards, for example
+   * by NTP or manually by the user. To enable client-side timestamping set this key to `true`.
+   *
+   * <p>Be aware that client-side timestamping modifies only `Put`s - `Delete`s, `Increment`s and
+   * `Append`s are not affected by this setting and will cause inconsistencies between databases.
+   *
+   * <p>Default value: false.
+   */
+  public static final String MIRRORING_ENABLE_DEFAULT_CLIENT_SIDE_TIMESTAMPS =
+      "google.bigtable.mirroring.enable-default-client-side-timestamps";
+
   public static void fillConnectionConfigWithClassImplementation(
       Configuration connectionConfig,
       Configuration config,