add recycle backoff test

Author: nimf

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java

@@ -23,6 +23,7 @@
 import com.google.bigtable.v2.TelemetryConfiguration;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ticker;
 import com.google.common.collect.HashMultiset;
 import com.google.common.collect.Multiset;
 import io.grpc.CallOptions;
@@ -100,6 +101,8 @@ public class ChannelPoolDpImpl implements ChannelPool {
   @GuardedBy("this")
   private boolean closed = false;
 
+  private final Ticker ticker;
+
   @GuardedBy("this")
   private long lastRecycleNano = 0;
 
@@ -111,7 +114,14 @@ public ChannelPoolDpImpl(
       ChannelPoolConfiguration config,
       DebugTagTracer debugTagTracer,
       ScheduledExecutorService executor) {
-    this(channelSupplier, config, DEFAULT_LOG_NAME, debugTagTracer, executor, Clock.systemUTC());
+    this(
+        channelSupplier,
+        config,
+        DEFAULT_LOG_NAME,
+        debugTagTracer,
+        executor,
+        Clock.systemUTC(),
+        Ticker.systemTicker());
   }
 
   public ChannelPoolDpImpl(
@@ -120,7 +130,14 @@ public ChannelPoolDpImpl(
       String logName,
       DebugTagTracer debugTagTracer,
       ScheduledExecutorService executor) {
-    this(channelSupplier, config, logName, debugTagTracer, executor, Clock.systemUTC());
+    this(
+        channelSupplier,
+        config,
+        logName,
+        debugTagTracer,
+        executor,
+        Clock.systemUTC(),
+        Ticker.systemTicker());
   }
 
   public ChannelPoolDpImpl(
@@ -130,8 +147,20 @@ public ChannelPoolDpImpl(
       DebugTagTracer debugTagTracer,
       ScheduledExecutorService executor,
       Clock clock) {
+    this(channelSupplier, config, logName, debugTagTracer, executor, clock, Ticker.systemTicker());
+  }
+
+  public ChannelPoolDpImpl(
+      Supplier<ManagedChannel> channelSupplier,
+      ChannelPoolConfiguration config,
+      String logName,
+      DebugTagTracer debugTagTracer,
+      ScheduledExecutorService executor,
+      Clock clock,
+      Ticker ticker) {
     this.poolLogId = String.format("%d-%s", INDEX.getAndIncrement(), logName);
     this.clock = clock;
+    this.ticker = ticker;
     this.channelSupplier = channelSupplier;
     this.executor = executor;
     updateConfig(config);
@@ -351,11 +380,12 @@ private void recycleChannel(ChannelWrapper channelWrapper) {
       return;
     }
 
-    if (lastRecycleNano > System.nanoTime() - recycleBackoff.toNanos()) {
+    long nowNano = ticker.read();
+    if (nowNano - lastRecycleNano < recycleBackoff.toNanos()) {
       return;
     }
 
-    lastRecycleNano = System.nanoTime();
+    lastRecycleNano = nowNano;
     recycleBackoff = recycleBackoff.multipliedBy(2);
     if (recycleBackoff.compareTo(MAX_RECYCLE_BACKOFF) > 0) {
       recycleBackoff = MAX_RECYCLE_BACKOFF;
@@ -388,9 +418,6 @@ private synchronized void serviceChannelsSafe() {
     log(Level.FINE, "Servicing channels");
     dumpState();
 
-    Instant now = Instant.now(clock);
-    Instant createdAtThreshold = now.minus(Duration.ofMinutes(50));
-
     // Thin out the channels in each group, so that each AFEGroup only has 1 channel
     for (AfeChannelGroup group : channelGroups) {
       if (LOGGER.isLoggable(Level.FINEST) && group.channels.size() > 1) {

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImplTest.java

@@ -30,6 +30,7 @@
 import com.google.cloud.bigtable.data.v2.internal.channels.SessionStream.Listener;
 import com.google.cloud.bigtable.data.v2.internal.csm.NoopMetrics;
 import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
+import com.google.common.base.Ticker;
 import io.grpc.Attributes;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
@@ -41,6 +42,8 @@
 import java.util.Base64;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -235,7 +238,6 @@ void testDownsize() {
       listener.onClose(Status.OK, new Metadata());
     }
 
-    when(clock.instant()).thenReturn(Instant.now());
     pool.serviceChannels();
     verify(channel, times(numChannels - pool.minGroups)).shutdown();
 
@@ -295,7 +297,6 @@ void testDownsizeToOptimal() {
     // I.e. dumpState
     // FINE: ChannelPool channelGroups: 5, channels: 5, starting channels: 0, totalStreams: 19,
     // AFEs: 5, distribution: [4, 4, 4, 4, 3]
-    when(clock.instant()).thenReturn(Instant.now());
     pool.serviceChannels();
 
     // Should scale down to 4 channels. 19 / 5 round up = 4.
@@ -563,4 +564,75 @@ void testCancelledDoesNotIncrementFailures() {
 
     pool.close();
   }
+
+  @Test
+  void testRecycleChannelBackoff() {
+    when(channelSupplier.get()).thenReturn(channel);
+    when(channel.newCall(any(), any())).thenReturn(clientCall);
+    doNothing().when(clientCall).start(listener.capture(), any());
+
+    Ticker ticker = mock(Ticker.class);
+    long startNanos = TimeUnit.SECONDS.toNanos(1);
+    final AtomicLong time = new AtomicLong(startNanos);
+    when(ticker.read()).thenAnswer(invocation -> time.get());
+
+    ChannelPoolDpImpl pool =
+        new ChannelPoolDpImpl(
+            channelSupplier,
+            defaultConfig,
+            "pool",
+            debugTagTracer,
+            bgExecutor,
+            Clock.systemUTC(),
+            ticker);
+
+    // --- First Recycle ---
+    for (int i = 0; i < 5; i++) {
+      pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
+          .start(mock(Listener.class), new Metadata());
+      listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
+    }
+    // Should be recycled once
+    verify(channel, times(1)).shutdown();
+    verify(channelSupplier, times(2)).get(); // 1 initial + 1 recycle
+
+    // --- Second Recycle (Immediate, same time) ---
+    // Time has not advanced. Backoff is now 2ms.
+    for (int i = 0; i < 5; i++) {
+      pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
+          .start(mock(Listener.class), new Metadata());
+      listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
+    }
+    // Should NOT be recycled again because of backoff
+    verify(channel, times(1)).shutdown();
+    verify(channelSupplier, times(2)).get();
+
+    // --- Third Recycle (After partial backoff, still blocked) ---
+    // Advance time by 1ms (backoff is 2ms, so still blocked)
+    time.addAndGet(TimeUnit.MILLISECONDS.toNanos(1));
+
+    for (int i = 0; i < 5; i++) {
+      pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
+          .start(mock(Listener.class), new Metadata());
+      listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
+    }
+    // Should still NOT be recycled
+    verify(channel, times(1)).shutdown();
+    verify(channelSupplier, times(2)).get();
+
+    // --- Fourth Recycle (After full backoff) ---
+    // Advance time by another 2ms (total 3ms from last recycle, which is > 2ms backoff)
+    time.addAndGet(TimeUnit.MILLISECONDS.toNanos(2));
+
+    for (int i = 0; i < 5; i++) {
+      pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
+          .start(mock(Listener.class), new Metadata());
+      listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
+    }
+    // Now it should be recycled again
+    verify(channel, times(2)).shutdown();
+    verify(channelSupplier, times(3)).get();
+
+    pool.close();
+  }
 }