feat: concurrent writes in MirroringTable

Unoperate · Oct 14, 2021 · 4dfb80a · 4dfb80a
1 parent f115ceb
commit 4dfb80a
Show file tree

Hide file tree

Showing 11 changed files with 647 additions and 249 deletions.
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
@@ -64,6 +64,7 @@ public class MirroringConnection implements Connection {
   private final Connection secondaryConnection;
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final AtomicBoolean aborted = new AtomicBoolean(false);
+  private final boolean performWritesConcurrently;
 
   /**
    * The constructor called from {@link
@@ -117,6 +118,7 @@ public MirroringConnection(Configuration conf, boolean managed, ExecutorService
         new SecondaryWriteErrorConsumerWithMetrics(
             this.mirroringTracer, secondaryWriteErrorConsumer);
     this.readSampler = new ReadSampler(this.configuration.mirroringOptions.readSamplingRate);
+    this.performWritesConcurrently = this.configuration.mirroringOptions.performWritesConcurrently;
   }
 
   @Override
@@ -154,6 +156,7 @@ public Table call() throws IOException {
               this.flowController,
               this.secondaryWriteErrorConsumer,
               this.readSampler,
+              this.performWritesConcurrently,
               this.mirroringTracer);
       this.referenceCounter.holdReferenceUntilClosing(table);
       return table;

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
@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.mirroring.hbase1_x;
 
 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_FLOW_CONTROLLER_MAX_OUTSTANDING_REQUESTS;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_FLOW_CONTROLLER_STRATEGY_CLASS;
 import static com.google.cloud.bigtable.mirroring.hbase1_x.utils.MirroringConfigurationHelper.MIRRORING_MISMATCH_DETECTOR_CLASS;
@@ -46,6 +47,8 @@ public class MirroringOptions {
   public final String writeErrorLogAppenderClass;
   public final String writeErrorLogSerializerClass;
 
+  public final boolean performWritesConcurrently;
+
   public MirroringOptions(Configuration configuration) {
     this.mismatchDetectorClass =
         configuration.get(
@@ -76,5 +79,6 @@ public MirroringOptions(Configuration configuration) {
     this.writeErrorLogSerializerClass =
         configuration.get(
             MIRRORING_WRITE_ERROR_LOG_SERIALIZER_CLASS, DefaultSerializer.class.getCanonicalName());
+    this.performWritesConcurrently = configuration.getBoolean(MIRRORING_CONCURRENT_WRITES, false);
   }
 }
diff --git a/....x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringResultScanner.java b/....x/src/main/java/com/google/cloud/bigtable/mirroring/hbase1_x/MirroringResultScanner.java
@@ -220,7 +220,7 @@ private <T> void scheduleRequest(
       return;
     }
     this.listenableReferenceCounter.holdReferenceUntilCompletion(
-        RequestScheduling.scheduleVerificationAndRequestWithFlowControl(
+        RequestScheduling.scheduleRequestAndVerificationWithFlowControl(
             requestResourcesDescription,
             nextSupplier,
             this.mirroringTracer.spanFactory.wrapReadVerificationCallback(scannerNext),