Add logs to find the issue at root

googleapis · Jan 6, 2025 · 4072c7c · 4072c7c
1 parent 21733d5
commit 4072c7c
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 14 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
@@ -469,19 +469,19 @@ 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.
-        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();
-          }
+        //        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);
         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(
@@ -510,7 +511,7 @@ public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
 
   private void initiateProduceRows() {
     synchronized (monitor) {
-      if (this.state == State.STREAMING_IN_PROGRESS) {
+      if (this.state == State.STREAMING_INITIALIZED) {
         this.state = State.RUNNING;
       }
       produceRowsInitiated = true;
@@ -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,8 @@ 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
@@ -60,7 +60,11 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameter;
@@ -75,6 +79,24 @@ public Long apply(StructReader input) {
     }
   }
 
+  @Rule
+  public TestWatcher testWatcher = new TestWatcher() {
+    @Override
+    protected void succeeded(Description description) {
+      System.out.println("Succeeded test: " + description.getMethodName());
+    }
+
+    @Override
+    protected void failed(Throwable e, Description description) {
+      System.out.println("Failed test: " + description.getMethodName());
+    }
+
+    @Override
+    protected void starting(Description description) {
+      System.out.println("Starting test: " + description.getMethodName());
+    }
+  };
+
   private static final ToLongTransformer TO_LONG = new ToLongTransformer();
 
   @Parameter(0)
@@ -1434,6 +1456,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 +1474,7 @@ private void asyncRunner_withReadFunction(
                                     }
                                   }
                                 });
+                        System.out.println("After setCallback");
                         return ApiFutures.transform(
                             fut, input -> counter.get(), MoreExecutors.directExecutor());
                       },
@@ -1573,6 +1598,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,