Sitelet https://github.com/googleapis/java-bigtable/pull/2348/files
Skip to content
This repository was archived by the owner on Aug 30, 2026. It is now read-only.

fix: add RetryCallable to the callable chain #2348

Merged
mutianf merged 3 commits into from
Sep 24, 2024
Merged

fix: add RetryCallable to the callable chain #2348

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
971aa7a
fix: add StreamingAttemptCallable to the callable chain
mutianf Sep 20, 2024
3f36f48
add a test
mutianf Sep 20, 2024
413c0e2
add a comment
mutianf Sep 20, 2024
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
18 changes: 13 additions & 5 deletions ...d-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1335,17 +1335,16 @@ public Map<String, String> extract(ExecuteQueryRequest executeQueryRequest) {
ServerStreamingCallable<ExecuteQueryRequest, ExecuteQueryResponse> withStatsHeaders =
new StatsHeadersServerStreamingCallable<>(base);

ServerStreamingCallSettings<ExecuteQueryRequest, ExecuteQueryResponse> innerSettings =
ServerStreamingCallSettings<ExecuteQueryRequest, ExecuteQueryResponse> watchdogSettings =
ServerStreamingCallSettings.<ExecuteQueryRequest, ExecuteQueryResponse>newBuilder()
// TODO resumption strategy and retry settings
.setIdleTimeout(settings.executeQuerySettings().getIdleTimeout())
.setWaitTimeout(settings.executeQuerySettings().getWaitTimeout())
.build();

// Watchdog needs to stay above the metadata observer so that watchdog errors
// are passed through to the metadata future.
ServerStreamingCallable<ExecuteQueryRequest, ExecuteQueryResponse> watched =
Callables.watched(withStatsHeaders, innerSettings, clientContext);
Callables.watched(withStatsHeaders, watchdogSettings, clientContext);

ServerStreamingCallable<ExecuteQueryCallContext, ExecuteQueryResponse> withMetadataObserver =
new MetadataResolvingCallable(watched);
Expand All @@ -1356,10 +1355,19 @@ public Map<String, String> extract(ExecuteQueryRequest executeQueryRequest) {
ServerStreamingCallable<ExecuteQueryCallContext, SqlRow> withBigtableTracer =
new BigtableTracerStreamingCallable<>(merging);

ServerStreamingCallSettings<ExecuteQueryCallContext, SqlRow> retrySettings =
ServerStreamingCallSettings.<ExecuteQueryCallContext, SqlRow>newBuilder()
// TODO resumption strategy and retry settings
.build();

// Adding RetryingCallable to the callable chain so that client side metrics can be
// measured correctly. Retries are currently disabled.
ServerStreamingCallable<ExecuteQueryCallContext, SqlRow> retries =

@jackdingilian jackdingilian Sep 20, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Will you put a comment noting that this is just for client side metrics and we don't want any retries to happen right now?

withRetries(withBigtableTracer, retrySettings);

SpanName span = getSpanName("ExecuteQuery");
ServerStreamingCallable<ExecuteQueryCallContext, SqlRow> traced =
new TracedServerStreamingCallable<>(
withBigtableTracer, clientContext.getTracerFactory(), span);
new TracedServerStreamingCallable<>(retries, clientContext.getTracerFactory(), span);

return new ExecuteQueryCallable(
traced.withDefaultCallContext(clientContext.getDefaultCallContext()), requestContext);
Expand Down
63 changes: 63 additions & 0 deletions ...le/src/test/java/com/google/cloud/bigtable/data/v2/stub/sql/ExecuteQueryCallableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,30 @@
import static com.google.cloud.bigtable.data.v2.stub.sql.SqlProtoFactory.stringType;
import static com.google.cloud.bigtable.data.v2.stub.sql.SqlProtoFactory.stringValue;
import static com.google.common.truth.Truth.assertThat;
import static org.junit.Assert.assertThrows;

import com.google.api.gax.rpc.UnavailableException;
import com.google.bigtable.v2.BigtableGrpc;
import com.google.bigtable.v2.ExecuteQueryRequest;
import com.google.bigtable.v2.ExecuteQueryResponse;
import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
import com.google.cloud.bigtable.data.v2.FakeServiceBuilder;
import com.google.cloud.bigtable.data.v2.internal.ProtoResultSetMetadata;
import com.google.cloud.bigtable.data.v2.internal.ProtoSqlRow;
import com.google.cloud.bigtable.data.v2.internal.RequestContext;
import com.google.cloud.bigtable.data.v2.internal.SqlRow;
import com.google.cloud.bigtable.data.v2.models.sql.Statement;
import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStub;
import com.google.cloud.bigtable.gaxx.testing.FakeStreamingApi.ServerStreamingStashCallable;
import io.grpc.Server;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.util.Collections;
import java.util.Iterator;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
Expand All @@ -39,6 +54,29 @@ public class ExecuteQueryCallableTest {
private static final RequestContext REQUEST_CONTEXT =
RequestContext.create("fake-project", "fake-instance", "fake-profile");

private Server server;
private FakeService fakeService = new FakeService();
private EnhancedBigtableStub stub;

@Before
public void setup() throws IOException {
server = FakeServiceBuilder.create(fakeService).start();

BigtableDataSettings settings =
BigtableDataSettings.newBuilderForEmulator(server.getPort())
.setProjectId("fake-project")
.setInstanceId("fake-instance")
.build();

stub = EnhancedBigtableStub.create(settings.getStubSettings());
}

@After
public void tearDown() {
stub.close();
server.shutdown();
}

@Test
public void testCallContextAndServerStreamSetup() {
SqlRow row =
Expand All @@ -57,4 +95,29 @@ public void testCallContextAndServerStreamSetup() {
assertThat(responseIterator.next()).isEqualTo(row);
assertThat(responseIterator.hasNext()).isFalse();
}

@Test
public void testExecuteQueryRequestsAreNotRetried() {
// TODO: retries for execute query is currently disabled. This test should be
// updated once resumption token is in place.
SqlServerStream stream = stub.executeQueryCallable().call(Statement.of("SELECT * FROM table"));

Iterator<SqlRow> iterator = stream.rows().iterator();

assertThrows(UnavailableException.class, iterator::next).getCause();
assertThat(fakeService.attempts).isEqualTo(1);
}

private static class FakeService extends BigtableGrpc.BigtableImplBase {

private int attempts = 0;

@Override
public void executeQuery(
ExecuteQueryRequest request, StreamObserver<ExecuteQueryResponse> responseObserver) {
attempts++;
responseObserver.onNext(metadata(columnMetadata("test", stringType())));
responseObserver.onError(new StatusRuntimeException(Status.UNAVAILABLE));
}
}
}