fix: fix a NullPtr when user closes a writer without connection being ever established (#1454)

Author: yirutang

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Exceptions.java

@@ -27,6 +27,11 @@
 
 /** Exceptions for Storage Client Libraries. */
 public final class Exceptions {
+  public static class WriterClosedException extends Exception {
+    public WriterClosedException(String streamName) {
+      super("Writer closed on: " + streamName);
+    }
+  }
   /** Main Storage Exception. Might contain map of streams to errors for that stream. */
   public static class StorageException extends RuntimeException {
 

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java

@@ -502,11 +502,13 @@ private AppendRowsRequest prepareRequestBasedOnPosition(
   }
 
   private void cleanupInflightRequests() {
-    Throwable finalStatus;
+    Throwable finalStatus = new Exceptions.WriterClosedException(streamName);
     Deque<AppendRequestAndResponse> localQueue = new LinkedList<AppendRequestAndResponse>();
     this.lock.lock();
     try {
-      finalStatus = this.connectionFinalStatus;
+      if (this.connectionFinalStatus != null) {
+        finalStatus = this.connectionFinalStatus;
+      }
       while (!this.inflightRequestQueue.isEmpty()) {
         localQueue.addLast(pollInflightRequestQueue());
       }

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1/StreamWriterTest.java

@@ -638,4 +638,12 @@ public void testRetryAfterAllRecordsInflight() throws Exception {
       assertEquals(1, appendFuture2.get().getAppendResult().getOffset().getValue());
     }
   }
+
+  @Test
+  public void testWriterClosedStream() throws Exception {
+    try (StreamWriter writer = getTestStreamWriter()) {
+      // Writer is closed without any traffic.
+      TimeUnit.SECONDS.sleep(1);
+    }
+  }
 }