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 21:29:39 UTC
(beam) 01/01: Dont exit execution loop before bundle is done (#35337)
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch prism-cp
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 353fabb2b4787716bd2c79f316a9f74235d4dd37
Author: Danny McCormick <da...@google.com>
AuthorDate: Tue Jun 17 17:24:24 2025 -0400
Dont exit execution loop before bundle is done (#35337)
* Dont exit execution loop before bundle is done
* Turn off progress updates when there is a failure
---
sdks/go/pkg/beam/runners/prism/internal/stage.go | 12 ++++++++----
1 file changed, 8 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..1b30cf31bff 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/stage.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/stage.go
@@ -208,8 +208,9 @@ 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 and turning off future progress updates", "bundle", rb, "error", err.Error())
+ progTick.Stop()
+ continue progress
}
index, unknownIDs := j.ContributeTentativeMetrics(resp)
if len(unknownIDs) > 0 {
@@ -224,8 +225,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)