You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2023/08/17 20:22:14 UTC
[GitHub] [beam] scwhittle commented on a diff in pull request #28050: Don't invalidate streams on quota errors
scwhittle commented on code in PR #28050:
URL: https://github.com/apache/beam/pull/28050#discussion_r1297697681
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -663,7 +663,16 @@ long flush(
retrieveErrorDetails(contexts));
failedContext.failureCount += 1;
- invalidateWriteStream();
+ Throwable error = Preconditions.checkStateNotNull(failedContext.getError());
Review Comment:
Line 607 and retrieveErrorDetails seems to expect this is null sometimes, perhaps safer to just set quotaError to false if getError is null?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -663,7 +663,16 @@ long flush(
retrieveErrorDetails(contexts));
failedContext.failureCount += 1;
- invalidateWriteStream();
+ Throwable error = Preconditions.checkStateNotNull(failedContext.getError());
+ Status.Code statusCode = Status.fromThrowable(error).getCode();
+ boolean quotaError = statusCode.equals(Status.Code.RESOURCE_EXHAUSTED);
+
+ if (!quotaError) {
+ // This forces us to close and reopen all gRPC connections to Storage API on error,
+ // which empirically
+ // fixes random stuckness issues.
+ invalidateWriteStream();
Review Comment:
invalidateWriteStream is currently just unpinning. Should it close() as well?
In the recent issue we appeared to have leaks of ManagedChannels in AppendClient, could that have been because they were not closed there?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java:
##########
@@ -663,7 +663,16 @@ long flush(
retrieveErrorDetails(contexts));
failedContext.failureCount += 1;
- invalidateWriteStream();
+ Throwable error = Preconditions.checkStateNotNull(failedContext.getError());
+ Status.Code statusCode = Status.fromThrowable(error).getCode();
+ boolean quotaError = statusCode.equals(Status.Code.RESOURCE_EXHAUSTED);
+
+ if (!quotaError) {
+ // This forces us to close and reopen all gRPC connections to Storage API on error,
+ // which empirically
+ // fixes random stuckness issues.
Review Comment:
nit: weird wrapping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org