chore: use an own executor for session maintenance

The session pool uses the gRPC transport executor for session maintenance.
This executor should not be used by any other tasks than internal gRPC
tasks. This change therefore creates simple executors for session
maintenance and session creation.

Updates GoogleCloudPlatform/pgadapter#2422

Author: olavloite

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

@@ -20,8 +20,8 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.pathtemplate.PathTemplate;
-import com.google.cloud.grpc.GrpcTransportOptions.ExecutorFactory;
 import com.google.cloud.spanner.spi.v1.SpannerRpc;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
@@ -30,6 +30,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import javax.annotation.concurrent.GuardedBy;
 
 /** Client for creating single sessions and batches of sessions. */
@@ -167,26 +169,37 @@ interface SessionConsumer {
   }
 
   private final SpannerImpl spanner;
-  private final ExecutorFactory<ScheduledExecutorService> executorFactory;
   private final ScheduledExecutorService executor;
   private final DatabaseId db;
 
   @GuardedBy("this")
   private volatile long sessionChannelCounter;
 
-  SessionClient(
-      SpannerImpl spanner,
-      DatabaseId db,
-      ExecutorFactory<ScheduledExecutorService> executorFactory) {
+  SessionClient(SpannerImpl spanner, DatabaseId db) {
     this.spanner = spanner;
     this.db = db;
-    this.executorFactory = executorFactory;
-    this.executor = executorFactory.get();
+    this.executor = createExecutor(spanner);
+  }
+
+  private static ScheduledThreadPoolExecutor createExecutor(Spanner spanner) {
+    ScheduledThreadPoolExecutor executor =
+        new ScheduledThreadPoolExecutor(
+            spanner.getOptions().getNumChannels(),
+            ThreadFactoryUtil.createVirtualOrPlatformDaemonThreadFactory("session-client", true));
+    executor.setKeepAliveTime(5L, TimeUnit.SECONDS);
+    executor.allowCoreThreadTimeOut(true);
+
+    return executor;
   }
 
   @Override
   public void close() {
-    executorFactory.release(executor);
+    this.executor.shutdown();
+  }
+
+  @VisibleForTesting
+  ScheduledExecutorService getExecutor() {
+    return this.executor;
   }
 
   SpannerImpl getSpanner() {

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

@@ -53,8 +53,6 @@
 import com.google.api.gax.rpc.ServerStream;
 import com.google.cloud.Timestamp;
 import com.google.cloud.Tuple;
-import com.google.cloud.grpc.GrpcTransportOptions;
-import com.google.cloud.grpc.GrpcTransportOptions.ExecutorFactory;
 import com.google.cloud.spanner.Options.QueryOption;
 import com.google.cloud.spanner.Options.ReadOption;
 import com.google.cloud.spanner.Options.TransactionOption;
@@ -102,6 +100,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -2263,7 +2262,6 @@ enum Position {
   private final SessionClient sessionClient;
   private final int numChannels;
   private final ScheduledExecutorService executor;
-  private final ExecutorFactory<ScheduledExecutorService> executorFactory;
 
   final PoolMaintainer poolMaintainer;
   private final Clock clock;
@@ -2369,7 +2367,6 @@ static SessionPool createPool(
     return createPool(
         sessionPoolOptions,
         spannerOptions.getDatabaseRole(),
-        ((GrpcTransportOptions) spannerOptions.getTransportOptions()).getExecutorFactory(),
         sessionClient,
         poolMaintainerClock == null ? new Clock() : poolMaintainerClock,
         Position.RANDOM,
@@ -2384,23 +2381,15 @@ static SessionPool createPool(
 
   static SessionPool createPool(
       SessionPoolOptions poolOptions,
-      ExecutorFactory<ScheduledExecutorService> executorFactory,
       SessionClient sessionClient,
       TraceWrapper tracer,
       OpenTelemetry openTelemetry) {
     return createPool(
-        poolOptions,
-        executorFactory,
-        sessionClient,
-        new Clock(),
-        Position.RANDOM,
-        tracer,
-        openTelemetry);
+        poolOptions, sessionClient, new Clock(), Position.RANDOM, tracer, openTelemetry);
   }
 
   static SessionPool createPool(
       SessionPoolOptions poolOptions,
-      ExecutorFactory<ScheduledExecutorService> executorFactory,
       SessionClient sessionClient,
       Clock clock,
       Position initialReleasePosition,
@@ -2409,7 +2398,6 @@ static SessionPool createPool(
     return createPool(
         poolOptions,
         null,
-        executorFactory,
         sessionClient,
         clock,
         initialReleasePosition,
@@ -2425,7 +2413,6 @@ static SessionPool createPool(
   static SessionPool createPool(
       SessionPoolOptions poolOptions,
       String databaseRole,
-      ExecutorFactory<ScheduledExecutorService> executorFactory,
       SessionClient sessionClient,
       Clock clock,
       Position initialReleasePosition,
@@ -2440,8 +2427,6 @@ static SessionPool createPool(
         new SessionPool(
             poolOptions,
             databaseRole,
-            executorFactory,
-            executorFactory.get(),
             sessionClient,
             clock,
             initialReleasePosition,
@@ -2459,8 +2444,6 @@ static SessionPool createPool(
   private SessionPool(
       SessionPoolOptions options,
       String databaseRole,
-      ExecutorFactory<ScheduledExecutorService> executorFactory,
-      ScheduledExecutorService executor,
       SessionClient sessionClient,
       Clock clock,
       Position initialReleasePosition,
@@ -2473,8 +2456,11 @@ private SessionPool(
       AtomicLong numMultiplexedSessionsReleased) {
     this.options = options;
     this.databaseRole = databaseRole;
-    this.executorFactory = executorFactory;
-    this.executor = executor;
+    this.executor =
+        Executors.newScheduledThreadPool(
+            1,
+            ThreadFactoryUtil.createVirtualOrPlatformDaemonThreadFactory(
+                "session-pool-maintainer", true));
     this.sessionClient = sessionClient;
     this.numChannels = sessionClient.getSpanner().getOptions().getNumChannels();
     this.clock = clock;
@@ -3064,6 +3050,7 @@ ListenableFuture<Void> closeAsync(ClosedException closedException) {
         pendingClosure += 1; // For pool maintenance thread
         poolMaintainer.close();
       }
+      executor.shutdown();
 
       sessions.clear();
       for (PooledSessionFuture session : checkedOutSessions) {
@@ -3097,7 +3084,6 @@ ListenableFuture<Void> closeAsync(ClosedException closedException) {
       }
     }
 
-    retFuture.addListener(() -> executorFactory.release(executor), MoreExecutors.directExecutor());
     return retFuture;
   }
 

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

@@ -22,7 +22,6 @@
 import com.google.cloud.BaseService;
 import com.google.cloud.PageImpl;
 import com.google.cloud.PageImpl.NextPageFetcher;
-import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.spanner.SessionClient.SessionId;
 import com.google.cloud.spanner.SpannerOptions.CloseableExecutorProvider;
 import com.google.cloud.spanner.admin.database.v1.stub.DatabaseAdminStubSettings;
@@ -199,11 +198,7 @@ SessionClient getSessionClient(DatabaseId db) {
       if (sessionClients.containsKey(db)) {
         return sessionClients.get(db);
       } else {
-        SessionClient client =
-            new SessionClient(
-                this,
-                db,
-                ((GrpcTransportOptions) getOptions().getTransportOptions()).getExecutorFactory());
+        SessionClient client = new SessionClient(this, db);
         sessionClients.put(db, client);
         return client;
       }

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

@@ -18,15 +18,12 @@
 
 import static com.google.common.truth.Truth.assertThat;
 
-import com.google.cloud.grpc.GrpcTransportOptions.ExecutorFactory;
 import com.google.cloud.spanner.SessionPool.PooledSessionFuture;
 import io.opencensus.trace.Tracing;
 import io.opentelemetry.api.OpenTelemetry;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -83,18 +80,6 @@ public void setUp() {
     pool =
         SessionPool.createPool(
             options,
-            new ExecutorFactory<ScheduledExecutorService>() {
-
-              @Override
-              public void release(ScheduledExecutorService executor) {
-                executor.shutdown();
-              }
-
-              @Override
-              public ScheduledExecutorService get() {
-                return new ScheduledThreadPoolExecutor(2);
-              }
-            },
             ((SpannerImpl) env.getTestHelper().getClient()).getSessionClient(db.getId()),
             new TraceWrapper(Tracing.getTracer(), OpenTelemetry.noop().getTracer(""), false),
             OpenTelemetry.noop());