You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2025/06/17 18:35:49 UTC
(beam) 01/01: Dont exit execution loop before bundle is done
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch users/damccorm/prismFailures
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9eb7bf0e98b15afba97cef2b9acbfba8275c5c4b
Author: Danny Mccormick <da...@google.com>
AuthorDate: Tue Jun 17 14:35:30 2025 -0400
Dont exit execution loop before bundle is done
---
sdks/go/pkg/beam/runners/prism/internal/stage.go | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git a/sdks/go/pkg/beam/runners/prism/internal/stage.go b/sdks/go/pkg/beam/runners/prism/internal/stage.go
index a1fd49f609d..a2713196104 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/stage.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/stage.go
@@ -208,8 +208,8 @@ progress:
ticked = true
resp, err := b.Progress(ctx, wk)
if err != nil {
- slog.Debug("SDK Error from progress, aborting progress", "bundle", rb, "error", err.Error())
- break progress
+ slog.Debug("SDK Error from progress request, aborting progress update", "bundle", rb, "error", err.Error())
+ continue progress
}
index, unknownIDs := j.ContributeTentativeMetrics(resp)
if len(unknownIDs) > 0 {
@@ -224,8 +224,11 @@ progress:
slog.Debug("splitting report", "bundle", rb, "index", index)
sr, err := b.Split(ctx, wk, 0.5 /* fraction of remainder */, nil /* allowed splits */)
if err != nil {
- slog.Warn("SDK Error from split, aborting splits", "bundle", rb, "error", err.Error())
- break progress
+ slog.Warn("SDK Error from split, aborting splits and failing bundle", "bundle", rb, "error", err.Error())
+ if b.BundleErr != nil {
+ b.BundleErr = err
+ }
+ return b.BundleErr
}
if sr.GetChannelSplits() == nil {
slog.Debug("SDK returned no splits", "bundle", rb)