feat: allow configuring batch timeout property in session client (#7965)

Creating NewClientWithConfig supports BatchTimeout property.

spanner/client.go

@@ -176,6 +176,9 @@ type ClientConfig struct {
 	//
 	//  Default: identity
 	Compression string
+
+	// BatchTimeout specifies the timeout for a batch of sessions managed sessionClient.
+	BatchTimeout time.Duration
 }
 
 func contextWithOutgoingMetadata(ctx context.Context, md metadata.MD, disableRouteToLeader bool) context.Context {
@@ -252,12 +255,17 @@ func NewClientWithConfig(ctx context.Context, database string, config ClientConf
 	if config.incStep == 0 {
 		config.incStep = DefaultSessionPoolConfig.incStep
 	}
+	if config.BatchTimeout == 0 {
+		config.BatchTimeout = time.Minute
+	}
+
 	md := metadata.Pairs(resourcePrefixHeader, database)
 	if config.Compression == gzip.Name {
 		md.Append(requestsCompressionHeader, gzip.Name)
 	}
 	// Create a session client.
-	sc := newSessionClient(pool, database, config.UserAgent, sessionLabels, config.DatabaseRole, config.DisableRouteToLeader, md, config.Logger, config.CallOptions)
+	sc := newSessionClient(pool, database, config.UserAgent, sessionLabels, config.DatabaseRole, config.DisableRouteToLeader, md, config.BatchTimeout, config.Logger, config.CallOptions)
+
 	// Create a session pool.
 	config.SessionPoolConfig.sessionLabels = sessionLabels
 	sp, err := newSessionPool(sc, config.SessionPoolConfig)

spanner/client_test.go

@@ -2678,6 +2678,38 @@ func TestClient_WithGRPCConnectionPoolAndNumChannels_Misconfigured(t *testing.T)
 	}
 }
 
+func TestClient_WithCustomBatchTimeout(t *testing.T) {
+	t.Parallel()
+
+	_, opts, serverTeardown := NewMockedSpannerInMemTestServer(t)
+	defer serverTeardown()
+
+	wantBatchTimeout := time.Second * 42
+	client, err := NewClientWithConfig(context.Background(), "projects/p/instances/i/databases/d", ClientConfig{BatchTimeout: wantBatchTimeout}, opts...)
+	if err != nil {
+		t.Fatalf("failed to get a client: %v", err)
+	}
+	if wantBatchTimeout != client.sc.batchTimeout {
+		t.Fatalf("mismatch in client configuration for property BatchTimeout: got %v, want %v", client.sc.batchTimeout, wantBatchTimeout)
+	}
+}
+
+func TestClient_WithoutCustomBatchTimeout(t *testing.T) {
+	t.Parallel()
+
+	_, opts, serverTeardown := NewMockedSpannerInMemTestServer(t)
+	defer serverTeardown()
+
+	wantBatchTimeout := time.Minute
+	client, err := NewClientWithConfig(context.Background(), "projects/p/instances/i/databases/d", ClientConfig{}, opts...)
+	if err != nil {
+		t.Fatalf("failed to get a client: %v", err)
+	}
+	if wantBatchTimeout != client.sc.batchTimeout {
+		t.Fatalf("mismatch in client configuration for property BatchTimeout: got %v, want %v", client.sc.batchTimeout, wantBatchTimeout)
+	}
+}
+
 func TestClient_CallOptions(t *testing.T) {
 	t.Parallel()
 	co := &vkit.CallOptions{

spanner/sessionclient.go

@@ -102,7 +102,7 @@ type sessionClient struct {
 }
 
 // newSessionClient creates a session client to use for a database.
-func newSessionClient(connPool gtransport.ConnPool, database, userAgent string, sessionLabels map[string]string, databaseRole string, disableRouteToLeader bool, md metadata.MD, logger *log.Logger, callOptions *vkit.CallOptions) *sessionClient {
+func newSessionClient(connPool gtransport.ConnPool, database, userAgent string, sessionLabels map[string]string, databaseRole string, disableRouteToLeader bool, md metadata.MD, batchTimeout time.Duration, logger *log.Logger, callOptions *vkit.CallOptions) *sessionClient {
 	return &sessionClient{
 		connPool:             connPool,
 		database:             database,
@@ -112,7 +112,7 @@ func newSessionClient(connPool gtransport.ConnPool, database, userAgent string,
 		databaseRole:         databaseRole,
 		disableRouteToLeader: disableRouteToLeader,
 		md:                   md,
-		batchTimeout:         time.Minute,
+		batchTimeout:         batchTimeout,
 		logger:               logger,
 		callOptions:          callOptions,
 	}