fix(spanner): add logs

Author: sakthivelmanii

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

@@ -31,6 +31,7 @@
 import com.google.spanner.v1.PartialResultSet;
 import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.ResultSetStats;
+import java.time.OffsetDateTime;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
@@ -342,6 +343,9 @@ public void run() {
       boolean stop = false;
       boolean hasNext = false;
       try {
+        System.out.printf(
+            "[%s][%s] Inside ProduceRowsRunnable thread\n",
+            OffsetDateTime.now(), Thread.currentThread().getName());
         hasNext = delegateResultSet.get().next();
       } catch (Throwable e) {
         synchronized (monitor) {
@@ -468,6 +472,9 @@ 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.
+        System.out.printf(
+            "[%s][%s] Inside InitiateStreamingRunnable thread\n",
+            OffsetDateTime.now(), Thread.currentThread().getName());
         if (!initiateStreaming(AsyncResultSetImpl.this)) {
           initiateProduceRows();
         }
@@ -486,6 +493,8 @@ public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
       Preconditions.checkState(
           this.state == State.INITIALIZED, "callback may not be set multiple times");
 
+      System.out.printf(
+          "[%s][%s] Inside callback\n", OffsetDateTime.now(), Thread.currentThread().getName());
       // Start to fetch data and buffer these.
       this.result = SettableApiFuture.create();
       this.state = State.STREAMING_INITIALIZED;
@@ -498,6 +507,9 @@ public ApiFuture<Void> setCallback(Executor exec, ReadyCallback cb) {
   }
 
   private void initiateProduceRows() {
+    System.out.printf(
+        "[%s][%s] Inside initiateProduceRows\n",
+        OffsetDateTime.now(), Thread.currentThread().getName());
     synchronized (monitor) {
       if (this.state == State.STREAMING_INITIALIZED) {
         this.state = State.RUNNING;
@@ -639,6 +651,9 @@ public Struct getCurrentRowAsStruct() {
 
   @Override
   public void onStreamMessage(PartialResultSet partialResultSet, boolean bufferIsFull) {
+    System.out.printf(
+        "[%s][%s] Inside onStreamMessage\n",
+        OffsetDateTime.now(), Thread.currentThread().getName());
     synchronized (monitor) {
       if (produceRowsInitiated) {
         return;

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

@@ -22,6 +22,7 @@
 import com.google.common.base.Suppliers;
 import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.ResultSetStats;
+import java.time.OffsetDateTime;
 
 /** Forwarding implementation of ResultSet that forwards all calls to a delegate. */
 public class ForwardingResultSet extends ForwardingStructReader
@@ -108,6 +109,9 @@ public ResultSetMetadata getMetadata() {
   @Override
   @InternalApi
   public boolean initiateStreaming(AsyncResultSet.StreamMessageListener streamMessageListener) {
+    System.out.printf(
+        "[%s][%s] Inside initiateStreaming ForwardingResultSet\n",
+        OffsetDateTime.now(), Thread.currentThread().getName());
     return StreamingUtil.initiateStreaming(delegate.get(), streamMessageListener);
   }
 }

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

@@ -25,6 +25,7 @@
 import com.google.spanner.v1.PartialResultSet;
 import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.ResultSetStats;
+import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -80,6 +81,9 @@ public boolean next() throws SpannerException {
     }
     try {
       if (currRow == null) {
+        System.out.printf(
+            "[%s][%s] Inside GrpcResultSet next()\n",
+            OffsetDateTime.now(), Thread.currentThread().getName());
         metadata = iterator.getMetadata();
         if (metadata.hasTransaction()) {
           listener.onTransactionMetadata(
@@ -128,6 +132,9 @@ public ResultSetMetadata getMetadata() {
   @Override
   @InternalApi
   public boolean initiateStreaming(AsyncResultSet.StreamMessageListener streamMessageListener) {
+    System.out.printf(
+        "[%s][%s] Inside initiateStreaming GrpcResultSet\n",
+        OffsetDateTime.now(), Thread.currentThread().getName());
     return iterator.initiateStreaming(streamMessageListener);
   }
 

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

@@ -37,6 +37,7 @@
 import io.grpc.Context;
 import io.opentelemetry.api.common.Attributes;
 import java.io.IOException;
+import java.time.OffsetDateTime;
 import java.util.LinkedList;
 import java.util.Objects;
 import java.util.Set;
@@ -65,7 +66,7 @@ abstract class ResumableStreamIterator extends AbstractIterator<PartialResultSet
   private static final Logger logger = Logger.getLogger(ResumableStreamIterator.class.getName());
   private BackOff backOff;
   private final LinkedList<PartialResultSet> buffer = new LinkedList<>();
-  private final Object monitor = new Object();
+  //  private final Object monitor = new Object();
   private final int maxBufferSize;
   private final ISpan span;
   private final TraceWrapper tracer;
@@ -318,19 +319,25 @@ && prepareIteratorForRetryOnDifferentGrpcChannel()) {
   }
 
   private void startGrpcStreaming() {
-    synchronized (monitor) {
-      if (stream == null) {
-        span.addAnnotation(
-            "Starting/Resuming stream",
-            "ResumeToken",
-            resumeToken == null ? "null" : resumeToken.toStringUtf8());
-        try (IScope scope = tracer.withSpan(span)) {
-          // When start a new stream set the Span as current to make the gRPC Span a child of
-          // this Span.
-          stream = checkNotNull(startStream(resumeToken, streamMessageListener));
-        }
+    //    synchronized (monitor) {
+    System.out.printf(
+        "[%s][%s] Inside startGrpcStreaming\n",
+        OffsetDateTime.now(), Thread.currentThread().getName());
+    if (stream == null) {
+      span.addAnnotation(
+          "Starting/Resuming stream",
+          "ResumeToken",
+          resumeToken == null ? "null" : resumeToken.toStringUtf8());
+      try (IScope scope = tracer.withSpan(span)) {
+        // When start a new stream set the Span as current to make the gRPC Span a child of
+        // this Span.
+        System.out.printf(
+            "[%s][%s] Inside creating stream\n",
+            OffsetDateTime.now(), Thread.currentThread().getName());
+        stream = checkNotNull(startStream(resumeToken, streamMessageListener));
       }
     }
+    //    }
   }
 
   boolean isRetryable(SpannerException spannerException) {

google-cloud-spanner/src/test/java/com/google/cloud/spanner/RetryOnInvalidatedSessionTest.java

@@ -46,6 +46,7 @@
 import io.grpc.Server;
 import io.grpc.inprocess.InProcessServerBuilder;
 import java.io.IOException;
+import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -60,7 +61,10 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
+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,30 @@ public Long apply(StructReader input) {
     }
   }
 
+  @Rule
+  public TestWatcher testWatcher = new TestWatcher() {
+    @Override
+    protected void succeeded(Description description) {
+      System.out.printf(
+          "[%s][%s] Succeeded test %s\n",
+          OffsetDateTime.now(), Thread.currentThread().getName(), description.getMethodName());
+    }
+
+    @Override
+    protected void failed(Throwable e, Description description) {
+      System.out.printf(
+          "[%s][%s] Failed test %s\n",
+          OffsetDateTime.now(), Thread.currentThread().getName(), description.getMethodName());
+    }
+
+    @Override
+    protected void starting(Description description) {
+      System.out.printf(
+          "[%s][%s] Starting test %s\n",
+          OffsetDateTime.now(), Thread.currentThread().getName(), description.getMethodName());
+    }
+  };
+
   private static final ToLongTransformer TO_LONG = new ToLongTransformer();
 
   @Parameter(0)