Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,9 @@
import com.google.bigtable.v2.GetClientConfigurationRequest;
import com.google.bigtable.v2.PeerInfo;
import com.google.bigtable.v2.PeerInfo.TransportType;
import com.google.bigtable.v2.Row;
import com.google.bigtable.v2.SessionReadModifyWriteRowResponse;
import com.google.bigtable.v2.SessionReadRowResponse;
import com.google.cloud.bigtable.data.v2.internal.api.ChannelProviders.ChannelProvider;
import com.google.cloud.bigtable.data.v2.internal.api.ChannelProviders.ConfiguredChannelProvider;
import com.google.cloud.bigtable.data.v2.internal.api.Client;
Expand All @@ -47,6 +50,7 @@
import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DirectPathCompatibleTracer;
import com.google.cloud.bigtable.data.v2.internal.dp.DirectAccessInvestigator;
import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
import com.google.cloud.bigtable.data.v2.internal.util.SessionScope;
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
import com.google.cloud.bigtable.data.v2.models.Query;
import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
Expand All @@ -69,6 +73,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
Expand All @@ -82,6 +87,19 @@ public class ShimImpl implements Shim {

private static final Duration DA_CHECK_TIMEOUT = Duration.ofSeconds(5);

// A ReadRow on a missing key legitimately has no row, and the classic path surfaces that as a
// null Row.
@VisibleForTesting
static final Function<SessionReadRowResponse, Row> READ_ROW_EXTRACTOR =
r -> r.hasRow() ? r.getRow() : null;

// ReadModifyWriteRow always mutates and so always yields a row. The classic path builds a Row
// from the (possibly unset) response row rather than returning null, so extract unconditionally
// to keep both paths non-null.
@VisibleForTesting
static final Function<SessionReadModifyWriteRowResponse, Row> READ_MODIFY_WRITE_ROW_EXTRACTOR =
SessionReadModifyWriteRowResponse::getRow;

private final ClientConfigurationManager configManager;
private final Resource<ClientConfigurationManager> configManagerResource;
private final Client client;
Expand Down Expand Up @@ -374,11 +392,10 @@ public <RowT> UnaryCallable<Query, RowT> decorateReadRow(
return new DivertingUnaryCallable<>(
configManager,
classic,
new RowBuilderShim<>(readRowShim, rowAdapter, r -> r.hasRow() ? r.getRow() : null),
new RowBuilderShim<>(readRowShim, rowAdapter, READ_ROW_EXTRACTOR),
Util.extractTimeout(settings),
debugTagTracer,
"method:Bigtable.ReadRow",
/* fallbackToGlobalSessionLoad= */ true);
SessionScope.READ_ROW);
}

@Override
Expand All @@ -390,8 +407,7 @@ public UnaryCallable<RowMutation, Void> decorateMutateRow(
mutateRowShim,
Util.extractTimeout(settings),
debugTagTracer,
"method:Bigtable.MutateRow",
/* fallbackToGlobalSessionLoad= */ true);
SessionScope.MUTATE_ROW);
}

@Override
Expand All @@ -403,8 +419,7 @@ public UnaryCallable<ConditionalRowMutation, Boolean> decorateCheckAndMutateRow(
checkAndMutateRowShim,
Util.extractTimeout(settings),
debugTagTracer,
"method:Bigtable.CheckAndMutateRow",
/* fallbackToGlobalSessionLoad= */ false);
SessionScope.CHECK_AND_MUTATE_ROW);
}

@Override
Expand All @@ -415,11 +430,9 @@ public <RowT> UnaryCallable<ReadModifyWriteRow, RowT> decorateReadModifyWriteRow
return new DivertingUnaryCallable<>(
configManager,
classic,
new RowBuilderShim<>(
readModifyWriteRowShim, rowAdapter, r -> r.hasRow() ? r.getRow() : null),
new RowBuilderShim<>(readModifyWriteRowShim, rowAdapter, READ_MODIFY_WRITE_ROW_EXTRACTOR),
Util.extractTimeout(settings),
debugTagTracer,
"method:Bigtable.ReadModifyWriteRow",
/* fallbackToGlobalSessionLoad= */ false);
SessionScope.READ_MODIFY_WRITE_ROW);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,14 +22,12 @@
import com.google.api.gax.rpc.ApiException;
import com.google.api.gax.rpc.ApiExceptionFactory;
import com.google.api.gax.rpc.UnaryCallable;
import com.google.bigtable.v2.SessionClientConfiguration;
import com.google.bigtable.v2.SessionDiversionConfiguration;
import com.google.bigtable.v2.SessionScopeDiversionConfiguration;
import com.google.bigtable.v2.TelemetryConfiguration;
import com.google.cloud.bigtable.data.v2.internal.compat.FutureAdapter;
import com.google.cloud.bigtable.data.v2.internal.compat.Util;
import com.google.cloud.bigtable.data.v2.internal.csm.tracers.DebugTagTracer;
import com.google.cloud.bigtable.data.v2.internal.util.ClientConfigurationManager;
import com.google.cloud.bigtable.data.v2.internal.util.SessionScope;
import com.google.common.base.Throwables;
import io.grpc.Context;
import io.grpc.Context.CancellableContext;
Expand All @@ -52,28 +50,22 @@ public class DivertingUnaryCallable<ReqT, RespT> extends UnaryCallable<ReqT, Res

private final Duration defaultTimeout;
private final DebugTagTracer debugTagTracer;
// Scope key for per-method diversion lookup, e.g. "method:Bigtable.ReadRow".
private final String methodScopeKey;
// When true, falls back to the flat SessionClientConfiguration.session_load when no
// SessionDiversionConfiguration is present. Used for ReadRow/MutateRow during the server
// rollout window before the server starts sending SessionDiversionConfiguration.
private final boolean fallbackToGlobalSessionLoad;
// The diversion scope this callable's method belongs to.
private final SessionScope scope;

public DivertingUnaryCallable(
ClientConfigurationManager configurationManager,
UnaryCallable<ReqT, RespT> classic,
UnaryShim<ReqT, RespT> experimental,
Duration defaultTimeout,
DebugTagTracer debugTagTracer,
String methodScopeKey,
boolean fallbackToGlobalSessionLoad) {
SessionScope scope) {
this.configurationManager = configurationManager;
this.classic = classic;
this.experimental = experimental;
this.defaultTimeout = defaultTimeout;
this.debugTagTracer = debugTagTracer;
this.methodScopeKey = methodScopeKey;
this.fallbackToGlobalSessionLoad = fallbackToGlobalSessionLoad;
this.scope = scope;
}

@Override
Expand Down Expand Up @@ -138,25 +130,8 @@ private boolean useExperimental(ReqT req) {
}

private float resolveSessionLoad() {
SessionClientConfiguration sessionConfig =
configurationManager.getClientConfiguration().getSessionConfiguration();

if (sessionConfig.hasSessionDiversionConfiguration()) {
SessionDiversionConfiguration diversionConfig =
sessionConfig.getSessionDiversionConfiguration();
switch (diversionConfig.getDiversionStrategyCase()) {
case GLOBAL_DIVERSION:
return diversionConfig.getGlobalDiversion().getSessionLoad();
case PER_SCOPE_DIVERSION:
SessionScopeDiversionConfiguration scopeConfig =
diversionConfig.getPerScopeDiversion().getScopeDiversionsMap().get(methodScopeKey);
return scopeConfig != null ? scopeConfig.getSessionLoad() : 0f;
default:
return 0f;
}
}

return fallbackToGlobalSessionLoad ? sessionConfig.getSessionLoad() : 0f;
return scope.resolveSessionLoad(
configurationManager.getClientConfiguration().getSessionConfiguration());
}

ApiException translateException(Throwable e) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import com.google.bigtable.v2.GetClientConfigurationRequest;
import com.google.bigtable.v2.SessionClientConfiguration;
import com.google.bigtable.v2.SessionClientConfiguration.ChannelPoolConfiguration.DirectAccessOnly;
import com.google.bigtable.v2.SessionDiversionConfiguration;
import com.google.bigtable.v2.TelemetryConfiguration;
import com.google.cloud.bigtable.data.v2.internal.api.ChannelProviders.ChannelProvider;
import com.google.cloud.bigtable.data.v2.internal.api.Util;
Expand Down Expand Up @@ -458,24 +457,12 @@ private ClientConfiguration normalizeConfig(ClientConfiguration cfg) {
}

/**
* Returns true if the configuration routes a nonzero share of traffic to sessions, either via the
* flat {@code session_load} or via a diversion configuration that carries a nonzero load.
* Returns true if the configuration routes a nonzero share of traffic to sessions for any scope
* this client serves. Shares the resolution rule with the per-RPC diversion decision so the two
* cannot disagree; see {@link SessionScope}.
*/
private static boolean divertsAnyTrafficToSessions(SessionClientConfiguration session) {
if (session.getSessionLoad() > 0) {
return true;
}

SessionDiversionConfiguration diversion = session.getSessionDiversionConfiguration();
switch (diversion.getDiversionStrategyCase()) {
case GLOBAL_DIVERSION:
return diversion.getGlobalDiversion().getSessionLoad() > 0;
case PER_SCOPE_DIVERSION:
return diversion.getPerScopeDiversion().getScopeDiversionsMap().values().stream()
.anyMatch(scope -> scope.getSessionLoad() > 0);
default:
return false;
}
return SessionScope.divertsAnyTraffic(session);
}

public boolean areSessionsRequired() {
Expand Down
Original file line number Diff line number Diff line change
@@ -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.util;

import com.google.bigtable.v2.SessionClientConfiguration;
import com.google.bigtable.v2.SessionDiversionConfiguration;
import com.google.bigtable.v2.SessionScopeDiversionConfiguration;

/**
* The per-method diversion scopes this client implements, along with the single rule for resolving
* a session load for each of them.
*
* <p>Two decisions depend on this rule: the per-RPC fork in {@code DivertingUnaryCallable}, and the
* client wide "are sessions in use at all" decision in {@link ClientConfigurationManager}. They
* must agree. If the client-wide check is the more permissive of the two, the client keeps session
* pools alive and advertises session feature flags while never diverting a single request. Keeping
* the scope list and the resolution rule in one place is what keeps them in sync.
*/
public enum SessionScope {
READ_ROW("method:Bigtable.ReadRow", /* fallbackToGlobalSessionLoad= */ true),
MUTATE_ROW("method:Bigtable.MutateRow", /* fallbackToGlobalSessionLoad= */ true),
CHECK_AND_MUTATE_ROW(
"method:Bigtable.CheckAndMutateRow", /* fallbackToGlobalSessionLoad= */ false),
READ_MODIFY_WRITE_ROW(
"method:Bigtable.ReadModifyWriteRow", /* fallbackToGlobalSessionLoad= */ false);

private final String key;
private final boolean fallbackToGlobalSessionLoad;

SessionScope(String key, boolean fallbackToGlobalSessionLoad) {
this.key = key;
this.fallbackToGlobalSessionLoad = fallbackToGlobalSessionLoad;
}

/** The {@code method:<service>.<method>} key the server uses to address this scope. */
public String getKey() {
return key;
}

/**
* Returns the fraction of this scope's traffic that should be served by sessions.
*
* <p>A {@link SessionDiversionConfiguration} is authoritative when present: the server only sends
* the deprecated flat {@code session_load} to keep ReadRow and MutateRow working on older
* clients, so this client ignores it entirely once a diversion configuration arrives.
*/
@SuppressWarnings("deprecation")
public float resolveSessionLoad(SessionClientConfiguration sessionConfig) {
if (sessionConfig.hasSessionDiversionConfiguration()) {
SessionDiversionConfiguration diversionConfig =
sessionConfig.getSessionDiversionConfiguration();
switch (diversionConfig.getDiversionStrategyCase()) {
case GLOBAL_DIVERSION:
return diversionConfig.getGlobalDiversion().getSessionLoad();
case PER_SCOPE_DIVERSION:
SessionScopeDiversionConfiguration scopeConfig =
diversionConfig.getPerScopeDiversion().getScopeDiversionsMap().get(key);
return scopeConfig != null ? scopeConfig.getSessionLoad() : 0f;
default:
return 0f;
}
}

return fallbackToGlobalSessionLoad ? sessionConfig.getSessionLoad() : 0f;
}

/**
* Returns true if the configuration routes a nonzero share of traffic to sessions for at least
* one scope this client implements. Scopes the client does not serve are ignored: a diversion
* entry for them cannot produce any session traffic, so it must not keep the session machinery
* alive either.
*/
public static boolean divertsAnyTraffic(SessionClientConfiguration sessionConfig) {
for (SessionScope scope : values()) {
if (scope.resolveSessionLoad(sessionConfig) > 0) {
return true;
}
}
return false;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -941,7 +941,11 @@ private UnaryCallable<ConditionalRowMutation, Boolean> createCheckAndMutateRowCa
return bigtableClientContext
.getSessionShim()
.decorateCheckAndMutateRow(classic, perOpSettings.checkAndMutateRowSettings)
.withDefaultCallContext(bigtableClientContext.getClientContext().getDefaultCallContext());
.withDefaultCallContext(
bigtableClientContext
.getClientContext()
.getDefaultCallContext()
.withRetrySettings(perOpSettings.checkAndMutateRowSettings.getRetrySettings()));
}

/**
Expand Down Expand Up @@ -970,7 +974,11 @@ private UnaryCallable<ReadModifyWriteRow, Row> createReadModifyWriteRowCallable(
return bigtableClientContext
.getSessionShim()
.decorateReadModifyWriteRow(classic, rowAdapter, perOpSettings.readModifyWriteRowSettings)
.withDefaultCallContext(bigtableClientContext.getClientContext().getDefaultCallContext());
.withDefaultCallContext(
bigtableClientContext
.getClientContext()
.getDefaultCallContext()
.withRetrySettings(perOpSettings.readModifyWriteRowSettings.getRetrySettings()));
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,23 +12,6 @@ polling_configuration {
}

session_configuration {
session_diversion_configuration {
per_scope_diversion {
scope_diversions {
key: "method:Bigtable.ReadRow"
value {
session_load: 0
}
}
scope_diversions {
key: "method:Bigtable.MutateRow"
value {
session_load: 0
}
}
}
}

channel_configuration {
min_server_count: 2
max_server_count: 25
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@

import com.google.api.gax.core.FixedExecutorProvider;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
import com.google.bigtable.v2.SessionReadModifyWriteRowResponse;
import com.google.bigtable.v2.SessionReadRowResponse;
import com.google.cloud.bigtable.data.v2.internal.api.Client.Resource;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
Expand All @@ -35,6 +37,26 @@ public class ShimImplTest {
private static final InstantiatingGrpcChannelProvider EMPTY_TRANSPORT =
InstantiatingGrpcChannelProvider.newBuilder().build();

/**
* ReadRow and ReadModifyWriteRow disagree on what "no row in the response" means, and the session
* path must match the classic path for each. A ReadRow miss is a null Row; a ReadModifyWriteRow
* always produces one, so a response without a row must still build a Row rather than handing
* user code a null the classic path never produces.
*/
@Test
public void readRowExtractor_noRow_returnsNull() {
assertThat(ShimImpl.READ_ROW_EXTRACTOR.apply(SessionReadRowResponse.getDefaultInstance()))
.isNull();
}

@Test
public void readModifyWriteRowExtractor_noRow_returnsEmptyRowNotNull() {
assertThat(
ShimImpl.READ_MODIFY_WRITE_ROW_EXTRACTOR.apply(
SessionReadModifyWriteRowResponse.getDefaultInstance()))
.isNotNull();
}

@Test
public void selectUserCallbackExecutor_transportSet_returnsSharedTransportExecutor() {
Executor userExec = mock(Executor.class);
Expand Down
Loading
Loading