Add logs to find the issue at root

googleapis · Jan 6, 2025 · 7a96270 · 7a96270
1 parent 21733d5
commit 7a96270
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 9 deletions.
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AsyncResultSetImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/AsyncResultSetImpl.java
@@ -470,14 +470,14 @@ public void run() {
         // Non-streaming result sets do not trigger this callback, and for those result sets, we
         // need to eagerly start the ProduceRowsRunnable.
         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 (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();
           }
@@ -492,6 +492,7 @@ public void run() {
   /** Sets the callback for this {@link AsyncResultSet}. */
   @Override
   public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
+    System.out.println("Inside setCallback");
     synchronized (monitor) {
       Preconditions.checkState(!closed, "This AsyncResultSet has been closed");
       Preconditions.checkState(
@@ -660,7 +661,7 @@ public void onStreamMessage(PartialResultSet partialResultSet, boolean bufferIsF
           !partialResultSet.getResumeToken().isEmpty()
               || bufferIsFull
               || partialResultSet == GrpcStreamIterator.END_OF_STREAM;
-      if (startJobThread || state != State.STREAMING_IN_PROGRESS) {
+      if (startJobThread || state != State.STREAMING_INITIALIZED) {
         initiateProduceRows();
       }
     }

diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/MockSpannerServiceImpl.java
@@ -651,6 +651,7 @@ private ByteString generateTransactionName(String session) {
       counter = new AtomicLong();
       transactionCounters.put(session, counter);
     }
+    System.out.printf("Generating session using Session ID %s\n", session);
     return ByteString.copyFromUtf8(
         String.format("%s/transactions/%d", session, counter.incrementAndGet()));
   }
@@ -1999,6 +2000,7 @@ private void ensureMostRecentTransaction(Session session, ByteString transaction
       if (index > -1) {
         long id = Long.parseLong(transactionId.toStringUtf8().substring(index + 1));
         if (id != counter.get()) {
+          System.out.printf("Session ID %s TransactionId %s\n", session.getName(), transactionId.toStringUtf8());
           throw Status.FAILED_PRECONDITION
               .withDescription(
                   String.format(

diff --git a/...e-cloud-spanner/src/test/java/com/google/cloud/spanner/RetryOnInvalidatedSessionTest.java b/...e-cloud-spanner/src/test/java/com/google/cloud/spanner/RetryOnInvalidatedSessionTest.java
@@ -1434,6 +1434,8 @@ private void asyncRunner_withReadFunction(
                   runner.runAsync(
                       txn -> {
                         AsyncResultSet rs = readFunction.apply(txn);
+                        System.out.println(rs);
+                        System.out.println("Creating a new AsyncResultSet");
                         ApiFuture<Void> fut =
                             rs.setCallback(
                                 queryExecutor,
@@ -1450,6 +1452,7 @@ private void asyncRunner_withReadFunction(
                                     }
                                   }
                                 });
+                        System.out.println("After setCallback");
                         return ApiFutures.transform(
                             fut, input -> counter.get(), MoreExecutors.directExecutor());
                       },
@@ -1573,6 +1576,8 @@ private void asyncTransactionManager_readAsync(
               context.then(
                   (transaction, ignored) -> {
                     AsyncResultSet rs = fn.apply(transaction);
+                    System.out.println(rs);
+                    System.out.println("1. Creating a new AsyncResultSet");
                     ApiFuture<Void> fut =
                         rs.setCallback(
                             queryExecutor,