Skip to content
This repository was archived by the owner on May 8, 2026. It is now read-only.

feat: recycle channel on consecutive new stream failures #2903

Closed
nimf wants to merge 3 commits into from
+124 −5
Closed

feat: recycle channel on consecutive new stream failures #2903

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c6206f8
feat: recycle channel on consecutive new stream failures
nimf Apr 23, 2026
a1ff8ec
add backoff for recycling a channel
nimf May 1, 2026
dd64940
chore: generate libraries at Fri May 1 20:43:37 UTC 2026
cloud-java-bot May 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .kokoro/presubmit/graalvm-native-a.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Configure the docker image for kokoro-trampoline.
env_vars: {
key: "TRAMPOLINE_IMAGE"
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.61.0" # {x-version-update:google-cloud-shared-dependencies:current}
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_a:3.62.0-SNAPSHOT" # {x-version-update:google-cloud-shared-dependencies:current}
}

env_vars: {
Expand Down
2 changes: 1 addition & 1 deletion .kokoro/presubmit/graalvm-native-b.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Configure the docker image for kokoro-trampoline.
env_vars: {
key: "TRAMPOLINE_IMAGE"
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.61.0" # {x-version-update:google-cloud-shared-dependencies:current}
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_b:3.62.0-SNAPSHOT" # {x-version-update:google-cloud-shared-dependencies:current}
}

env_vars: {
Expand Down
2 changes: 1 addition & 1 deletion .kokoro/presubmit/graalvm-native-c.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# Configure the docker image for kokoro-trampoline.
env_vars: {
key: "TRAMPOLINE_IMAGE"
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_c:3.61.0" # {x-version-update:google-cloud-shared-dependencies:current}
value: "gcr.io/cloud-devrel-public-resources/graalvm_sdk_platform_c:3.62.0-SNAPSHOT" # {x-version-update:google-cloud-shared-dependencies:current}
}

env_vars: {
Expand Down
29 changes: 27 additions & 2 deletions .../src/main/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,9 @@ public class ChannelPoolDpImpl implements ChannelPool {
private static final String DEFAULT_LOG_NAME = "pool";
private static final AtomicInteger INDEX = new AtomicInteger();

// TODO: Move to client configuration.
private static final int CONSECUTIVE_OPEN_SESSION_FAILURE_THRESHOLD = 5;

private final String poolLogId;

@VisibleForTesting volatile int minGroups;
Expand Down Expand Up @@ -95,6 +98,12 @@ public class ChannelPoolDpImpl implements ChannelPool {
@GuardedBy("this")
private boolean closed = false;

@GuardedBy("this")
private long lastRecycleNano = 0;

@GuardedBy("this")
private Duration recycleBackoff = Duration.ofMillis(1);

public ChannelPoolDpImpl(
Supplier<ManagedChannel> channelSupplier,
ChannelPoolConfiguration config,
Expand Down Expand Up @@ -221,6 +230,8 @@ public void start(Listener responseListener, Metadata headers) {
public void onBeforeSessionStart(PeerInfo peerInfo) {
afeId = AfeId.extract(peerInfo);
synchronized (ChannelPoolDpImpl.this) {
channelWrapper.consecutiveFailures = 0;
recycleBackoff = Duration.ofMillis(1);
rehomeChannel(channelWrapper, afeId);
sessionsPerAfeId.add(afeId);
}
Expand All @@ -232,6 +243,8 @@ public void onClose(Status status, Metadata trailers) {
synchronized (ChannelPoolDpImpl.this) {
if (afeId != null) {
sessionsPerAfeId.remove(afeId);
} else if (!status.isOk() && status.getCode() != Code.CANCELLED) {
channelWrapper.consecutiveFailures++;
}
releaseChannel(channelWrapper, status);
Copy link
Copy Markdown
Contributor

@mutianf mutianf Apr 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not related to this PR, but I feel releaseChannel should be called releaseStream or something, with recycleChannel the naming gets a little confusing..

Copy link
Copy Markdown
Contributor Author

@nimf nimf May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess the meaning here was that a session is closing and releasing the channel it was using.

}
Expand Down Expand Up @@ -306,12 +319,12 @@ private void releaseChannel(ChannelWrapper channelWrapper, Status status) {
channelWrapper.group.numStreams--;
channelWrapper.numOutstanding--;

if (shouldRecycleChannel(status)) {
if (shouldRecycleChannel(channelWrapper, status)) {
recycleChannel(channelWrapper);
}
}

private static boolean shouldRecycleChannel(Status status) {
private static boolean shouldRecycleChannel(ChannelWrapper channelWrapper, Status status) {
Copy link
Copy Markdown
Contributor

@mutianf mutianf Apr 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also unrelated to this PR.. I'm wondering how this works..

if we get unimplemented, shouldn't the client fallback to unary? (iiuc which doesn't use this channel pool) recycling the channel and try to create a new one likely still won't work?

Copy link
Copy Markdown
Contributor Author

@nimf nimf May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, we will eventually fallback to unary if recycling doesn't help. But I've added a backoff not to spam new connections rapidly and indefinitely.

if (status.getCode() == Code.UNIMPLEMENTED) {
return true;
}
Expand All @@ -322,6 +335,10 @@ private static boolean shouldRecycleChannel(Status status) {
return true;
}

if (channelWrapper.consecutiveFailures >= CONSECUTIVE_OPEN_SESSION_FAILURE_THRESHOLD) {
return true;
}

return false;
}

Expand All @@ -332,6 +349,13 @@ private void recycleChannel(ChannelWrapper channelWrapper) {
return;
}

if (lastRecycleNano > System.nanoTime() - recycleBackoff.toNanos()) {
return;
}

lastRecycleNano = System.nanoTime();
recycleBackoff = recycleBackoff.multipliedBy(2);

channelWrapper.group.channels.remove(channelWrapper);
channelWrapper.channel.shutdown();
// Checking for starting group because we don't want to delete the stating group.
Expand Down Expand Up @@ -480,6 +504,7 @@ static class ChannelWrapper {
private final ManagedChannel channel;
private final Instant createdAt;
private int numOutstanding = 0;
private int consecutiveFailures = 0;

public ChannelWrapper(AfeChannelGroup group, ManagedChannel channel, Clock clock) {
this.group = group;
Expand Down
94 changes: 94 additions & 0 deletions .../test/java/com/google/cloud/bigtable/data/v2/internal/channels/ChannelPoolDpImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -469,4 +469,98 @@ void testRecycledChannelDoesNotRejoinPool() throws InterruptedException {

pool.close();
}

@Test
void testRecycleChannelOnConsecutiveFailures() {
when(channelSupplier.get()).thenReturn(channel);
when(channel.newCall(any(), any())).thenReturn(clientCall);
doNothing().when(clientCall).start(listener.capture(), any());

ChannelPoolDpImpl pool =
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);

for (int i = 0; i < 4; i++) {
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());

// Should not be recycled yet
verify(channel, times(0)).shutdown();
verify(channelSupplier, times(1)).get();
}

// 5th failure
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());

// Now it should be recycled
verify(channel, times(1)).shutdown();
verify(channelSupplier, times(2)).get();

pool.close();
}

@Test
void testResetConsecutiveFailuresOnSuccess() {
when(channelSupplier.get()).thenReturn(channel);
when(channel.newCall(any(), any())).thenReturn(clientCall);
doNothing().when(clientCall).start(listener.capture(), any());
doReturn(Attributes.EMPTY).when(clientCall).getAttributes();

ChannelPoolDpImpl pool =
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);

// 4 failures
for (int i = 0; i < 4; i++) {
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
}
verify(channel, times(0)).shutdown();

// A success: onHeaders (which calls onBeforeSessionStart)
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());

PeerInfo peerInfo = PeerInfo.newBuilder().setApplicationFrontendId(555).build();
Metadata headers = new Metadata();
headers.put(
SessionStreamImpl.PEER_INFO_KEY,
Base64.getEncoder().encodeToString(peerInfo.toByteArray()));
listener.getValue().onHeaders(headers);
listener.getValue().onClose(Status.OK, new Metadata());

// Another 4 failures - should still not recycle because counter was reset
for (int i = 0; i < 4; i++) {
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());
listener.getValue().onClose(Status.UNAVAILABLE, new Metadata());
}
verify(channel, times(0)).shutdown();

pool.close();
}

@Test
void testCancelledDoesNotIncrementFailures() {
when(channelSupplier.get()).thenReturn(channel);
when(channel.newCall(any(), any())).thenReturn(clientCall);
doNothing().when(clientCall).start(listener.capture(), any());

ChannelPoolDpImpl pool =
new ChannelPoolDpImpl(channelSupplier, defaultConfig, debugTagTracer, bgExecutor);

for (int i = 0; i < 10; i++) {
pool.newStream(FakeSessionGrpc.getOpenSessionMethod(), CallOptions.DEFAULT)
.start(mock(Listener.class), new Metadata());
listener.getValue().onClose(Status.CANCELLED, new Metadata());
}

// Should never be recycled
verify(channel, times(0)).shutdown();
verify(channelSupplier, times(1)).get();

pool.close();
}
}
Loading