fix(spanner): unflake unit tests

Author: sakthivelmanii

google-cloud-spanner/src/main/java/com/google/cloud/spanner/AsyncResultSetImpl.java

@@ -52,6 +52,7 @@ class AsyncResultSetImpl extends ForwardingStructReader
   private enum State {
     INITIALIZED,
     STREAMING_INITIALIZED,
+    STREAMING_IN_PROGRESS,
     /** SYNC indicates that the {@link ResultSet} is used in sync pattern. */
     SYNC,
     CONSUMING,
@@ -468,8 +469,18 @@ public void run() {
         // Those result sets will trigger initiateProduceRows() when the first results are received.
         // Non-streaming result sets do not trigger this callback, and for those result sets, we
         // need to eagerly start the ProduceRowsRunnable.
-        if (!initiateStreaming(AsyncResultSetImpl.this)) {
-          initiateProduceRows();
+        synchronized (monitor) {
+          if (state == State.STREAMING_IN_PROGRESS
+              || state == State.RUNNING
+              || state == State.CONSUMING) {
+            return;
+          }
+          if (state == State.STREAMING_INITIALIZED) {
+            state = State.STREAMING_IN_PROGRESS;
+          }
+          if (!initiateStreaming(AsyncResultSetImpl.this)) {
+            initiateProduceRows();
+          }
         }
       } catch (Throwable exception) {
         executionException = SpannerExceptionFactory.asSpannerException(exception);
@@ -499,7 +510,7 @@ public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
 
   private void initiateProduceRows() {
     synchronized (monitor) {
-      if (this.state == State.STREAMING_INITIALIZED) {
+      if (this.state == State.STREAMING_IN_PROGRESS) {
         this.state = State.RUNNING;
       }
       produceRowsInitiated = true;
@@ -649,7 +660,7 @@ public void onStreamMessage(PartialResultSet partialResultSet, boolean bufferIsF
           !partialResultSet.getResumeToken().isEmpty()
               || bufferIsFull
               || partialResultSet == GrpcStreamIterator.END_OF_STREAM;
-      if (startJobThread || state != State.STREAMING_INITIALIZED) {
+      if (startJobThread || state != State.STREAMING_IN_PROGRESS) {
         initiateProduceRows();
       }
     }