From 758a5f2e44a882a0a1404889d0f8dfd661508034 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Tue, 5 Jan 2021 18:00:44 +0000 Subject: [PATCH 1/2] fix: Retry "received rst stream" --- .../data/v2/stub/EnhancedBigtableStub.java | 7 +- .../ReadRowsConvertExceptionCallable.java | 85 +++++++++++++++++++ .../v2/stub/readrows/ReadRowsRetryTest.java | 37 ++++++++ 3 files changed, 128 insertions(+), 1 deletion(-) create mode 100644 google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java index 448390396f2d..7a3056a6af83 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java @@ -75,6 +75,7 @@ import com.google.cloud.bigtable.data.v2.stub.mutaterows.MutateRowsRetryingCallable; import com.google.cloud.bigtable.data.v2.stub.readrows.FilterMarkerRowsCallable; import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsBatchingDescriptor; +import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsConvertExceptionCallable; import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsResumptionStrategy; import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsRetryCompletedCallable; import com.google.cloud.bigtable.data.v2.stub.readrows.ReadRowsUserCallable; @@ -365,10 +366,14 @@ public Map extract(ReadRowsRequest readRowsRequest) { new HeaderTracerStreamingCallable<>( watched, settings.getHeaderTracer(), getSpanName("ReadRows").toString()); + // Check for "received rst stream" exceptions and convert them to retryable ApiExceptions + ServerStreamingCallable convertException = + new ReadRowsConvertExceptionCallable<>(withHeaderTracer); + // Retry logic is split into 2 parts to workaround a rare edge case described in // ReadRowsRetryCompletedCallable ServerStreamingCallable retrying1 = - new ReadRowsRetryCompletedCallable<>(withHeaderTracer); + new ReadRowsRetryCompletedCallable<>(convertException); ServerStreamingCallable retrying2 = Callables.retrying(retrying1, innerSettings, clientContext); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java new file mode 100644 index 000000000000..38d9e954b3db --- /dev/null +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java @@ -0,0 +1,85 @@ +/* + * Copyright 2021 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.readrows; + +import com.google.api.core.InternalApi; +import com.google.api.gax.grpc.GrpcStatusCode; +import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.ResponseObserver; +import com.google.api.gax.rpc.ServerStreamingCallable; +import com.google.api.gax.rpc.StreamController; +import io.grpc.Status; +import io.grpc.Status.Code; + +/** + * This callable converts the "Received rst stream" exception into a retryable {@link ApiException}. + */ +@InternalApi +public final class ReadRowsConvertExceptionCallable + extends ServerStreamingCallable { + + private final ServerStreamingCallable innerCallable; + + public ReadRowsConvertExceptionCallable( + ServerStreamingCallable innerCallable) { + this.innerCallable = innerCallable; + } + + @Override + public void call( + ReadRowsRequest request, ResponseObserver responseObserver, ApiCallContext context) { + ReadRowsConvertExceptionResponseObserver observer = + new ReadRowsConvertExceptionResponseObserver<>(responseObserver); + innerCallable.call(request, observer, context); + } + + private class ReadRowsConvertExceptionResponseObserver implements ResponseObserver { + + private final ResponseObserver outerObserver; + + ReadRowsConvertExceptionResponseObserver(ResponseObserver outerObserver) { + this.outerObserver = outerObserver; + } + + @Override + public void onStart(StreamController controller) { + outerObserver.onStart(controller); + } + + @Override + public void onResponse(RowT response) { + outerObserver.onResponse(response); + } + + @Override + public void onError(Throwable t) { + Status status = Status.fromThrowable(t); + if (status.getCode() == Code.INTERNAL + && status.getDescription() != null + && status.getDescription().contains("Received Rst stream")) { + outerObserver.onError(new ApiException(t, GrpcStatusCode.of(status.getCode()), true)); + } else { + outerObserver.onError(t); + } + } + + @Override + public void onComplete() { + outerObserver.onComplete(); + } + } +} diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsRetryTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsRetryTest.java index 96a22f51825d..1536b01e0181 100644 --- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsRetryTest.java +++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsRetryTest.java @@ -16,8 +16,11 @@ package com.google.cloud.bigtable.data.v2.stub.readrows; import com.google.api.gax.core.NoCredentialsProvider; +import com.google.api.gax.grpc.GrpcStatusCode; import com.google.api.gax.grpc.GrpcTransportChannel; +import com.google.api.gax.rpc.ApiException; import com.google.api.gax.rpc.FixedTransportChannelProvider; +import com.google.api.gax.rpc.InternalException; import com.google.api.gax.rpc.ServerStream; import com.google.bigtable.v2.BigtableGrpc; import com.google.bigtable.v2.ReadRowsRequest; @@ -39,6 +42,7 @@ import com.google.protobuf.StringValue; import io.grpc.Status; import io.grpc.Status.Code; +import io.grpc.StatusRuntimeException; import io.grpc.stub.StreamObserver; import io.grpc.testing.GrpcServerRule; import java.io.IOException; @@ -260,6 +264,30 @@ public void retryWithLastScannedKeyTest() { Truth.assertThat(actualResults).containsExactly("r7").inOrder(); } + @Test + public void retryRstStreamExceptionTest() { + ApiException exception = + new InternalException( + new StatusRuntimeException( + Status.INTERNAL.withDescription( + "HTTP/2 error code: INTERNAL_ERROR\nReceived Rst stream")), + GrpcStatusCode.of(Code.INTERNAL), + false); + service.expectations.add( + RpcExpectation.create() + .expectRequest("k1") + .expectRequest(Range.closedOpen("r1", "r3")) + .respondWithException(Code.INTERNAL, exception)); + service.expectations.add( + RpcExpectation.create() + .expectRequest("k1") + .expectRequest(Range.closedOpen("r1", "r3")) + .respondWith("k1", "r1", "r2")); + + List actualResults = getResults(Query.create(TABLE_ID).rowKey("k1").range("r1", "r3")); + Truth.assertThat(actualResults).containsExactly("k1", "r1", "r2").inOrder(); + } + private List getResults(Query query) { ServerStream actualRows = client.readRows(query); List actualValues = Lists.newArrayList(); @@ -292,6 +320,8 @@ public void readRows( } if (expectedRpc.statusCode.toStatus().isOk()) { responseObserver.onCompleted(); + } else if (expectedRpc.exception != null) { + responseObserver.onError(expectedRpc.exception); } else { responseObserver.onError(expectedRpc.statusCode.toStatus().asRuntimeException()); } @@ -301,6 +331,7 @@ public void readRows( private static class RpcExpectation { ReadRowsRequest.Builder requestBuilder; Status.Code statusCode; + ApiException exception; List responses; private RpcExpectation() { @@ -370,6 +401,12 @@ RpcExpectation respondWithStatus(Status.Code code) { return this; } + RpcExpectation respondWithException(Status.Code code, ApiException exception) { + this.statusCode = code; + this.exception = exception; + return this; + } + RpcExpectation respondWith(String... responses) { for (String response : responses) { this.responses.add( From 246f38929a0d4cff3e2bb9cad80a74319146bcd4 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Wed, 20 Jan 2021 18:38:50 +0000 Subject: [PATCH 2/2] Check if exception is InternalException instead of using fromThrowable --- .../data/v2/stub/EnhancedBigtableStub.java | 14 ++++++----- .../ReadRowsConvertExceptionCallable.java | 24 ++++++++++--------- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java index 7a3056a6af83..8f2505c58fcf 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java @@ -346,8 +346,14 @@ public Map extract(ReadRowsRequest readRowsRequest) { .build(), readRowsSettings.getRetryableCodes()); + // Sometimes ReadRows connections are disconnected via an RST frame. This error is transient and + // should be treated similar to UNAVAILABLE. However, this exception has an INTERNAL error code + // which by default is not retryable. Convert the exception so it can be retried in the client. + ServerStreamingCallable convertException = + new ReadRowsConvertExceptionCallable<>(base); + ServerStreamingCallable merging = - new RowMergingCallable<>(base, rowAdapter); + new RowMergingCallable<>(convertException, rowAdapter); // Copy settings for the middle ReadRowsRequest -> RowT callable (as opposed to the inner // ReadRowsRequest -> ReadRowsResponse callable). @@ -366,14 +372,10 @@ public Map extract(ReadRowsRequest readRowsRequest) { new HeaderTracerStreamingCallable<>( watched, settings.getHeaderTracer(), getSpanName("ReadRows").toString()); - // Check for "received rst stream" exceptions and convert them to retryable ApiExceptions - ServerStreamingCallable convertException = - new ReadRowsConvertExceptionCallable<>(withHeaderTracer); - // Retry logic is split into 2 parts to workaround a rare edge case described in // ReadRowsRetryCompletedCallable ServerStreamingCallable retrying1 = - new ReadRowsRetryCompletedCallable<>(convertException); + new ReadRowsRetryCompletedCallable<>(withHeaderTracer); ServerStreamingCallable retrying2 = Callables.retrying(retrying1, innerSettings, clientContext); diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java index 38d9e954b3db..69dd2b5b8abf 100644 --- a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java +++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable.java @@ -16,14 +16,12 @@ package com.google.cloud.bigtable.data.v2.stub.readrows; import com.google.api.core.InternalApi; -import com.google.api.gax.grpc.GrpcStatusCode; import com.google.api.gax.rpc.ApiCallContext; import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.InternalException; import com.google.api.gax.rpc.ResponseObserver; import com.google.api.gax.rpc.ServerStreamingCallable; import com.google.api.gax.rpc.StreamController; -import io.grpc.Status; -import io.grpc.Status.Code; /** * This callable converts the "Received rst stream" exception into a retryable {@link ApiException}. @@ -67,14 +65,7 @@ public void onResponse(RowT response) { @Override public void onError(Throwable t) { - Status status = Status.fromThrowable(t); - if (status.getCode() == Code.INTERNAL - && status.getDescription() != null - && status.getDescription().contains("Received Rst stream")) { - outerObserver.onError(new ApiException(t, GrpcStatusCode.of(status.getCode()), true)); - } else { - outerObserver.onError(t); - } + outerObserver.onError(convertException(t)); } @Override @@ -82,4 +73,15 @@ public void onComplete() { outerObserver.onComplete(); } } + + private Throwable convertException(Throwable t) { + // Long lived connections sometimes are disconnected via an RST frame. This error is + // transient and should be retried. + if (t instanceof InternalException) { + if (t.getMessage() != null && t.getMessage().contains("Received Rst stream")) { + return new InternalException(t, ((InternalException) t).getStatusCode(), true); + } + } + return t; + } }