feat: fix windows build failure by using nanoSeconds instead of Instant for better accuracy. (#1887)

GaoleMeng · gcf-owl-bot[bot] · web-flow · commit e5cd7df54e2f · 2022-11-16T13:17:33.000-08:00
* feat: Split writer into connection worker and wrapper, this is a prerequisite for multiplexing client * feat: add connection worker pool skeleton, used for multiplexing client * feat: add Load api for connection worker for multiplexing client * feat: add multiplexing support to connection worker. We will treat every new stream name as a switch of destinationt * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: port the multiplexing client core algorithm and basic tests also fixed a tiny bug inside fake bigquery write impl for getting thre response from offset * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: wire multiplexing connection pool to stream writer * feat: some fixes for multiplexing client * feat: fix some todos, and reject the mixed behavior of passed in client or not * feat: fix the bug that we may peek into the write_stream field but it's possible the proto schema does not contain this field * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: fix the bug that we may peek into the write_stream field but it's possible the proto schema does not contain this field * feat: add getInflightWaitSeconds implementation * feat: Add schema comparision in connection loop to ensure schema update for the same stream name can be notified * 🦉 Updates from OwlBot post-processor See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md * feat: add schema update support to multiplexing * fix: fix windows build bug: windows Instant resolution is different with linux * fix: fix another failing tests for windows build * fix: fix another test failure for Windows build Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java
@@ -31,7 +31,6 @@
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
-import java.time.Instant;
 import java.util.Comparator;
 import java.util.Deque;
 import java.util.HashMap;
@@ -610,7 +609,7 @@ private void requestCallback(AppendRowsResponse response) {
     this.lock.lock();
     if (response.hasUpdatedSchema()) {
       this.updatedSchema =
-          TableSchemaAndTimestamp.create(Instant.now(), response.getUpdatedSchema());
+          TableSchemaAndTimestamp.create(System.nanoTime(), response.getUpdatedSchema());
     }
     try {
       // Had a successful connection with at least one result, reset retries.
@@ -824,12 +823,12 @@ public static void setOverwhelmedCountsThreshold(double newThreshold) {
   @AutoValue
   abstract static class TableSchemaAndTimestamp {
     // Shows the timestamp updated schema is reported from response
-    abstract Instant updateTimeStamp();
+    abstract long updateTimeStamp();
 
     // The updated schema returned from server.
     abstract TableSchema updatedSchema();
 
-    static TableSchemaAndTimestamp create(Instant updateTimeStamp, TableSchema updatedSchema) {
+    static TableSchemaAndTimestamp create(long updateTimeStamp, TableSchema updatedSchema) {
       return new AutoValue_ConnectionWorker_TableSchemaAndTimestamp(updateTimeStamp, updatedSchema);
     }
   }
diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorkerPool.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorkerPool.java
@@ -26,7 +26,6 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.MoreExecutors;
 import java.io.IOException;
-import java.time.Instant;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
@@ -268,7 +267,7 @@ public ApiFuture<AppendRowsResponse> append(
           if (response.getWriteStream() != "" && response.hasUpdatedSchema()) {
             tableNameToUpdatedSchema.put(
                 response.getWriteStream(),
-                TableSchemaAndTimestamp.create(Instant.now(), response.getUpdatedSchema()));
+                TableSchemaAndTimestamp.create(System.nanoTime(), response.getUpdatedSchema()));
           }
           return response;
         },
diff --git a/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java b/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java
@@ -29,7 +29,6 @@
 import io.grpc.Status.Code;
 import io.grpc.StatusRuntimeException;
 import java.io.IOException;
-import java.time.Instant;
 import java.util.Map;
 import java.util.Objects;
 import java.util.UUID;
@@ -88,7 +87,7 @@ public class StreamWriter implements AutoCloseable {
       new ConcurrentHashMap<>();
 
   /** Creation timestamp of this streamwriter */
-  private final Instant creationTimestamp;
+  private final long creationTimestamp;
 
   /** The maximum size of one request. Defined by the API. */
   public static long getApiMaxRequestBytes() {
@@ -260,7 +259,7 @@ private StreamWriter(Builder builder) throws IOException {
         client.close();
       }
     }
-    this.creationTimestamp = Instant.now();
+    this.creationTimestamp = System.nanoTime();
   }
 
   @VisibleForTesting
@@ -414,12 +413,12 @@ public synchronized TableSchema getUpdatedSchema() {
     if (tableSchemaAndTimestamp == null) {
       return null;
     }
-    return creationTimestamp.compareTo(tableSchemaAndTimestamp.updateTimeStamp()) < 0
+    return creationTimestamp < tableSchemaAndTimestamp.updateTimeStamp()
         ? tableSchemaAndTimestamp.updatedSchema()
         : null;
   }
 
-  Instant getCreationTimestamp() {
+  long getCreationTimestamp() {
     return creationTimestamp;
   }
 
diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/JsonStreamWriterTest.java
@@ -530,8 +530,6 @@ public void run() throws Throwable {
   public void testSimpleSchemaUpdate() throws Exception {
     try (JsonStreamWriter writer =
         getTestJsonStreamWriterBuilder(TEST_STREAM, TABLE_SCHEMA).build()) {
-      // Sleep for a short period to make sure the creation timestamp is older.
-      Sleeper.DEFAULT.sleep(200);
       testBigQueryWrite.addResponse(
           AppendRowsResponse.newBuilder()
               .setAppendResult(
diff --git a/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java b/google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java
@@ -20,7 +20,6 @@
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
-import com.google.api.client.util.Sleeper;
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.batching.FlowController;
 import com.google.api.gax.core.NoCredentialsProvider;
@@ -310,8 +309,6 @@ private void testUpdatedSchemaFetch(boolean enableMultiplexing)
     AppendRowsResponse response =
         writer.append(createProtoRows(new String[] {String.valueOf(0)}), 0).get();
     assertEquals(writer.getUpdatedSchema(), UPDATED_TABLE_SCHEMA);
-    // Sleep for a short period to make sure the creation timestamp is older.
-    Sleeper.DEFAULT.sleep(200);
 
     // Create another writer, although it's the same stream name but the time stamp is newer, thus
     // the old updated schema won't get returned.

Original file line number	Diff line number	Diff line change
`@@ -29,7 +29,6 @@`
`29`	`29`	`import io.grpc.Status.Code;`
`30`	`30`	`import io.grpc.StatusRuntimeException;`
`31`	`31`	`import java.io.IOException;`
`32`		`-import java.time.Instant;`
`33`	`32`	`import java.util.Map;`
`34`	`33`	`import java.util.Objects;`
`35`	`34`	`import java.util.UUID;`
`@@ -88,7 +87,7 @@ public class StreamWriter implements AutoCloseable {`
`88`	`87`	`new ConcurrentHashMap<>();`
`89`	`88`
`90`	`89`	`/** Creation timestamp of this streamwriter */`
`91`		`- private final Instant creationTimestamp;`
	`90`	`+ private final long creationTimestamp;`
`92`	`91`
`93`	`92`	`/** The maximum size of one request. Defined by the API. */`
`94`	`93`	`public static long getApiMaxRequestBytes() {`
`@@ -260,7 +259,7 @@ private StreamWriter(Builder builder) throws IOException {`
`260`	`259`	`client.close();`
`261`	`260`	`}`
`262`	`261`	`}`
`263`		`- this.creationTimestamp = Instant.now();`
	`262`	`+ this.creationTimestamp = System.nanoTime();`
`264`	`263`	`}`
`265`	`264`
`266`	`265`	`@VisibleForTesting`
`@@ -414,12 +413,12 @@ public synchronized TableSchema getUpdatedSchema() {`
`414`	`413`	`if (tableSchemaAndTimestamp == null) {`
`415`	`414`	`return null;`
`416`	`415`	`}`
`417`		`- return creationTimestamp.compareTo(tableSchemaAndTimestamp.updateTimeStamp()) < 0`
	`416`	`+ return creationTimestamp < tableSchemaAndTimestamp.updateTimeStamp()`
`418`	`417`	`? tableSchemaAndTimestamp.updatedSchema()`
`419`	`418`	`: null;`
`420`	`419`	`}`
`421`	`420`
`422`		`- Instant getCreationTimestamp() {`
	`421`	`+ long getCreationTimestamp() {`
`423`	`422`	`return creationTimestamp;`
`424`	`423`	`}`
`425`	`424`