diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/Metrics.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/Metrics.java index 7df665c673..6e30d3dd2b 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/Metrics.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/Metrics.java @@ -18,6 +18,7 @@ import com.google.api.gax.tracing.ApiTracerFactory; import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo; import com.google.cloud.bigtable.data.v2.internal.csm.tracers.ChannelPoolMetricsTracer; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; import io.grpc.ManagedChannelBuilder; import java.io.Closeable; import java.io.IOException; @@ -31,6 +32,8 @@ public interface Metrics extends Closeable { @Nullable ChannelPoolMetricsTracer getChannelPoolMetricsTracer(); + DirectPathCompatibleTracer getDirectPathCompatibleTracer(); + void start(); @Override diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricsImpl.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricsImpl.java index f0efac7e96..69391b9796 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricsImpl.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/MetricsImpl.java @@ -31,6 +31,9 @@ import com.google.cloud.bigtable.data.v2.internal.csm.tracers.BuiltinMetricsTracerFactory; import com.google.cloud.bigtable.data.v2.internal.csm.tracers.ChannelPoolMetricsTracer; import com.google.cloud.bigtable.data.v2.internal.csm.tracers.CompositeTracerFactory; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DefaultDirectPathCompatibleTracer; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.NoopDirectPathCompatibleTracer; import com.google.cloud.bigtable.data.v2.internal.csm.tracers.Pacemaker; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; @@ -67,6 +70,7 @@ public class MetricsImpl implements Metrics, Closeable { @Nullable private final GrpcOpenTelemetry grpcOtel; @Nullable private final ChannelPoolMetricsTracer channelPoolMetricsTracer; + @Nullable private final DirectPathCompatibleTracer directPathCompatibleTracer; @Nullable private final Pacemaker pacemaker; private final List> tasks = new ArrayList<>(); @@ -94,6 +98,8 @@ public MetricsImpl( this.internalRecorder = metricRegistry.newRecorderRegistry(internalOtel.getMeterProvider()); this.pacemaker = new Pacemaker(internalRecorder, clientInfo, "background"); this.channelPoolMetricsTracer = new ChannelPoolMetricsTracer(internalRecorder, clientInfo); + this.directPathCompatibleTracer = + new DefaultDirectPathCompatibleTracer(clientInfo, internalRecorder); this.grpcOtel = GrpcOpenTelemetry.newBuilder() .sdk(internalOtel) @@ -109,6 +115,7 @@ public MetricsImpl( this.grpcOtel = null; this.pacemaker = null; this.channelPoolMetricsTracer = null; + this.directPathCompatibleTracer = NoopDirectPathCompatibleTracer.INSTANCE; } if (userOtel != null) { @@ -171,6 +178,12 @@ public ChannelPoolMetricsTracer getChannelPoolMetricsTracer() { return channelPoolMetricsTracer; } + @Nullable + @Override + public DirectPathCompatibleTracer getDirectPathCompatibleTracer() { + return directPathCompatibleTracer; + } + public static OpenTelemetrySdk createBuiltinOtel( MetricRegistry metricRegistry, ClientInfo clientInfo, diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DefaultDirectPathCompatibleTracer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DefaultDirectPathCompatibleTracer.java new file mode 100644 index 0000000000..465e147868 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DefaultDirectPathCompatibleTracer.java @@ -0,0 +1,42 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.csm.tracers; + +import com.google.api.core.InternalApi; +import com.google.cloud.bigtable.data.v2.internal.csm.MetricRegistry; +import com.google.cloud.bigtable.data.v2.internal.csm.attributes.ClientInfo; + +@InternalApi +public class DefaultDirectPathCompatibleTracer implements DirectPathCompatibleTracer { + private final ClientInfo clientInfo; + private final MetricRegistry.RecorderRegistry recorder; + + public DefaultDirectPathCompatibleTracer( + ClientInfo clientInfo, MetricRegistry.RecorderRegistry recorder) { + this.clientInfo = clientInfo; + this.recorder = recorder; + } + + @Override + public void recordSuccess(String ipPreference) { + recorder.dpCompatGuage.recordSuccess(clientInfo, ipPreference); + } + + @Override + public void recordFailure(String reason) { + recorder.dpCompatGuage.recordFailure(clientInfo, reason); + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DirectPathCompatibleTracer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DirectPathCompatibleTracer.java new file mode 100644 index 0000000000..4776785dc2 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/DirectPathCompatibleTracer.java @@ -0,0 +1,37 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.csm.tracers; + +import com.google.api.core.InternalApi; + +/** Interface for recording DirectPath/DirectAccess eligibility metrics. */ +@InternalApi +public interface DirectPathCompatibleTracer { + + /** + * Records that the environment is eligible and successfully connected via DirectPath. + * + * @param ipPreference The IP preference used (e.g., "ipv6"). + */ + void recordSuccess(String ipPreference); + + /** + * Records that the environment is not eligible or failed to connect via DirectPath. + * + * @param reason The reason for the failure (e.g., "routing_check_failed"). + */ + void recordFailure(String reason); +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/NoopDirectPathCompatibleTracer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/NoopDirectPathCompatibleTracer.java new file mode 100644 index 0000000000..bcfb8a48f7 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/csm/tracers/NoopDirectPathCompatibleTracer.java @@ -0,0 +1,37 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.csm.tracers; + +import com.google.api.core.InternalApi; + +@InternalApi +public class NoopDirectPathCompatibleTracer implements DirectPathCompatibleTracer { + + public static final NoopDirectPathCompatibleTracer INSTANCE = + new NoopDirectPathCompatibleTracer(); + + private NoopDirectPathCompatibleTracer() {} + + @Override + public void recordSuccess(String ipPreference) { + // No-op + } + + @Override + public void recordFailure(String reason) { + // No-op + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessChecker.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessChecker.java new file mode 100644 index 0000000000..cfd14741cc --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessChecker.java @@ -0,0 +1,94 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.dp; + +import com.google.api.core.InternalApi; +import com.google.bigtable.v2.PeerInfo; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; +import com.google.cloud.bigtable.data.v2.stub.MetadataExtractorInterceptor; +import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer; +import com.google.common.annotations.VisibleForTesting; +import io.grpc.Channel; +import io.grpc.ClientInterceptors; +import io.grpc.ManagedChannel; +import java.util.Optional; +import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; +import javax.annotation.Nullable; + +/** + * Evaluates whether a given channel has Direct Access (DirectPath) routing by executing a RPC and + * inspecting the response headers. + */ +@InternalApi +public class ClassicDirectAccessChecker implements DirectAccessChecker { + private static final Logger LOG = Logger.getLogger(ClassicDirectAccessChecker.class.getName()); + private final ChannelPrimer channelPrimer; + + private ClassicDirectAccessChecker(ChannelPrimer channelPrimer) { + this.channelPrimer = channelPrimer; + } + + public static ClassicDirectAccessChecker create(ChannelPrimer channelPrimer) { + return new ClassicDirectAccessChecker(channelPrimer); + } + + @VisibleForTesting + MetadataExtractorInterceptor createInterceptor() { + return new MetadataExtractorInterceptor(); + } + + @Override + public boolean check( + Supplier channelSupplier, @Nullable DirectPathCompatibleTracer tracer) { + ManagedChannel channel = null; + try { + channel = channelSupplier.get(); + MetadataExtractorInterceptor interceptor = createInterceptor(); + Channel interceptedChannel = ClientInterceptors.intercept(channel, interceptor); + channelPrimer.primeChannel(interceptedChannel); + + // Extract the sideband data populated by the interceptor + MetadataExtractorInterceptor.SidebandData sidebandData = interceptor.getSidebandData(); + + boolean isEligible = + Optional.ofNullable(sidebandData) + .map(MetadataExtractorInterceptor.SidebandData::getPeerInfo) + .map(PeerInfo::getTransportType) + .map(type -> type == PeerInfo.TransportType.TRANSPORT_TYPE_DIRECT_ACCESS) + .orElse(false); + + if (isEligible) { + String ipProtocolStr = + sidebandData.getIpProtocol() != null + ? sidebandData.getIpProtocol().toString().toLowerCase() + : "unknown"; + tracer.recordSuccess(ipProtocolStr); + } + + return isEligible; + + } catch (Exception e) { + LOG.log(Level.FINE, "Failed to evaluate direct access eligibility.", e); + return false; + } finally { + if (channel != null) { + channel.shutdownNow(); + } + } + } +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/DirectAccessChecker.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/DirectAccessChecker.java new file mode 100644 index 0000000000..beab21e8d4 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/dp/DirectAccessChecker.java @@ -0,0 +1,34 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.dp; + +import com.google.api.core.InternalApi; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; +import io.grpc.ManagedChannel; +import java.util.function.Supplier; +import javax.annotation.Nullable; + +@InternalApi +/* Evaluates whether a given channel supports Direct Access. */ +public interface DirectAccessChecker { + /** + * Evaluates if Direct Access is available by creating a test channel. + * + * @param supplier A supplier to create maybe direct access channel + * @return true if the channel is eligible for Direct Access + */ + boolean check(Supplier supplier, @Nullable DirectPathCompatibleTracer tracer); +} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer.java index e770e04ccb..bcec4c5403 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelPrimer.java @@ -26,9 +26,9 @@ import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer; import io.grpc.CallCredentials; import io.grpc.CallOptions; +import io.grpc.Channel; import io.grpc.ClientCall; import io.grpc.Deadline; -import io.grpc.ManagedChannel; import io.grpc.Metadata; import io.grpc.Status; import io.grpc.auth.MoreCallCredentials; @@ -88,21 +88,21 @@ static BigtableChannelPrimer create( } @Override - public void primeChannel(ManagedChannel managedChannel) { + public void primeChannel(Channel channel) { try { - primeChannelUnsafe(managedChannel); + primeChannelUnsafe(channel); } catch (IOException | RuntimeException e) { LOG.log(Level.WARNING, "Unexpected error while trying to prime a channel", e); } } - private void primeChannelUnsafe(ManagedChannel managedChannel) throws IOException { - sendPrimeRequestsBlocking(managedChannel); + private void primeChannelUnsafe(Channel channel) throws IOException { + sendPrimeRequestsBlocking(channel); } - private void sendPrimeRequestsBlocking(ManagedChannel managedChannel) { + private void sendPrimeRequestsBlocking(Channel channel) { try { - sendPrimeRequestsAsync(managedChannel).get(1, TimeUnit.MINUTES); + sendPrimeRequestsAsync(channel).get(1, TimeUnit.MINUTES); } catch (Throwable e) { // TODO: Not sure if we should swallow the error here. We are pre-emptively swapping // channels if the new @@ -112,7 +112,7 @@ private void sendPrimeRequestsBlocking(ManagedChannel managedChannel) { } @Override - public ApiFuture sendPrimeRequestsAsync(ManagedChannel managedChannel) { + public ApiFuture sendPrimeRequestsAsync(Channel managedChannel) { ClientCall clientCall = managedChannel.newCall( BigtableGrpc.getPingAndWarmMethod(), diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelSupplier.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelSupplier.java new file mode 100644 index 0000000000..24aa062e14 --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableChannelSupplier.java @@ -0,0 +1,23 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.stub; + +import com.google.api.core.InternalApi; +import io.grpc.ManagedChannel; +import java.util.function.Supplier; + +@InternalApi +public interface BigtableChannelSupplier extends Supplier {} diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java index 47bb9ccff9..2680558995 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/BigtableClientContext.java @@ -166,7 +166,8 @@ public static BigtableClientContext create( transportProvider.build(), channelPrimer, metrics.getChannelPoolMetricsTracer(), - backgroundExecutor); + backgroundExecutor, + metrics.getDirectPathCompatibleTracer()); builder.setTransportChannelProvider(btTransportProvider); } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java index db33e93fec..b7189ecc04 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java @@ -242,6 +242,25 @@ public ClientOperationSettings getPerOpSettings() { return perOpSettings; } + /** Applies common pool, message size, and keep-alive settings to the provided builder. */ + private static InstantiatingGrpcChannelProvider.Builder commonTraits( + InstantiatingGrpcChannelProvider.Builder builder) { + return builder + .setChannelPoolSettings( + ChannelPoolSettings.builder() + .setInitialChannelCount(10) + .setMinRpcsPerChannel(1) + // Keep it conservative as we scale the channel size every 1min + // and delta is 2 channels. + .setMaxRpcsPerChannel(25) + .setPreemptiveRefreshEnabled(true) + .build()) + .setMaxInboundMessageSize(MAX_MESSAGE_SIZE) + .setKeepAliveTime(Duration.ofSeconds(30)) // sends ping in this interval + .setKeepAliveTimeout( + Duration.ofSeconds(10)); // wait this long before considering the connection dead + } + /** Returns a builder for the default ChannelProvider for this service. */ public static InstantiatingGrpcChannelProvider.Builder defaultGrpcTransportProviderBuilder() { InstantiatingGrpcChannelProvider.Builder grpcTransportProviderBuilder = @@ -261,20 +280,24 @@ public static InstantiatingGrpcChannelProvider.Builder defaultGrpcTransportProvi Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS)); } } - return grpcTransportProviderBuilder - .setChannelPoolSettings( - ChannelPoolSettings.builder() - .setInitialChannelCount(10) - .setMinRpcsPerChannel(1) - // Keep it conservative as we scale the channel size every 1min - // and delta is 2 channels. - .setMaxRpcsPerChannel(25) - .setPreemptiveRefreshEnabled(true) - .build()) - .setMaxInboundMessageSize(MAX_MESSAGE_SIZE) - .setKeepAliveTime(Duration.ofSeconds(30)) // sends ping in this interval - .setKeepAliveTimeout( - Duration.ofSeconds(10)); // wait this long before considering the connection dead + return commonTraits(grpcTransportProviderBuilder); + } + + /** Applies Direct Access traits (DirectPath & ALTS) to an existing builder. */ + public static InstantiatingGrpcChannelProvider.Builder applyDirectAccessTraits( + InstantiatingGrpcChannelProvider.Builder builder) { + + builder + .setAttemptDirectPathXds() + .setAttemptDirectPath(true) + .setAllowNonDefaultServiceAccount(true); + + if (!DIRECT_PATH_BOUND_TOKEN_DISABLED) { + builder.setAllowHardBoundTokenTypes( + Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS)); + } + + return builder; } @SuppressWarnings("WeakerAccess") @@ -610,6 +633,12 @@ private Builder() { perOpSettings = new ClientOperationSettings.Builder(); + // Note: RouteLookup evaluates and returns directpath targets + // only if Traffic Director sends the request (with grpc as target type) + // For GFE/CFE, sending setDirectAccessRequested + // is fine as GFE/CFE sends with gslb target type + // TODO: flip the bit setDirectAccessRequested and setTrafficDirectorEnabled once we make + // client compatible by default. featureFlags = FeatureFlags.newBuilder() .setReverseScans(true) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/MetadataExtractorInterceptor.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/MetadataExtractorInterceptor.java index d38f164fc6..f2d3a1502f 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/MetadataExtractorInterceptor.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/MetadataExtractorInterceptor.java @@ -29,10 +29,15 @@ import io.grpc.ClientInterceptors; import io.grpc.ForwardingClientCall; import io.grpc.ForwardingClientCallListener; +import io.grpc.Grpc; import io.grpc.Metadata; import io.grpc.MethodDescriptor; import io.grpc.Status; import io.grpc.alts.AltsContextUtil; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.time.Duration; import java.util.Base64; import java.util.regex.Matcher; @@ -85,6 +90,12 @@ public SidebandData getSidebandData() { } public static class SidebandData { + public enum IpProtocol { + IPV4, + IPV6, + UNKNOWN + } + private static final CallOptions.Key KEY = CallOptions.Key.create("bigtable-sideband"); @@ -105,6 +116,7 @@ public static SidebandData from(CallOptions callOptions) { @Nullable private volatile ResponseParams responseParams; @Nullable private volatile PeerInfo peerInfo; @Nullable private volatile Duration gfeTiming; + @Nullable private volatile IpProtocol ipProtocol; @Nullable public ResponseParams getResponseParams() { @@ -121,16 +133,23 @@ public Duration getGfeTiming() { return gfeTiming; } + @Nullable + public IpProtocol getIpProtocol() { + return ipProtocol; + } + private void reset() { responseParams = null; peerInfo = null; gfeTiming = null; + ipProtocol = IpProtocol.UNKNOWN; } void onResponseHeaders(Metadata md, Attributes attributes) { responseParams = extractResponseParams(md); gfeTiming = extractGfeLatency(md); peerInfo = extractPeerInfo(md, gfeTiming, attributes); + ipProtocol = extractIpProtocol(attributes); } void onClose(Status status, Metadata trailers) { @@ -139,6 +158,20 @@ void onClose(Status status, Metadata trailers) { } } + @Nullable + private static IpProtocol extractIpProtocol(Attributes attributes) { + SocketAddress remoteAddr = attributes.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR); + if (remoteAddr instanceof InetSocketAddress) { + InetSocketAddress inetAddr = (InetSocketAddress) remoteAddr; + if (inetAddr.getAddress() instanceof Inet4Address) { + return IpProtocol.IPV4; + } else if (inetAddr.getAddress() instanceof Inet6Address) { + return IpProtocol.IPV6; + } + } + return IpProtocol.UNKNOWN; + } + @Nullable private static Duration extractGfeLatency(Metadata metadata) { String serverTiming = metadata.get(SERVER_TIMING_HEADER_KEY); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/NoOpChannelPrimer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/NoOpChannelPrimer.java index 3cb98d9dee..86c564bcc0 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/NoOpChannelPrimer.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/NoOpChannelPrimer.java @@ -16,11 +16,11 @@ package com.google.cloud.bigtable.data.v2.stub; import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; import com.google.api.core.InternalApi; -import com.google.api.core.SettableApiFuture; import com.google.bigtable.v2.PingAndWarmResponse; import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer; -import io.grpc.ManagedChannel; +import io.grpc.Channel; @InternalApi public class NoOpChannelPrimer implements ChannelPrimer { @@ -31,14 +31,12 @@ static NoOpChannelPrimer create() { private NoOpChannelPrimer() {} @Override - public void primeChannel(ManagedChannel channel) { + public void primeChannel(Channel channel) { // No op } @Override - public ApiFuture sendPrimeRequestsAsync(ManagedChannel channel) { - SettableApiFuture future = SettableApiFuture.create(); - future.set(PingAndWarmResponse.getDefaultInstance()); - return future; + public ApiFuture sendPrimeRequestsAsync(Channel channel) { + return ApiFutures.immediateFuture(PingAndWarmResponse.getDefaultInstance()); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java index 188dc83ac0..4a127a5502 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPool.java @@ -16,7 +16,6 @@ package com.google.cloud.bigtable.gaxx.grpc; import com.google.api.core.InternalApi; -import com.google.api.gax.grpc.ChannelFactory; import com.google.bigtable.v2.PeerInfo; import com.google.cloud.bigtable.data.v2.stub.MetadataExtractorInterceptor; import com.google.cloud.bigtable.gaxx.grpc.ChannelPoolHealthChecker.ProbeResult; @@ -67,7 +66,7 @@ public class BigtableChannelPool extends ManagedChannel implements BigtableChann private static final java.time.Duration REFRESH_PERIOD = java.time.Duration.ofMinutes(50); private final BigtableChannelPoolSettings settings; - private final ChannelFactory channelFactory; + private final Supplier channelSupplier; private final ChannelPrimer channelPrimer; private final Object entryWriteLock = new Object(); @@ -81,11 +80,11 @@ public class BigtableChannelPool extends ManagedChannel implements BigtableChann public static BigtableChannelPool create( BigtableChannelPoolSettings settings, - ChannelFactory channelFactory, + Supplier channelSupplier, ChannelPrimer channelPrimer, ScheduledExecutorService backgroundExecutor) throws IOException { - return new BigtableChannelPool(settings, channelFactory, channelPrimer, backgroundExecutor); + return new BigtableChannelPool(settings, channelSupplier, channelPrimer, backgroundExecutor); } /** @@ -98,12 +97,12 @@ public static BigtableChannelPool create( @VisibleForTesting BigtableChannelPool( BigtableChannelPoolSettings settings, - ChannelFactory channelFactory, + Supplier channelSupplier, ChannelPrimer channelPrimer, ScheduledExecutorService executor) throws IOException { this.settings = settings; - this.channelFactory = channelFactory; + this.channelSupplier = channelSupplier; this.channelPrimer = channelPrimer; Clock systemClock = Clock.systemUTC(); ChannelPoolHealthChecker channelPoolHealthChecker = @@ -113,7 +112,7 @@ public static BigtableChannelPool create( ImmutableList.Builder initialListBuilder = ImmutableList.builder(); for (int i = 0; i < settings.getInitialChannelCount(); i++) { - ManagedChannel newChannel = channelFactory.createSingleChannel(); + ManagedChannel newChannel = channelSupplier.get(); channelPrimer.primeChannel(newChannel); initialListBuilder.add(new Entry(newChannel)); } @@ -419,10 +418,10 @@ private void expand(int desiredSize) { for (int i = 0; i < desiredSize - localEntries.size(); i++) { try { - ManagedChannel newChannel = channelFactory.createSingleChannel(); + ManagedChannel newChannel = channelSupplier.get(); this.channelPrimer.primeChannel(newChannel); newEntries.add(new Entry(newChannel)); - } catch (IOException e) { + } catch (Exception e) { LOG.log(Level.WARNING, "Failed to add channel", e); } } @@ -459,10 +458,10 @@ void refresh() { for (int i = 0; i < newEntries.size(); i++) { try { - ManagedChannel newChannel = channelFactory.createSingleChannel(); + ManagedChannel newChannel = channelSupplier.get(); this.channelPrimer.primeChannel(newChannel); newEntries.set(i, new Entry(newChannel)); - } catch (IOException e) { + } catch (Exception e) { LOG.log(Level.WARNING, "Failed to refresh channel, leaving old channel", e); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java index b75f145108..0b7c44cff4 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/BigtableTransportChannelProvider.java @@ -16,7 +16,6 @@ package com.google.cloud.bigtable.gaxx.grpc; import com.google.api.core.InternalApi; -import com.google.api.gax.grpc.ChannelFactory; import com.google.api.gax.grpc.ChannelPoolSettings; import com.google.api.gax.grpc.GrpcTransportChannel; import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider; @@ -24,12 +23,20 @@ import com.google.api.gax.rpc.TransportChannelProvider; import com.google.auth.Credentials; import com.google.cloud.bigtable.data.v2.internal.csm.tracers.ChannelPoolMetricsTracer; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; +import com.google.cloud.bigtable.data.v2.internal.dp.ClassicDirectAccessChecker; +import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessChecker; +import com.google.cloud.bigtable.data.v2.stub.BigtableChannelSupplier; +import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings; import com.google.common.base.Preconditions; import io.grpc.ManagedChannel; import java.io.IOException; import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.ScheduledExecutorService; +import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.annotation.Nullable; /** @@ -38,20 +45,25 @@ */ @InternalApi public final class BigtableTransportChannelProvider implements TransportChannelProvider { + private static final Logger LOG = + Logger.getLogger(BigtableTransportChannelProvider.class.getName()); private final InstantiatingGrpcChannelProvider delegate; private final ChannelPrimer channelPrimer; @Nullable private final ChannelPoolMetricsTracer channelPoolMetricsTracer; @Nullable private final ScheduledExecutorService backgroundExecutor; + DirectPathCompatibleTracer directPathCompatibleTracer; private BigtableTransportChannelProvider( InstantiatingGrpcChannelProvider instantiatingGrpcChannelProvider, ChannelPrimer channelPrimer, @Nullable ChannelPoolMetricsTracer channelPoolMetricsTracer, - @Nullable ScheduledExecutorService backgroundExecutor) { + @Nullable ScheduledExecutorService backgroundExecutor, + DirectPathCompatibleTracer directPathCompatibleTracer) { delegate = Preconditions.checkNotNull(instantiatingGrpcChannelProvider); this.channelPrimer = channelPrimer; this.channelPoolMetricsTracer = channelPoolMetricsTracer; this.backgroundExecutor = backgroundExecutor; + this.directPathCompatibleTracer = directPathCompatibleTracer; } @Override @@ -78,7 +90,11 @@ public BigtableTransportChannelProvider withExecutor(Executor executor) { InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withExecutor(executor); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, backgroundExecutor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + backgroundExecutor, + directPathCompatibleTracer); } @Override @@ -91,7 +107,11 @@ public TransportChannelProvider withBackgroundExecutor(ScheduledExecutorService InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withBackgroundExecutor(executor); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, executor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + executor, + directPathCompatibleTracer); } @Override @@ -104,7 +124,11 @@ public BigtableTransportChannelProvider withHeaders(Map headers) InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withHeaders(headers); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, backgroundExecutor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + backgroundExecutor, + directPathCompatibleTracer); } @Override @@ -117,7 +141,11 @@ public TransportChannelProvider withEndpoint(String endpoint) { InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withEndpoint(endpoint); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, backgroundExecutor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + backgroundExecutor, + directPathCompatibleTracer); } @Deprecated @@ -132,12 +160,52 @@ public TransportChannelProvider withPoolSize(int size) { InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withPoolSize(size); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, backgroundExecutor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + backgroundExecutor, + directPathCompatibleTracer); } + // We need this for direct access checker. + /** Expected to only be called once when BigtableClientContext is created */ @Override public TransportChannel getTransportChannel() throws IOException { + InstantiatingGrpcChannelProvider directAccessProvider = + EnhancedBigtableStubSettings.applyDirectAccessTraits(delegate.toBuilder()) + .setChannelPoolSettings(ChannelPoolSettings.staticallySized(1)) + .build(); + + Supplier maybeDirectAccessChannelFactory = + () -> { + try { + GrpcTransportChannel channel = + (GrpcTransportChannel) directAccessProvider.getTransportChannel(); + return (ManagedChannel) channel.getChannel(); + } catch (IOException e) { + throw new java.io.UncheckedIOException(e); + } + }; + + DirectAccessChecker directAccessChecker = ClassicDirectAccessChecker.create(channelPrimer); + boolean isDirectAccessEligible = false; + + try { + isDirectAccessEligible = + directAccessChecker.check(maybeDirectAccessChannelFactory, directPathCompatibleTracer); + } catch (Exception e) { + LOG.log(Level.FINE, "Client is not direct access eligible, using standard transport.", e); + } + + InstantiatingGrpcChannelProvider selectedProvider; + + if (isDirectAccessEligible) { + selectedProvider = directAccessProvider; + } else { + selectedProvider = delegate; + } + // This provider's main purpose is to replace the default GAX ChannelPool // with a custom BigtableChannelPool, reusing the delegate's configuration. @@ -145,9 +213,11 @@ public TransportChannel getTransportChannel() throws IOException { // We achieve this by configuring our delegate to not use its own pooling // (by setting pool size to 1) and then calling getTransportChannel() on it. InstantiatingGrpcChannelProvider singleChannelProvider = - delegate.toBuilder().setChannelPoolSettings(ChannelPoolSettings.staticallySized(1)).build(); + selectedProvider.toBuilder() + .setChannelPoolSettings(ChannelPoolSettings.staticallySized(1)) + .build(); - ChannelFactory channelFactory = + BigtableChannelSupplier channelFactory = () -> { try { GrpcTransportChannel channel = @@ -189,7 +259,11 @@ public TransportChannelProvider withCredentials(Credentials credentials) { InstantiatingGrpcChannelProvider newChannelProvider = (InstantiatingGrpcChannelProvider) delegate.withCredentials(credentials); return new BigtableTransportChannelProvider( - newChannelProvider, channelPrimer, channelPoolMetricsTracer, backgroundExecutor); + newChannelProvider, + channelPrimer, + channelPoolMetricsTracer, + backgroundExecutor, + directPathCompatibleTracer); } /** Creates a BigtableTransportChannelProvider. */ @@ -197,11 +271,13 @@ public static BigtableTransportChannelProvider create( InstantiatingGrpcChannelProvider instantiatingGrpcChannelProvider, ChannelPrimer channelPrimer, ChannelPoolMetricsTracer outstandingRpcsMetricTracker, - ScheduledExecutorService backgroundExecutor) { + ScheduledExecutorService backgroundExecutor, + DirectPathCompatibleTracer directPathCompatibleTracer) { return new BigtableTransportChannelProvider( instantiatingGrpcChannelProvider, channelPrimer, outstandingRpcsMetricTracker, - backgroundExecutor); + backgroundExecutor, + directPathCompatibleTracer); } } diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelPrimer.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelPrimer.java index ea7cc70175..b9fb28bcb6 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelPrimer.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/gaxx/grpc/ChannelPrimer.java @@ -18,11 +18,22 @@ import com.google.api.core.ApiFuture; import com.google.api.core.InternalApi; import com.google.bigtable.v2.PingAndWarmResponse; +import io.grpc.Channel; import io.grpc.ManagedChannel; @InternalApi("For internal use by google-cloud-java clients only") public interface ChannelPrimer { - void primeChannel(ManagedChannel channel); + @Deprecated + default void primeChannel(ManagedChannel channel) { + primeChannel((Channel) channel); + } - ApiFuture sendPrimeRequestsAsync(ManagedChannel channel); + void primeChannel(Channel channel); + + @Deprecated + default ApiFuture sendPrimeRequestsAsync(ManagedChannel channel) { + return sendPrimeRequestsAsync((Channel) channel); + } + + ApiFuture sendPrimeRequestsAsync(Channel channel); } diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactoryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactoryTest.java index f5c9472b47..cbe6d45457 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactoryTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientFactoryTest.java @@ -285,8 +285,9 @@ public void testCreateWithRefreshingChannel() throws Exception { Mockito.verify(credentialsProvider, Mockito.times(2)).getCredentials(); Mockito.verify(executorProvider, Mockito.times(1)).getExecutor(); Mockito.verify(watchdogProvider, Mockito.times(1)).getWatchdog(); - - assertThat(warmedChannels).hasSize(poolSize); + // +1 accounts for the temporary channel created by UnaryDirectAccessChecker + int expectedChannelCount = poolSize + 1; + assertThat(warmedChannels).hasSize(expectedChannelCount); assertThat(warmedChannels.values()).doesNotContain(false); // Wait for all the connections to close asynchronously @@ -294,7 +295,7 @@ public void testCreateWithRefreshingChannel() throws Exception { long sleepTimeMs = 1000; Thread.sleep(sleepTimeMs); // Verify that all the channels are closed - assertThat(terminateAttributes).hasSize(poolSize); + assertThat(terminateAttributes).hasSize(expectedChannelCount); } @Test diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessCheckerTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessCheckerTest.java new file mode 100644 index 0000000000..39fcfb6a00 --- /dev/null +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/dp/ClassicDirectAccessCheckerTest.java @@ -0,0 +1,122 @@ +/* + * 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 + * + * https://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 com.google.cloud.bigtable.data.v2.internal.dp; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +import com.google.bigtable.v2.PeerInfo; +import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer; +import com.google.cloud.bigtable.data.v2.stub.MetadataExtractorInterceptor; +import com.google.cloud.bigtable.gaxx.grpc.ChannelPrimer; +import io.grpc.Channel; +import io.grpc.ManagedChannel; +import java.util.function.Supplier; +import org.junit.Before; +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; + +@RunWith(JUnit4.class) +public class ClassicDirectAccessCheckerTest { + + @Rule public final MockitoRule mockito = MockitoJUnit.rule(); + + @Mock private ChannelPrimer mockChannelPrimer; + @Mock private Supplier mockChannelFactory; + @Mock private DirectPathCompatibleTracer mockTracer; + @Mock private ManagedChannel mockChannel; + @Mock private MetadataExtractorInterceptor mockInterceptor; + @Mock private MetadataExtractorInterceptor.SidebandData mockSidebandData; + + private ClassicDirectAccessChecker checker; + + @Before + public void setUp() throws Exception { + // 1. Create a SPY of the checker so we can override just one method + checker = spy(ClassicDirectAccessChecker.create(mockChannelPrimer)); + + // 2. Tell the spy to return our mock interceptor instead of calling new() + doReturn(mockInterceptor).when(checker).createInterceptor(); + + // 3. Setup standard mocks + when(mockChannelFactory.get()).thenReturn(mockChannel); + when(mockInterceptor.getSidebandData()).thenReturn(mockSidebandData); + } + + @Test + public void testEligibleForDirectAccess() { + PeerInfo peerInfo = + PeerInfo.newBuilder() + .setTransportType(PeerInfo.TransportType.TRANSPORT_TYPE_DIRECT_ACCESS) + .build(); + when(mockSidebandData.getPeerInfo()).thenReturn(peerInfo); + when(mockSidebandData.getIpProtocol()) + .thenReturn(MetadataExtractorInterceptor.SidebandData.IpProtocol.IPV6); + + boolean isEligible = checker.check(mockChannelFactory, mockTracer); + + assertThat(isEligible).isTrue(); + verify(mockChannelPrimer).primeChannel(any(Channel.class)); + verify(mockTracer).recordSuccess("ipv6"); + verify(mockChannel).shutdownNow(); + } + + @Test + public void testNotEligibleProxiedRouting() { + PeerInfo peerInfo = + PeerInfo.newBuilder() + .setTransportType(PeerInfo.TransportType.TRANSPORT_TYPE_CLOUD_PATH) + .build(); + when(mockSidebandData.getPeerInfo()).thenReturn(peerInfo); + + boolean isEligible = checker.check(mockChannelFactory, mockTracer); + + assertThat(isEligible).isFalse(); + verifyNoInteractions(mockTracer); + verify(mockChannel).shutdownNow(); + } + + @Test + public void testMissingSidebandData() { + // Override the Before setup to return null for this specific test + when(mockInterceptor.getSidebandData()).thenReturn(null); + + boolean isEligible = checker.check(mockChannelFactory, mockTracer); + + assertThat(isEligible).isFalse(); + verifyNoInteractions(mockTracer); + verify(mockChannel).shutdownNow(); + } + + @Test + public void testExceptionSafetyAndCleanup() { + doThrow(new RuntimeException("Simulated primer failure")) + .when(mockChannelPrimer) + .primeChannel(any(Channel.class)); + + boolean isEligible = checker.check(mockChannelFactory, mockTracer); + + assertThat(isEligible).isFalse(); + verifyNoInteractions(mockTracer); + verify(mockChannel).shutdownNow(); + } +} diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubTest.java index a9eda04356..1037aa1c15 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubTest.java @@ -552,7 +552,8 @@ public void testChannelPrimerConfigured() throws IOException { defaultSettings.toBuilder().setRefreshingChannel(true).build(); try (EnhancedBigtableStub ignored = EnhancedBigtableStub.create(settings)) { - assertThat(fakeDataService.pingRequests).hasSize(1); + // direct access checker ping + assertThat(fakeDataService.pingRequests).hasSize(2); } } diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolTest.java index d1059c0362..6719a70d89 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/gaxx/grpc/BigtableChannelPoolTest.java @@ -19,7 +19,6 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; -import com.google.api.gax.grpc.ChannelFactory; import com.google.common.collect.Iterables; import io.grpc.CallOptions; import io.grpc.ClientCall; @@ -32,6 +31,7 @@ import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.function.Supplier; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -48,7 +48,7 @@ public class BigtableChannelPoolTest { @Rule public final MockitoRule mockito = MockitoJUnit.rule(); - @Mock private ChannelFactory mockChannelFactory; + @Mock private Supplier mockChannelFactory; @Mock private ChannelPrimer mockChannelPrimer; @Mock private ManagedChannel mockChannel; @Mock private ClientCall mockClientCall; @@ -75,7 +75,7 @@ public String parse(InputStream stream) { @Before public void setUp() throws IOException { - when(mockChannelFactory.createSingleChannel()).thenReturn(mockChannel); + when(mockChannelFactory.get()).thenReturn(mockChannel); when(mockChannel.newCall( ArgumentMatchers.>any(), any(CallOptions.class))) .thenReturn(mockClientCall);