fix: Fix watchdog to start with WAITING state

gax-java/gax/src/main/java/com/google/api/gax/rpc/Watchdog.java

@@ -194,7 +194,7 @@ class WatchdogStream<ResponseT> extends StateCheckingResponseObserver<ResponseT>
     private volatile StreamController innerController;
 
     @GuardedBy("lock")
-    private State state = State.IDLE;
+    private State state = State.WAITING;
 
     @GuardedBy("lock")
     private int pendingCount = 0;
@@ -220,6 +220,16 @@ public void onStartImpl(StreamController controller) {
             public void disableAutoInboundFlowControl() {
               Preconditions.checkState(
                   !hasStarted, "Can't disable automatic flow control after the stream has started");
+
+              synchronized (lock) {
+                // WatchdogStream is created with waiting state. If auto flow control is disabled,
+                // set the state to be IDLE, which will be updated to WAITING when onRequest() is
+                // called.
+                if (state == State.WAITING) {
+                  state = State.IDLE;
+                }
+              }
+
               autoAutoFlowControl = false;
               innerController.disableAutoInboundFlowControl();
             }

gax-java/gax/src/test/java/com/google/api/gax/rpc/WatchdogTest.java

@@ -154,6 +154,31 @@ public void testTimedOutBeforeStart() throws InterruptedException {
     assertThat(error).isInstanceOf(WatchdogTimeoutException.class);
   }
 
+  @Test
+  public void testTimedOutBeforeResponse() throws InterruptedException {
+    MockServerStreamingCallable<String, String> autoFlowControlCallable =
+        new MockServerStreamingCallable<>();
+    AutoFlowControlObserver<String> downstreamObserver = new AutoFlowControlObserver<>();
+
+    autoFlowControlCallable.call("request", watchdog.watch(downstreamObserver, waitTime, idleTime));
+    MockServerStreamingCall<String, String> call1 = autoFlowControlCallable.popLastCall();
+
+    clock.incrementNanoTime(idleTime.toNanos() + 1);
+    watchdog.run();
+    assertThat(downstreamObserver.done.isDone()).isFalse();
+    assertThat(call1.getController().isCancelled()).isTrue();
+    call1.getController().getObserver().onError(new CancellationException("cancelled"));
+
+    Throwable actualError = null;
+    try {
+      downstreamObserver.done.get();
+    } catch (ExecutionException e) {
+      actualError = e.getCause();
+    }
+    assertThat(actualError).isInstanceOf(WatchdogTimeoutException.class);
+    assertThat(actualError.getMessage()).contains("waiting for next response");
+  }
+
   @Test
   public void testMultiple() throws Exception {
     // Start stream1
@@ -310,4 +335,30 @@ public void onComplete() {
       done.set(null);
     }
   }
+
+  static class AutoFlowControlObserver<T> implements ResponseObserver<T> {
+    SettableApiFuture<StreamController> controller = SettableApiFuture.create();
+    Queue<T> responses = Queues.newLinkedBlockingDeque();
+    SettableApiFuture<Void> done = SettableApiFuture.create();
+
+    @Override
+    public void onStart(StreamController controller) {
+      this.controller.set(controller);
+    }
+
+    @Override
+    public void onResponse(T response) {
+      responses.add(response);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+      done.setException(t);
+    }
+
+    @Override
+    public void onComplete() {
+      done.set(null);
+    }
+  }
 }