From 323fd8aa8f51056f9a935e2847e7167300d3fd1c Mon Sep 17 00:00:00 2001 From: Kannan J Date: Wed, 12 Aug 2026 12:24:45 +0000 Subject: [PATCH 1/2] core: reference-count shared transport factory for OOB channels When a main channel enters IDLE mode (e.g. after 30 minutes of inactivity), it shuts down its Load Balancer (such as RLS or xDS). This in turn shuts down any Out-of-Band (OOB) child channels created by the LB. Previously, when an OOB channel finished terminating, it called `transportFactory.close()`, which prematurely closed the parent channel's shared `ClientTransportFactory` (`NettyTransportFactory`). As a result, when the main channel later attempted to exit IDLE mode or reconnect, `transportFactory.newClientTransport()` threw `IllegalStateException: The transport factory is closed`, causing an uncaught exception panic in the SynchronizationContext. This change introduces `RefCountedClientTransportFactory` to track active references to the underlying `ClientTransportFactory`. When OOB channels inherit `originalTransportFactory` via `createResolvingOobChannelBuilder()`, `retain()` is invoked. Calling `close()` on the transport factory now decrements the reference count, ensuring the underlying delegate factory is only closed when all parent and child channel references have been released. --- .../io/grpc/internal/ManagedChannelImpl.java | 12 ++-- .../RefCountedClientTransportFactory.java | 71 +++++++++++++++++++ .../grpc/internal/ManagedChannelImplTest.java | 40 +++++++++++ .../RefCountedClientTransportFactoryTest.java | 71 +++++++++++++++++++ 4 files changed, 190 insertions(+), 4 deletions(-) create mode 100644 core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java create mode 100644 core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java diff --git a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java index 00df05a0c00..4356e93df4f 100644 --- a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java +++ b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java @@ -173,7 +173,7 @@ public Result selectConfig(PickSubchannelArgs args) { private final NameResolverProvider nameResolverProvider; private final NameResolver.Args nameResolverArgs; private final LoadBalancerProvider loadBalancerFactory; - private final ClientTransportFactory originalTransportFactory; + private final RefCountedClientTransportFactory originalTransportFactory; @Nullable private final ChannelCredentials originalChannelCreds; private final ClientTransportFactory transportFactory; @@ -562,11 +562,15 @@ ClientStream newSubstream( this.executorPool = checkNotNull(builder.executorPool, "executorPool"); this.executor = checkNotNull(executorPool.getObject(), "executor"); this.originalChannelCreds = builder.channelCredentials; - this.originalTransportFactory = clientTransportFactory; + if (clientTransportFactory instanceof RefCountedClientTransportFactory) { + this.originalTransportFactory = (RefCountedClientTransportFactory) clientTransportFactory; + } else { + this.originalTransportFactory = new RefCountedClientTransportFactory(clientTransportFactory); + } this.offloadExecutorHolder = new ExecutorHolder(checkNotNull(builder.offloadExecutorPool, "offloadExecutorPool")); this.transportFactory = new CallCredentialsApplyingTransportFactory( - clientTransportFactory, builder.callCredentials, this.offloadExecutorHolder); + originalTransportFactory, builder.callCredentials, this.offloadExecutorHolder); this.scheduledExecutor = new RestrictedScheduledExecutor(transportFactory.getScheduledExecutorService()); maxTraceEvents = builder.maxTraceEvents; @@ -1462,7 +1466,7 @@ final class ResolvingOobChannelBuilder final ClientTransportFactory transportFactory; CallCredentials callCredentials; if (channelCreds instanceof DefaultChannelCreds) { - transportFactory = originalTransportFactory; + transportFactory = originalTransportFactory.retain(); callCredentials = null; } else { SwapChannelCredentialsResult swapResult = diff --git a/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java new file mode 100644 index 00000000000..7898359927a --- /dev/null +++ b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java @@ -0,0 +1,71 @@ +/* + * Copyright 2026 The gRPC Authors + * + * 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 io.grpc.internal; + +import com.google.common.base.Preconditions; +import io.grpc.ChannelCredentials; +import io.grpc.ChannelLogger; +import java.net.SocketAddress; +import java.util.Collection; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A wrapper for {@link ClientTransportFactory} that reference-counts calls to {@link #retain()} and + * {@link #close()}, ensuring the delegate factory is closed only when all references are released. + */ +final class RefCountedClientTransportFactory implements ClientTransportFactory { + private final ClientTransportFactory delegate; + private final AtomicInteger refCount = new AtomicInteger(1); + + public RefCountedClientTransportFactory(ClientTransportFactory delegate) { + this.delegate = Preconditions.checkNotNull(delegate, "delegate"); + } + + public RefCountedClientTransportFactory retain() { + refCount.incrementAndGet(); + return this; + } + + @Override + public ConnectionClientTransport newClientTransport( + SocketAddress serverAddress, ClientTransportOptions options, ChannelLogger channelLogger) { + return delegate.newClientTransport(serverAddress, options, channelLogger); + } + + @Override + public ScheduledExecutorService getScheduledExecutorService() { + return delegate.getScheduledExecutorService(); + } + + @Override + public Collection> getSupportedSocketAddressTypes() { + return delegate.getSupportedSocketAddressTypes(); + } + + @Override + public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials channelCreds) { + return delegate.swapChannelCredentials(channelCreds); + } + + @Override + public void close() { + if (refCount.decrementAndGet() == 0) { + delegate.close(); + } + } +} diff --git a/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java b/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java index e958fcdae00..d052cce137b 100644 --- a/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java +++ b/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java @@ -4824,6 +4824,46 @@ public void run() { }); } + @Test + public void oobChannelTermination_doesNotCloseSharedTransportFactory() { + channelBuilder.nameResolverRegistry.register(new NameResolverProvider() { + @Override + public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) { + NameResolver resolver = mock(NameResolver.class); + when(resolver.getServiceAuthority()).thenReturn( + targetUri.getAuthority() != null ? targetUri.getAuthority() : targetUri.getPath()); + return resolver; + } + + @Override + public String getDefaultScheme() { + return expectedUri.getScheme(); + } + + @Override + protected boolean isAvailable() { + return true; + } + + @Override + protected int priority() { + return 10; + } + }); + createChannel(); + ManagedChannel oob = helper.createResolvingOobChannelBuilder("oobauthority").build(); + + // Shutting down OOB channel should release its reference but not close the + // shared transport factory + oob.shutdownNow(); + verify(mockTransportFactory, never()).close(); + + // Terminating the main channel releases the final reference and closes the + // transport factory + channel.shutdownNow(); + verify(mockTransportFactory).close(); + } + @SuppressWarnings("unchecked") private static Map parseConfig(String json) throws Exception { return (Map) JsonParser.parse(json); diff --git a/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java new file mode 100644 index 00000000000..ffff6915cfa --- /dev/null +++ b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java @@ -0,0 +1,71 @@ +/* + * Copyright 2026 The gRPC Authors + * + * 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 io.grpc.internal; + +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** Unit tests for {@link RefCountedClientTransportFactory}. */ +@RunWith(JUnit4.class) +public class RefCountedClientTransportFactoryTest { + @Rule public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock private ClientTransportFactory mockDelegate; + + @Test + public void singleClose_closesDelegate() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.close(); + verify(mockDelegate).close(); + } + + @Test + public void retainAndClose_closesDelegateOnlyWhenCountReachesZero() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + RefCountedClientTransportFactory retained = factory.retain(); + + factory.close(); + verify(mockDelegate, never()).close(); + + retained.close(); + verify(mockDelegate).close(); + } + + @Test + public void multipleRetains_requiresEqualClosesToCloseDelegate() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.retain(); + factory.retain(); + + factory.close(); + verify(mockDelegate, never()).close(); + + factory.close(); + verify(mockDelegate, never()).close(); + + factory.close(); + verify(mockDelegate).close(); + } +} From 5cb67bc2c25358d48082097c24590167689d23c4 Mon Sep 17 00:00:00 2001 From: Kannan J Date: Thu, 13 Aug 2026 04:53:22 +0000 Subject: [PATCH 2/2] Address review comments. --- .../main/java/io/grpc/internal/ManagedChannelImpl.java | 3 +++ .../internal/RefCountedClientTransportFactory.java | 10 +++++++--- .../internal/RefCountedClientTransportFactoryTest.java | 10 ++++++++++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java index 4356e93df4f..e533770da22 100644 --- a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java +++ b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java @@ -1466,6 +1466,9 @@ final class ResolvingOobChannelBuilder final ClientTransportFactory transportFactory; CallCredentials callCredentials; if (channelCreds instanceof DefaultChannelCreds) { + // TODO(kannanjgithub) We should eventually refactor ManagedChannelImplBuilder so + // callCredentials can be resolved lazily at build() time, allowing transport factory + // retention to happen strictly inside buildClientTransportFactory(). transportFactory = originalTransportFactory.retain(); callCredentials = null; } else { diff --git a/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java index 7898359927a..b1962b1e9f0 100644 --- a/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java +++ b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java @@ -16,7 +16,9 @@ package io.grpc.internal; -import com.google.common.base.Preconditions; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; + import io.grpc.ChannelCredentials; import io.grpc.ChannelLogger; import java.net.SocketAddress; @@ -33,7 +35,7 @@ final class RefCountedClientTransportFactory implements ClientTransportFactory { private final AtomicInteger refCount = new AtomicInteger(1); public RefCountedClientTransportFactory(ClientTransportFactory delegate) { - this.delegate = Preconditions.checkNotNull(delegate, "delegate"); + this.delegate = checkNotNull(delegate, "delegate"); } public RefCountedClientTransportFactory retain() { @@ -64,7 +66,9 @@ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials ch @Override public void close() { - if (refCount.decrementAndGet() == 0) { + int count = refCount.decrementAndGet(); + checkState(count >= 0, "Reference count has gone negative: %s", count); + if (count == 0) { delegate.close(); } } diff --git a/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java index ffff6915cfa..3ead6396ce5 100644 --- a/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java +++ b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java @@ -16,6 +16,7 @@ package io.grpc.internal; +import static org.junit.Assert.assertThrows; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -68,4 +69,13 @@ public void multipleRetains_requiresEqualClosesToCloseDelegate() { factory.close(); verify(mockDelegate).close(); } + + @Test + public void closeMoreThanRetain_throwsIllegalStateException() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.close(); + verify(mockDelegate).close(); + + assertThrows(IllegalStateException.class, factory::close); + } }