-
Notifications
You must be signed in to change notification settings - Fork 122
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Implemented BigQueryRetryAlgorithm to retry on the basis of the…
… configured re-triable error messages (#1426) * Updated BigQueryImpl * Initial Commit * Using BigQueryRetryAlgorithm as the retry algorithm * Created BigQueryRetryAlgorithm as a subclass of RetryAlgorithm * BigQueryErrorMessages property file * Implemented Builder Logic for BigQueryRetryConfig * Using BigQueryRetryConfig for getQueryResults * Updated shouldRetry with the logic to retry based on error messages * Implemented null checks on shouldRetryBasedOnBigQueryRetryConfig * Removed `Status` from shouldRetryBasedOnBigQueryRetryConfig implementation * Removed unused imports * created DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG for getQueryResults * Added testGetQueryResultsRetry test for testing getQueryResults Retry * Overriding createNextAttempt method so that it generates an attempt based on the error message * Linted BigQueryRetryHelper * Linted testGetQueryResultsRetry * Linted BigQueryRetryAlgorithm * Linted BigQueryErrorMessages * Linted BigQueryRetryConfig * Fixed Linting * Fixed Linting * Fixed Linting * Created translateAndThrow(BigQueryRetryHelper.BigQueryRetryHelperException ex) method to handle BigQueryRetryHelperException * Handling BigQueryRetryHelper.BigQueryRetryHelperException for getQueryResults * Implementing BigQueryRetryHelper.runWithRetries from TableResult.queryRPC method * Implementing testFastQueryRateLimitIdempotency Method to test Idempotency of the BigQueryRetryHelper.runWithRetries for TableResult.query(...) * Changed DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG to DEFAULT_RETRY_CONFIG * Implemented `BigQueryRetryHelper.runWithRetries` on `QueryResponse waitForQueryResults` method, which is used by `TableResult getQueryResults` method * Revert "Implemented `BigQueryRetryHelper.runWithRetries` on `QueryResponse waitForQueryResults` method, which is used by `TableResult getQueryResults` method" This reverts commit 84a3418. * Revert "Changed DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG to DEFAULT_RETRY_CONFIG" This reverts commit 22b1706. * Renamed DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG to DEFAULT_RETRY_CONFIG * Revert "Renamed DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG to DEFAULT_RETRY_CONFIG" This reverts commit 2d21e11. * Renamed DEFAULT_RATE_LIMIT_EXCEEDED_RETRY_CONFIG to DEFAULT_RETRY_CONFIG * Implemented BigQueryRetryHelper.runWithRetries on `QueryResponse waitForQueryResults` method, which is used by `TableResult getQueryResults` method
- Loading branch information
Showing
8 changed files
with
433 additions
and
13 deletions.
There are no files selected for viewing
22 changes: 22 additions & 0 deletions
22
google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryErrorMessages.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * 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 com.google.cloud.bigquery; | ||
|
|
||
| public class BigQueryErrorMessages { | ||
| public static final String RATE_LIMIT_EXCEEDED_MSG = | ||
| "Exceeded rate limits:"; // Error Message for RateLimitExceeded Error | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
152 changes: 152 additions & 0 deletions
152
google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryAlgorithm.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,152 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * 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 com.google.cloud.bigquery; | ||
|
|
||
| import static com.google.common.base.Preconditions.checkNotNull; | ||
|
|
||
| import com.google.api.gax.retrying.ResultRetryAlgorithm; | ||
| import com.google.api.gax.retrying.ResultRetryAlgorithmWithContext; | ||
| import com.google.api.gax.retrying.RetryAlgorithm; | ||
| import com.google.api.gax.retrying.RetryingContext; | ||
| import com.google.api.gax.retrying.TimedAttemptSettings; | ||
| import com.google.api.gax.retrying.TimedRetryAlgorithm; | ||
| import com.google.api.gax.retrying.TimedRetryAlgorithmWithContext; | ||
| import java.util.Iterator; | ||
| import java.util.concurrent.CancellationException; | ||
|
|
||
| public class BigQueryRetryAlgorithm<ResponseT> extends RetryAlgorithm<ResponseT> { | ||
| private final BigQueryRetryConfig bigQueryRetryConfig; | ||
| private final ResultRetryAlgorithm<ResponseT> resultAlgorithm; | ||
| private final TimedRetryAlgorithm timedAlgorithm; | ||
| private final ResultRetryAlgorithmWithContext<ResponseT> resultAlgorithmWithContext; | ||
| private final TimedRetryAlgorithmWithContext timedAlgorithmWithContext; | ||
|
|
||
| public BigQueryRetryAlgorithm( | ||
| ResultRetryAlgorithm<ResponseT> resultAlgorithm, | ||
| TimedRetryAlgorithm timedAlgorithm, | ||
| BigQueryRetryConfig bigQueryRetryConfig) { | ||
| super(resultAlgorithm, timedAlgorithm); | ||
| this.bigQueryRetryConfig = checkNotNull(bigQueryRetryConfig); | ||
| this.resultAlgorithm = checkNotNull(resultAlgorithm); | ||
| this.timedAlgorithm = checkNotNull(timedAlgorithm); | ||
| this.resultAlgorithmWithContext = null; | ||
| this.timedAlgorithmWithContext = null; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean shouldRetry( | ||
| RetryingContext context, | ||
| Throwable previousThrowable, | ||
| ResponseT previousResponse, | ||
| TimedAttemptSettings nextAttemptSettings) | ||
| throws CancellationException { | ||
| // Implementing shouldRetryBasedOnBigQueryRetryConfig so that we can retry exceptions based on | ||
| // the exception messages | ||
| return (shouldRetryBasedOnResult(context, previousThrowable, previousResponse) | ||
| || shouldRetryBasedOnBigQueryRetryConfig(previousThrowable, bigQueryRetryConfig)) | ||
| && shouldRetryBasedOnTiming(context, nextAttemptSettings); | ||
| } | ||
|
|
||
| private boolean shouldRetryBasedOnBigQueryRetryConfig( | ||
| Throwable previousThrowable, BigQueryRetryConfig bigQueryRetryConfig) { | ||
| /* | ||
| We are deciding if a given error should be retried on the basis of error message. | ||
| Cannot rely on Error/Status code as for example error code 400 (which is not retriable) could be thrown due to rateLimitExceed, which is retriable | ||
| */ | ||
| String errorDesc; | ||
| if (previousThrowable != null && (errorDesc = previousThrowable.getMessage()) != null) { | ||
| for (Iterator<String> retriableMessages = | ||
| bigQueryRetryConfig.getRetriableErrorMessages().iterator(); | ||
| retriableMessages.hasNext(); ) { | ||
| if (errorDesc.contains(retriableMessages.next())) { // Error message should be retried | ||
| return true; | ||
| } | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| /*Duplicating this method as it can not be inherited from the RetryAlgorithm due to the default access modifier*/ | ||
| boolean shouldRetryBasedOnResult( | ||
| RetryingContext context, Throwable previousThrowable, ResponseT previousResponse) { | ||
| if (resultAlgorithmWithContext != null && context != null) { | ||
| return resultAlgorithmWithContext.shouldRetry(context, previousThrowable, previousResponse); | ||
| } | ||
| return getResultAlgorithm().shouldRetry(previousThrowable, previousResponse); | ||
| } | ||
|
|
||
| /*Duplicating this method as it can not be inherited from the RetryAlgorithm due to the private access modifier*/ | ||
| private boolean shouldRetryBasedOnTiming( | ||
| RetryingContext context, TimedAttemptSettings nextAttemptSettings) { | ||
| if (nextAttemptSettings == null) { | ||
| return false; | ||
| } | ||
| if (timedAlgorithmWithContext != null && context != null) { | ||
| return timedAlgorithmWithContext.shouldRetry(context, nextAttemptSettings); | ||
| } | ||
| return getTimedAlgorithm().shouldRetry(nextAttemptSettings); | ||
| } | ||
|
|
||
| @Override | ||
| public TimedAttemptSettings createNextAttempt( | ||
| RetryingContext context, | ||
| Throwable previousThrowable, | ||
| ResponseT previousResponse, | ||
| TimedAttemptSettings previousSettings) { | ||
| // a small optimization that avoids calling relatively heavy methods | ||
| // like timedAlgorithm.createNextAttempt(), when it is not necessary. | ||
|
|
||
| if (!((shouldRetryBasedOnResult(context, previousThrowable, previousResponse) | ||
| || shouldRetryBasedOnBigQueryRetryConfig( | ||
| previousThrowable, | ||
| bigQueryRetryConfig)))) { // Calling shouldRetryBasedOnBigQueryRetryConfig to check if | ||
| // the error message could be retried | ||
| return null; | ||
| } | ||
|
|
||
| TimedAttemptSettings newSettings = | ||
| createNextAttemptBasedOnResult( | ||
| context, previousThrowable, previousResponse, previousSettings); | ||
| if (newSettings == null) { | ||
| newSettings = createNextAttemptBasedOnTiming(context, previousSettings); | ||
| } | ||
| return newSettings; | ||
| } | ||
|
|
||
| /*Duplicating this method as it can not be inherited from the RetryAlgorithm due to the private access modifier*/ | ||
| private TimedAttemptSettings createNextAttemptBasedOnResult( | ||
| RetryingContext context, | ||
| Throwable previousThrowable, | ||
| ResponseT previousResponse, | ||
| TimedAttemptSettings previousSettings) { | ||
| if (resultAlgorithmWithContext != null && context != null) { | ||
| return resultAlgorithmWithContext.createNextAttempt( | ||
| context, previousThrowable, previousResponse, previousSettings); | ||
| } | ||
| return getResultAlgorithm() | ||
| .createNextAttempt(previousThrowable, previousResponse, previousSettings); | ||
| } | ||
|
|
||
| /*Duplicating this method as it can not be inherited from the RetryAlgorithm due to the private access modifier*/ | ||
| private TimedAttemptSettings createNextAttemptBasedOnTiming( | ||
| RetryingContext context, TimedAttemptSettings previousSettings) { | ||
| if (timedAlgorithmWithContext != null && context != null) { | ||
| return timedAlgorithmWithContext.createNextAttempt(context, previousSettings); | ||
| } | ||
| return getTimedAlgorithm().createNextAttempt(previousSettings); | ||
| } | ||
| } |
54 changes: 54 additions & 0 deletions
54
google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,54 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * 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 com.google.cloud.bigquery; | ||
|
|
||
| import static com.google.common.base.Preconditions.checkNotNull; | ||
|
|
||
| import com.google.common.collect.ImmutableSet; | ||
|
|
||
| public class BigQueryRetryConfig { | ||
| private final ImmutableSet<String> retriableErrorMessages; | ||
|
|
||
| private BigQueryRetryConfig(Builder builder) { | ||
| retriableErrorMessages = builder.retriableErrorMessages.build(); | ||
| } | ||
|
|
||
| public ImmutableSet<String> getRetriableErrorMessages() { | ||
| return retriableErrorMessages; | ||
| } | ||
|
|
||
| // BigQueryRetryConfig builder | ||
| public static class Builder { | ||
| private final ImmutableSet.Builder<String> retriableErrorMessages = ImmutableSet.builder(); | ||
|
|
||
| private Builder() {} | ||
|
|
||
| public final Builder retryOnMessage(String... errorMessages) { | ||
| for (String errorMessage : errorMessages) { | ||
| retriableErrorMessages.add(checkNotNull(errorMessage)); | ||
| } | ||
| return this; | ||
| } | ||
|
|
||
| public BigQueryRetryConfig build() { | ||
| return new BigQueryRetryConfig(this); | ||
| } | ||
| } | ||
|
|
||
| public static Builder newBuilder() { | ||
| return new Builder(); | ||
| } | ||
| } |
76 changes: 76 additions & 0 deletions
76
google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryRetryHelper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /* | ||
| * 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 | ||
| * | ||
| * 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 com.google.cloud.bigquery; | ||
|
|
||
| import com.google.api.core.ApiClock; | ||
| import com.google.api.gax.retrying.*; | ||
| import com.google.cloud.RetryHelper; | ||
| import java.util.concurrent.Callable; | ||
| import java.util.concurrent.ExecutionException; | ||
|
|
||
| public class BigQueryRetryHelper extends RetryHelper { | ||
|
|
||
| public static <V> V runWithRetries( | ||
| Callable<V> callable, | ||
| RetrySettings retrySettings, | ||
| ResultRetryAlgorithm<?> resultRetryAlgorithm, | ||
| ApiClock clock, | ||
| BigQueryRetryConfig bigQueryRetryConfig) | ||
| throws RetryHelperException { | ||
| try { | ||
| // Suppressing should be ok as a workaraund. Current and only ResultRetryAlgorithm | ||
| // implementation does not use response at all, so ignoring its type is ok. | ||
| @SuppressWarnings("unchecked") | ||
| ResultRetryAlgorithm<V> algorithm = (ResultRetryAlgorithm<V>) resultRetryAlgorithm; | ||
| return run( | ||
| callable, | ||
| new ExponentialRetryAlgorithm(retrySettings, clock), | ||
| algorithm, | ||
| bigQueryRetryConfig); | ||
| } catch (Exception e) { | ||
| throw new BigQueryRetryHelperException(e.getCause()); | ||
| } | ||
| } | ||
|
|
||
| private static <V> V run( | ||
| Callable<V> callable, | ||
| TimedRetryAlgorithm timedAlgorithm, | ||
| ResultRetryAlgorithm<V> resultAlgorithm, | ||
| BigQueryRetryConfig bigQueryRetryConfig) | ||
| throws ExecutionException, InterruptedException { | ||
| RetryAlgorithm<V> retryAlgorithm = | ||
| new BigQueryRetryAlgorithm<>( | ||
| resultAlgorithm, | ||
| timedAlgorithm, | ||
| bigQueryRetryConfig); // using BigQueryRetryAlgorithm in place of | ||
| // com.google.api.gax.retrying.RetryAlgorithm, as | ||
| // BigQueryRetryAlgorithm retries considering bigQueryRetryConfig | ||
| RetryingExecutor<V> executor = new DirectRetryingExecutor<>(retryAlgorithm); | ||
|
|
||
| RetryingFuture<V> retryingFuture = executor.createFuture(callable); | ||
| executor.submit(retryingFuture); | ||
| return retryingFuture.get(); | ||
| } | ||
|
|
||
| public static class BigQueryRetryHelperException extends RuntimeException { | ||
|
|
||
| private static final long serialVersionUID = -8519852520090965314L; | ||
|
|
||
| BigQueryRetryHelperException(Throwable cause) { | ||
| super(cause); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.