Merge branch 'master' into release-please/branches/master

miraleung · web-flow · commit 44a018be8ccb · 2021-04-26T13:33:11.000-07:00
diff --git a/gax/src/main/java/com/google/api/gax/batching/DynamicFlowControlSettings.java b/gax/src/main/java/com/google/api/gax/batching/DynamicFlowControlSettings.java
@@ -115,11 +115,14 @@ public DynamicFlowControlSettings build() {
     }
 
     private void verifyElementCountSettings(DynamicFlowControlSettings settings) {
-      boolean isEnabled =
-          settings.getInitialOutstandingElementCount() != null
-              || settings.getMinOutstandingElementCount() != null
-              || settings.getMaxOutstandingElementCount() != null;
-      if (!isEnabled) {
+      // If LimitExceededBehavior is Ignore, dynamic flow control is disabled, there's no need to
+      // check element count limit settings
+      if (settings.getLimitExceededBehavior() == LimitExceededBehavior.Ignore) {
+        return;
+      }
+      if (settings.getInitialOutstandingElementCount() == null
+          && settings.getMinOutstandingElementCount() == null
+          && settings.getMaxOutstandingElementCount() == null) {
         return;
       }
       Preconditions.checkState(
@@ -141,11 +144,14 @@ private void verifyElementCountSettings(DynamicFlowControlSettings settings) {
     }
 
     private void verifyRequestBytesSettings(DynamicFlowControlSettings settings) {
-      boolean isEnabled =
-          settings.getInitialOutstandingRequestBytes() != null
-              || settings.getMinOutstandingRequestBytes() != null
-              || settings.getMaxOutstandingRequestBytes() != null;
-      if (!isEnabled) {
+      // If LimitExceededBehavior is Ignore, dynamic flow control is disabled, there's no need to
+      // check request bytes limit settings
+      if (settings.getLimitExceededBehavior() == LimitExceededBehavior.Ignore) {
+        return;
+      }
+      if (settings.getInitialOutstandingRequestBytes() == null
+          && settings.getMinOutstandingRequestBytes() == null
+          && settings.getMaxOutstandingRequestBytes() == null) {
         return;
       }
       Preconditions.checkState(
diff --git a/gax/src/main/java/com/google/api/gax/rpc/Watchdog.java b/gax/src/main/java/com/google/api/gax/rpc/Watchdog.java
@@ -183,7 +183,7 @@ class WatchdogStream<ResponseT> extends StateCheckingResponseObserver<ResponseT>
     private boolean autoAutoFlowControl = true;
 
     private final ResponseObserver<ResponseT> outerResponseObserver;
-    private StreamController innerController;
+    private volatile StreamController innerController;
 
     @GuardedBy("lock")
     private State state = State.IDLE;
@@ -296,6 +296,12 @@ public void onCompleteImpl() {
      * @return True if the stream was canceled.
      */
     boolean cancelIfStale() {
+      // If the stream hasn't started yet, innerController will be null. Skip the check this time
+      // and return false so the stream is still watched.
+      if (innerController == null) {
+        return false;
+      }
+
       Throwable myError = null;
 
       synchronized (lock) {
diff --git a/gax/src/test/java/com/google/api/gax/batching/DynamicFlowControlSettingsTest.java b/gax/src/test/java/com/google/api/gax/batching/DynamicFlowControlSettingsTest.java
@@ -55,6 +55,22 @@ public void testEmptyBuilder() {
     assertEquals(LimitExceededBehavior.Block, settings.getLimitExceededBehavior());
   }
 
+  @Test
+  public void testPartialSettingsIgnored() {
+    // If behavior is ignore, build shouldn't throw exceptions even when only one of the bytes or
+    // element limits is set
+    DynamicFlowControlSettings.Builder builder =
+        DynamicFlowControlSettings.newBuilder()
+            .setLimitExceededBehavior(LimitExceededBehavior.Ignore)
+            .setMaxOutstandingElementCount(1L);
+    builder.build();
+    builder =
+        DynamicFlowControlSettings.newBuilder()
+            .setLimitExceededBehavior(LimitExceededBehavior.Ignore)
+            .setMinOutstandingRequestBytes(1L);
+    builder.build();
+  }
+
   @Test
   public void testBuilder() {
     DynamicFlowControlSettings.Builder builder =
diff --git a/gax/src/test/java/com/google/api/gax/rpc/WatchdogTest.java b/gax/src/test/java/com/google/api/gax/rpc/WatchdogTest.java
@@ -128,6 +128,31 @@ public void testIdleTimeout() throws InterruptedException {
     assertThat(actualError).isInstanceOf(WatchdogTimeoutException.class);
   }
 
+  @Test
+  public void testTimedOutBeforeStart() throws InterruptedException {
+    MockServerStreamingCallable<String, String> callable1 = new MockServerStreamingCallable<>();
+    AccumulatingObserver<String> downstreamObserver1 = new AccumulatingObserver<>();
+    ResponseObserver observer = watchdog.watch(downstreamObserver1, waitTime, idleTime);
+    clock.incrementNanoTime(idleTime.toNanos() + 1);
+    // This should not remove callable1 from watched list
+    watchdog.run();
+    assertThat(downstreamObserver1.done.isDone()).isFalse();
+
+    callable1.call("request", observer);
+    // This should cancel callable1
+    watchdog.run();
+    MockServerStreamingCall<String, String> call1 = callable1.popLastCall();
+    assertThat(call1.getController().isCancelled()).isTrue();
+    call1.getController().getObserver().onError(new CancellationException("User cancelled"));
+    Throwable error = null;
+    try {
+      downstreamObserver1.done.get();
+    } catch (ExecutionException t) {
+      error = t.getCause();
+    }
+    assertThat(error).isInstanceOf(WatchdogTimeoutException.class);
+  }
+
   @Test
   public void testMultiple() throws Exception {
     // Start stream1
diff --git a/samples/pom.xml b/samples/pom.xml
@@ -25,7 +25,7 @@
     <dependency>
       <groupId>com.google.auto.value</groupId>
       <artifactId>auto-value</artifactId>
-      <version>1.7.5</version>
+      <version>1.8</version>
     </dependency>
     <dependency>
       <groupId>junit</groupId>
