Merge branch 'main' into remove-session-pool

Author: olavloite

.github/sync-repo-settings.yaml

@@ -18,8 +18,6 @@ branchProtectionRules:
       - checkstyle
       - compile (8)
       - compile (11)
-      - units-with-multiplexed-session (8)
-      - units-with-multiplexed-session (11)
       - unmanaged_dependency_check
       - library_generation
   - pattern: 3.3.x

.github/workflows/ci.yaml

@@ -36,41 +36,6 @@ jobs:
     - run: .kokoro/build.sh
       env:
         JOB_TYPE: test
-  units-with-multiplexed-session:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        java: [ 11, 17, 21 ]
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/setup-java@v3
-      with:
-        distribution: temurin
-        java-version: ${{matrix.java}}
-    - run: java -version
-    - run: .kokoro/build.sh
-      env:
-        JOB_TYPE: test
-        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS: true
-        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_FOR_RW: true
-  units-with-regular-session:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        java: [ 11, 17, 21 ]
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/setup-java@v3
-      with:
-        distribution: temurin
-        java-version: ${{matrix.java}}
-    - run: java -version
-    - run: .kokoro/build.sh
-      env:
-        JOB_TYPE: test
-        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: false
   units-java8:
     # Building using Java 17 and run the tests with Java 8 runtime
     name: "units (8)"
@@ -90,48 +55,6 @@ jobs:
     - run: .kokoro/build.sh
       env:
         JOB_TYPE: test
-  units-with-multiplexed-session8:
-    # Building using Java 17 and run the tests with Java 8 runtime
-    name: "units-with-multiplexed-session (8)"
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-java@v3
-        with:
-          java-version: 8
-          distribution: temurin
-      - run: echo "SUREFIRE_JVM_OPT=-Djvm=${JAVA_HOME}/bin/java" >> $GITHUB_ENV
-        shell: bash
-      - uses: actions/setup-java@v3
-        with:
-          java-version: 17
-          distribution: temurin
-      - run: .kokoro/build.sh
-        env:
-          JOB_TYPE: test
-          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: true
-          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_PARTITIONED_OPS: true
-          GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS_FOR_RW: true
-  units-with-regular-session8:
-    # Building using Java 17 and run the tests with Java 8 runtime
-    name: "units-with-regular-session (8)"
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/setup-java@v3
-      with:
-        java-version: 8
-        distribution: temurin
-    - run: echo "SUREFIRE_JVM_OPT=-Djvm=${JAVA_HOME}/bin/java" >> $GITHUB_ENV
-      shell: bash
-    - uses: actions/setup-java@v3
-      with:
-        java-version: 17
-        distribution: temurin
-    - run: .kokoro/build.sh
-      env:
-        JOB_TYPE: test
-        GOOGLE_CLOUD_SPANNER_MULTIPLEXED_SESSIONS: false
   windows:
     runs-on: windows-latest
     steps:

.github/workflows/integration-tests-against-emulator-with-regular-session.yaml

@@ -104,21 +104,6 @@ integration)
       verify
     RETURN_CODE=$?
     ;;
-integration-regular-sessions)
-    mvn -B ${INTEGRATION_TEST_ARGS} \
-      -ntp \
-      -Penable-integration-tests \
-      -Djava.net.preferIPv4Stack=true \
-      -DtrimStackTrace=false \
-      -Dclirr.skip=true \
-      -Denforcer.skip=true \
-      -Dmaven.main.skip=true \
-      -Dspanner.gce.config.project_id=gcloud-devel \
-      -Dspanner.testenv.instance=projects/gcloud-devel/instances/java-client-integration-tests-regular-sessions \
-      -fae \
-      verify
-    RETURN_CODE=$?
-    ;;
 integration-directpath-enabled)
     mvn -B ${INTEGRATION_TEST_ARGS} \
       -ntp \

.kokoro/build.sh

@@ -818,7 +818,11 @@ CloseableIterator<PartialResultSet> startStream(
               @Nullable ByteString resumeToken,
               AsyncResultSet.StreamMessageListener streamListener) {
             GrpcStreamIterator stream =
-                new GrpcStreamIterator(statement, prefetchChunks, cancelQueryWhenClientIsClosed);
+                new GrpcStreamIterator(
+                    statement,
+                    request.getLastStatement(),
+                    prefetchChunks,
+                    cancelQueryWhenClientIsClosed);
             if (streamListener != null) {
               stream.registerListener(streamListener);
             }
@@ -935,7 +939,8 @@ String getTransactionTag() {
   public void onTransactionMetadata(Transaction transaction, boolean shouldIncludeId) {}
 
   @Override
-  public SpannerException onError(SpannerException e, boolean withBeginTransaction) {
+  public SpannerException onError(
+      SpannerException e, boolean withBeginTransaction, boolean lastStatement) {
     this.session.onError(e);
     return e;
   }
@@ -1009,6 +1014,8 @@ ResultSet readInternalWithOptions(
     }
     final int prefetchChunks =
         readOptions.hasPrefetchChunks() ? readOptions.prefetchChunks() : defaultPrefetchChunks;
+    final boolean lastStatement =
+        readOptions.hasLastStatement() ? readOptions.isLastStatement() : false;
     ResumableStreamIterator stream =
         new ResumableStreamIterator(
             MAX_BUFFERED_CHUNKS,
@@ -1025,7 +1032,8 @@ CloseableIterator<PartialResultSet> startStream(
               @Nullable ByteString resumeToken,
               AsyncResultSet.StreamMessageListener streamListener) {
             GrpcStreamIterator stream =
-                new GrpcStreamIterator(prefetchChunks, cancelQueryWhenClientIsClosed);
+                new GrpcStreamIterator(
+                    lastStatement, prefetchChunks, cancelQueryWhenClientIsClosed);
             if (streamListener != null) {
               stream.registerListener(streamListener);
             }

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

@@ -56,7 +56,8 @@ void onTransactionMetadata(Transaction transaction, boolean shouldIncludeId)
         throws SpannerException;
 
     /** Called when the read finishes with an error. Returns the error that should be thrown. */
-    SpannerException onError(SpannerException e, boolean withBeginTransaction);
+    SpannerException onError(
+        SpannerException e, boolean withBeginTransaction, boolean lastStatement);
 
     /** Called when the read finishes normally. */
     void onDone(boolean withBeginTransaction);
@@ -153,6 +154,8 @@ interface CloseableIterator<T> extends Iterator<T> {
 
     boolean isWithBeginTransaction();
 
+    boolean isLastStatement();
+
     /**
      * @param streamMessageListener A class object which implements StreamMessageListener
      * @return true if streaming is supported by the iterator, otherwise false

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

@@ -109,7 +109,8 @@ public boolean next() throws SpannerException {
     } catch (Throwable t) {
       throw yieldError(
           SpannerExceptionFactory.asSpannerException(t),
-          iterator.isWithBeginTransaction() && currRow == null);
+          iterator.isWithBeginTransaction() && currRow == null,
+          iterator.isLastStatement());
     }
   }
 
@@ -149,8 +150,9 @@ public Type getType() {
     return currRow.getType();
   }
 
-  private SpannerException yieldError(SpannerException e, boolean beginTransaction) {
-    SpannerException toThrow = listener.onError(e, beginTransaction);
+  private SpannerException yieldError(
+      SpannerException e, boolean beginTransaction, boolean lastStatement) {
+    SpannerException toThrow = listener.onError(e, beginTransaction, lastStatement);
     close();
     throw toThrow;
   }

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

@@ -49,20 +49,26 @@ class GrpcStreamIterator extends AbstractIterator<PartialResultSet>
 
   private SpannerRpc.StreamingCall call;
   private volatile boolean withBeginTransaction;
+  private final boolean lastStatement;
   private TimeUnit streamWaitTimeoutUnit;
   private long streamWaitTimeoutValue;
   private SpannerException error;
   private boolean done;
 
   @VisibleForTesting
-  GrpcStreamIterator(int prefetchChunks, boolean cancelQueryWhenClientIsClosed) {
-    this(null, prefetchChunks, cancelQueryWhenClientIsClosed);
+  GrpcStreamIterator(
+      boolean lastStatement, int prefetchChunks, boolean cancelQueryWhenClientIsClosed) {
+    this(null, lastStatement, prefetchChunks, cancelQueryWhenClientIsClosed);
   }
 
   @VisibleForTesting
   GrpcStreamIterator(
-      Statement statement, int prefetchChunks, boolean cancelQueryWhenClientIsClosed) {
+      Statement statement,
+      boolean lastStatement,
+      int prefetchChunks,
+      boolean cancelQueryWhenClientIsClosed) {
     this.statement = statement;
+    this.lastStatement = lastStatement;
     this.prefetchChunks = prefetchChunks;
     this.consumer = new ConsumerImpl(cancelQueryWhenClientIsClosed);
     // One extra to allow for END_OF_STREAM message.
@@ -118,6 +124,11 @@ public boolean isWithBeginTransaction() {
     return withBeginTransaction;
   }
 
+  @Override
+  public boolean isLastStatement() {
+    return lastStatement;
+  }
+
   @Override
   protected final PartialResultSet computeNext() {
     PartialResultSet next;

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

@@ -183,6 +183,10 @@ boolean isWithBeginTransaction() {
     return stream.isWithBeginTransaction();
   }
 
+  boolean isLastStatement() {
+    return stream.isLastStatement();
+  }
+
   /**
    * @param a is a mutable list and b will be concatenated into a.
    */

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

@@ -247,8 +247,7 @@ public void onSessionCreateFailure(Throwable t, int createFailureForSessionCount
                     || spannerException instanceof InstanceNotFoundException
                     || spannerException instanceof SessionNotFoundException)) {
               // This could in theory set this field more than once, but we don't want to bother
-              // with
-              // synchronizing, as it does not really matter exactly which error is set.
+              // with synchronizing, as it does not really matter exactly which error is set.
               MultiplexedSessionDatabaseClient.this.resourceNotFoundException.set(
                   (ResourceNotFoundException) spannerException);
             }