feat(bigtable): add preemptive connection recycler

bigtable/client.go

@@ -49,6 +49,7 @@ type Client struct {
 	enableDirectAccess      bool
 	featureFlagsMD          metadata.MD // Pre-computed feature flags metadata to be sent with each request.
 	dynamicScaleMonitor     *btransport.DynamicScaleMonitor
+	connsRecycler           *btransport.ConnectionRecycler
 }
 
 // ClientConfig has configurations for the client.
@@ -64,8 +65,13 @@ type ClientConfig struct {
 	// TODO: support user provided meter provider
 	MetricsProvider MetricsProvider
 
-	// If true, enable dynamic channel pool
-	EnableDynamicChannelPool bool
+	// DisableDynamicChannelPool disables the dynamic channel resizing based on load
+	// Dynamic channel resizing  is enabled by default to resize based on load and avoid queuing of requests.
+	DisableDynamicChannelPool bool
+
+	// DisableConnectionRecycler disables the automatic preemptive refresh of connection.
+	// Preemptive connection is default to true
+	DisableConnectionRecycler bool
 }
 
 // MetricsProvider is a wrapper for built in metrics meter provider
@@ -153,6 +159,8 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 	var connPool gtransport.ConnPool
 	var connPoolErr error
 	var dsm *btransport.DynamicScaleMonitor
+	var connRecycler *btransport.ConnectionRecycler
+
 	enableBigtableConnPool := btopt.EnableBigtableConnectionPool()
 	if enableBigtableConnPool {
 		fullInstanceName := fmt.Sprintf("projects/%s/instances/%s", project, instance)
@@ -182,14 +190,20 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 			connPool = btPool
 
 			// Validate dynamic config early if enabled
-			if config.EnableDynamicChannelPool {
+			if !config.DisableDynamicChannelPool {
 				if err := btransport.ValidateDynamicConfig(btopt.DefaultDynamicChannelPoolConfig(), defaultBigtableConnPoolSize); err != nil {
 					return nil, fmt.Errorf("invalid DynamicChannelPoolConfig: %w", err)
 				}
 
 				dsm = btransport.NewDynamicScaleMonitor(btopt.DefaultDynamicChannelPoolConfig(), btPool)
 				dsm.Start(ctx) // Start the monitor's background goroutine
 			}
+			// connection recyler.
+			if !config.DisableConnectionRecycler {
+				connRecycler = btransport.NewConnectionRecycler(btopt.DefaultConnectionRecycleConfig(), btPool)
+				connRecycler.Start(ctx) // Start the monitor's background goroutine
+			}
+
 		}
 
 	} else {
@@ -214,6 +228,7 @@ func NewClientWithConfig(ctx context.Context, project, instance string, config C
 		enableDirectAccess:      enableDirectAccess,
 		featureFlagsMD:          ffMD,
 		dynamicScaleMonitor:     dsm,
+		connsRecycler:           connRecycler,
 	}, nil
 }
 
@@ -225,6 +240,9 @@ func (c *Client) Close() error {
 	if c.metricsTracerFactory != nil {
 		c.metricsTracerFactory.shutdown()
 	}
+	if c.connsRecycler != nil {
+		c.connsRecycler.Stop()
+	}
 	return c.connPool.Close()
 }
 

bigtable/internal/option/option.go

@@ -271,3 +271,23 @@ func DefaultMetricsReporterConfig() MetricsReporterConfig {
 		ReportingInterval: 1 * time.Minute,
 	}
 }
+
+// ConnectionRecycleConfig controls the behavior of the connection recycler.
+type ConnectionRecycleConfig struct {
+	// MaxAge is the base lifespan of a connection.
+	MaxAge time.Duration
+	// Jitter is the random buffer added to MaxAge which can allow for connection to be recycled.
+	MaxJitter time.Duration
+	// RunFrequency determines how often the recycler checks for expired connections.
+	RunFrequency time.Duration
+}
+
+// DefaultConnectionRecycleConfig returns the default configuration:
+// MaxAge: 45 minutes, Jitter: 5 minutes, RunFrequency: 1 minute, Enabled: true.
+func DefaultConnectionRecycleConfig() ConnectionRecycleConfig {
+	return ConnectionRecycleConfig{
+		MaxAge:       45 * time.Minute,
+		MaxJitter:    5 * time.Minute,
+		RunFrequency: 1 * time.Minute,
+	}
+}

bigtable/internal/transport/conn_recycler.go

@@ -0,0 +1,115 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+	"math/rand"
+	"sync"
+	"time"
+
+	btopt "cloud.google.com/go/bigtable/internal/option"
+)
+
+// maxRecyclePerBatch limits the number of connections we replace in a single pass.
+const maxRecyclePerBatch = 2
+
+// ConnectionRecycler monitors connection age and recycles them to prevent long-lived connections.
+type ConnectionRecycler struct {
+	pool     *BigtableChannelPool
+	config   btopt.ConnectionRecycleConfig
+	ticker   *time.Ticker
+	done     chan struct{}
+	stopOnce sync.Once
+	rng      *rand.Rand
+}
+
+// NewConnectionRecycler creates a new recycler with the provided configuration.
+func NewConnectionRecycler(config btopt.ConnectionRecycleConfig, pool *BigtableChannelPool) *ConnectionRecycler {
+	return &ConnectionRecycler{
+		pool:   pool,
+		config: config,
+		done:   make(chan struct{}),
+		rng:    rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+}
+
+// Start begins the periodic monitoring.
+func (cr *ConnectionRecycler) Start(ctx context.Context) {
+	btopt.Debugf(cr.pool.logger, "bigtable_connpool: ConnectionRecyler starting...")
+
+	// default to 1 minute
+	freq := cr.config.RunFrequency
+	if freq < 1*time.Minute {
+		freq = 1 * time.Minute
+	}
+
+	// at least once per MaxAge interval.
+	if cr.config.MaxAge > 0 && freq > cr.config.MaxAge {
+		freq = cr.config.MaxAge
+	}
+
+	cr.ticker = time.NewTicker(freq)
+	go func() {
+		defer cr.ticker.Stop()
+		for {
+			select {
+			case <-cr.ticker.C:
+				cr.checkRecycle()
+			case <-cr.done:
+				return
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+}
+
+// Stop terminates the ConnectionRecycler.
+func (cr *ConnectionRecycler) Stop() {
+	cr.stopOnce.Do(func() {
+		close(cr.done)
+	})
+}
+
+// background period task
+func (cr *ConnectionRecycler) checkRecycle() {
+	conns := cr.pool.getConns()
+	recycledCount := 0
+
+	hasJitter := cr.config.MaxJitter > 0
+	jitterVal := int64(cr.config.MaxJitter)
+
+	for _, entry := range conns {
+		if recycledCount >= maxRecyclePerBatch {
+			btopt.Debugf(cr.pool.logger, "bigtable_connpool: Hit max recycle cap (%d) for this round", maxRecyclePerBatch)
+			break
+		}
+
+		createdAt := time.UnixMilli(entry.createdAt())
+		age := time.Since(createdAt)
+
+		var currentJitter time.Duration
+		if hasJitter {
+			currentJitter = time.Duration(cr.rng.Int63n(jitterVal))
+		}
+
+		if age > cr.config.MaxAge+currentJitter {
+			btopt.Debugf(cr.pool.logger, "bigtable_connpool: Recycling connection age %v > %v + %v", age, cr.config.MaxAge, currentJitter)
+			cr.pool.replaceConnection(entry)
+			recycledCount++
+		}
+	}
+}

bigtable/internal/transport/conn_recycler_test.go

@@ -0,0 +1,132 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	btopt "cloud.google.com/go/bigtable/internal/option"
+)
+
+func TestConnectionRecycler_CheckRecycle(t *testing.T) {
+	fake := &fakeService{}
+	addr := setupTestServer(t, fake)
+	dialFunc := func() (*BigtableConn, error) { return dialBigtableserver(addr) }
+	ctx := context.Background()
+
+	setAge := func(entry *connEntry, age time.Duration) {
+		entry.conn.createdAt.Store(time.Now().Add(-age).UnixMilli())
+	}
+
+	t.Run("RecycleOldConnection", func(t *testing.T) {
+		config := btopt.ConnectionRecycleConfig{
+			MaxAge:    10 * time.Minute,
+			MaxJitter: 0,
+		}
+
+		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now())
+		if err != nil {
+			t.Fatalf("Failed to create pool: %v", err)
+		}
+		defer pool.Close()
+
+		recycler := NewConnectionRecycler(config, pool)
+
+		conns := pool.getConns()
+		if len(conns) != 1 {
+			t.Fatalf("Expected 1 connection, got %d", len(conns))
+		}
+		originalEntry := conns[0]
+		originalConnPtr := originalEntry.conn
+
+		// maxAge > 20m
+		setAge(originalEntry, 20*time.Minute)
+		recycler.checkRecycle()
+
+		// recycled fast as it does not have any pending rpcs
+		newConns := pool.getConns()
+		if newConns[0].conn == originalConnPtr {
+			t.Error("Connection was older than MaxAge but was NOT recycled")
+		}
+	})
+
+	t.Run("DoesNotReplaceIfConnWithinMaxAge", func(t *testing.T) {
+		config := btopt.ConnectionRecycleConfig{
+			MaxAge:    10 * time.Minute,
+			MaxJitter: 0,
+		}
+
+		pool, err := NewBigtableChannelPool(ctx, 1, btopt.RoundRobin, dialFunc, time.Now())
+		if err != nil {
+			t.Fatalf("Failed to create pool: %v", err)
+		}
+		defer pool.Close()
+
+		recycler := NewConnectionRecycler(config, pool)
+
+		entry := pool.getConns()[0]
+		originalConnPtr := entry.conn
+
+		// < 10mins
+		setAge(entry, 5*time.Minute)
+
+		// recycled fast as it does not have any pending rpcs
+		recycler.checkRecycle()
+
+		if pool.getConns()[0].conn != originalConnPtr {
+			t.Error("Connection WAS recycled unexpectedly")
+		}
+	})
+
+	t.Run("RespectsMaxRecyclePerBatch", func(t *testing.T) {
+		config := btopt.ConnectionRecycleConfig{
+			MaxAge:    10 * time.Minute,
+			MaxJitter: 0,
+		}
+		// 5 conns
+		poolSize := 5
+		pool, err := NewBigtableChannelPool(ctx, poolSize, btopt.RoundRobin, dialFunc, time.Now())
+		if err != nil {
+			t.Fatalf("Failed to create pool: %v", err)
+		}
+		defer pool.Close()
+
+		recycler := NewConnectionRecycler(config, pool)
+
+		// force age to be old
+		conns := pool.getConns()
+		originalConns := make(map[*BigtableConn]bool)
+		for _, e := range conns {
+			setAge(e, 60*time.Minute)
+			originalConns[e.conn] = true
+		}
+
+		// Trigger recycle
+		recycler.checkRecycle()
+
+		currentConns := pool.getConns()
+		changedCount := 0
+		for _, e := range currentConns {
+			if !originalConns[e.conn] {
+				changedCount++
+			}
+		}
+		if changedCount != maxRecyclePerBatch {
+			t.Errorf("Expected exactly %d recycled connections (batch limit), but got %d", maxRecyclePerBatch, changedCount)
+		}
+	})
+}