You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lo...@apache.org on 2023/03/31 23:46:02 UTC
[beam] branch stagecheckErr created (now a2df67be258)
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a change to branch stagecheckErr
in repository https://gitbox.apache.org/repos/asf/beam.git
at a2df67be258 Return server error on stream close.
This branch includes the following new commits:
new a2df67be258 Return server error on stream close.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[beam] 01/01: Return server error on stream close.
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lostluck pushed a commit to branch stagecheckErr
in repository https://gitbox.apache.org/repos/asf/beam.git
commit a2df67be258b6b2e6b1d50c338814204dd4121b1
Author: lostluck <13...@users.noreply.github.com>
AuthorDate: Fri Mar 31 16:45:24 2023 -0700
Return server error on stream close.
---
sdks/go/pkg/beam/artifact/stage.go | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
diff --git a/sdks/go/pkg/beam/artifact/stage.go b/sdks/go/pkg/beam/artifact/stage.go
index 0a23b49a479..de017b4c3c0 100644
--- a/sdks/go/pkg/beam/artifact/stage.go
+++ b/sdks/go/pkg/beam/artifact/stage.go
@@ -159,11 +159,11 @@ func Stage(ctx context.Context, client jobpb.LegacyArtifactStagingServiceClient,
}
stagedHash, err := stageChunks(stream, fd)
if err != nil {
- stream.CloseAndRecv() // ignore error
- return nil, errors.Wrapf(err, "failed to send chunks for %v", filename)
+ _, errClose := stream.CloseAndRecv()
+ return nil, errors.Wrapf(err, "failed to send chunks for %v; close error: %v", filename, errClose)
}
- if _, err := stream.CloseAndRecv(); err != nil && err != io.EOF {
- return nil, errors.Wrapf(err, "failed to close stream for %v", filename)
+ if resp, err := stream.CloseAndRecv(); err != nil && err != io.EOF {
+ return nil, errors.Wrapf(err, "failed to close stream for %v; response: %v", filename, resp)
}
if hash != stagedHash {
return nil, errors.Errorf("unexpected SHA256 for sent chunks for %v: %v, want %v", filename, stagedHash, hash)
@@ -188,7 +188,11 @@ func stageChunks(stream jobpb.LegacyArtifactStagingService_PutArtifactClient, r
},
},
}
- if err := stream.Send(chunk); err != nil {
+ err := stream.Send(chunk)
+ if err == io.EOF {
+ return "", err
+ }
+ if err != nil {
return "", errors.Wrap(err, "chunk send failed")
}
}