fix: fix watchdog NPE red herring

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

@@ -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) {

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

@@ -128,6 +128,20 @@ public void testIdleTimeout() throws InterruptedException {
     assertThat(actualError).isInstanceOf(WatchdogTimeoutException.class);
   }
 
+  @Test
+  public void testTimedOutBeforeStart() {
+    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();
+    callable1.call("request", observer);
+    // This should cancel callable1
+    watchdog.run();
+    assertThat(callable1.popLastCall().getController().isCancelled()).isTrue();
+  }
+
   @Test
   public void testMultiple() throws Exception {
     // Start stream1