You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "shunping (via GitHub)" <gi...@apache.org> on 2026/05/17 02:37:16 UTC
[PR] Debug grpc deadline exceed exceptions [beam]
shunping opened a new pull request, #38523:
URL: https://github.com/apache/beam/pull/38523
(no comment)
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4512905857
The failed YAML Xlang test is irrelevant to the change.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4501515655
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request addresses stability issues in the Prism runner where worker failures during bundle processing or dynamic splits caused the system to hang and eventually time out with gRPC DEADLINE_EXCEEDED errors. By improving error propagation and implementing a signaling mechanism to abort stalled data streams, the runner can now fail fast with descriptive errors, significantly improving the robustness of pipeline execution.
### Highlights
* **Dynamic Split Error Handling**: Fixed an issue where dynamic split errors were not correctly captured, ensuring that the runner can now properly identify and report these failures.
* **Bundle Processing Deadlock Prevention**: Introduced a bundle-level 'Done' signal to allow immediate abortion of data streaming when a worker fails, preventing the runner from hanging during bundle processing.
<details>
<summary><b>New Features</b></summary>
🧠You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping merged PR #38523:
URL: https://github.com/apache/beam/pull/38523
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3278079942
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -102,7 +104,11 @@ func (b *B) Respond(resp *fnpb.InstructionResponse) {
return
}
b.responded = true
+ if b.Done != nil {
+ close(b.Done)
+ }
if resp.GetError() != "" {
+ slog.Error("DEBUG: Prism received bundle error from worker response", "bundle", resp.GetInstructionId())
b.BundleErr = fmt.Errorf("bundle %v %v failed:%v", resp.GetInstructionId(), b.PBDID, resp.GetError())
Review Comment:
Fixed.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3278085109
##########
sdks/go/pkg/beam/runners/prism/internal/testdofns_test.go:
##########
@@ -404,3 +405,37 @@ func (fn *selfCheckpointingDoFn) ProcessElement(rt *sdf.LockRTracker, _ []byte,
}
}
}
+
+type errorSplitTracker struct {
+ *offsetrange.Tracker
+}
+
+func (t *errorSplitTracker) TrySplit(fraction float64) (any, any, error) {
+ return nil, nil, fmt.Errorf("intentional split error from tracker")
+}
+
+type slowFailSDF struct{}
+
+func (fn *slowFailSDF) CreateInitialRestriction(config SourceConfig) offsetrange.Restriction {
+ return offsetrange.Restriction{Start: 0, End: config.NumElements}
+}
+
+func (fn *slowFailSDF) SplitRestriction(config SourceConfig, rest offsetrange.Restriction) []offsetrange.Restriction {
+ return rest.EvenSplits(config.InitialSplits)
+}
+
+func (fn *slowFailSDF) RestrictionSize(_ SourceConfig, rest offsetrange.Restriction) float64 {
+ return rest.Size()
+}
+
+func (fn *slowFailSDF) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker {
+ return sdf.NewLockRTracker(&errorSplitTracker{Tracker: offsetrange.NewTracker(rest)})
+}
+
+func (fn *slowFailSDF) ProcessElement(ctx context.Context, rt *sdf.LockRTracker, config SourceConfig, emit func(int64)) error {
+ fmt.Println("DEBUG: slowFailSDF.ProcessElement invoked")
Review Comment:
Fixed.
--
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
Re: [PR] Debug grpc deadline exceed exceptions [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4490087053
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `14.28571%` with `6 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.72%. Comparing base ([`1ca1faf`](https://app.codecov.io/gh/apache/beam/commit/1ca1fafcd62b0e3f1bd6a071842cab7566366d1c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d606a85`](https://app.codecov.io/gh/apache/beam/commit/d606a851e4329d45bfe9a3e359888a510ef41b4a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 7 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...on/apache\_beam/runners/portability/prism\_runner.py](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&filepath=sdks%2Fpython%2Fapache_beam%2Frunners%2Fportability%2Fprism_runner.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9wcmlzbV9ydW5uZXIucHk=) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #38523 +/- ##
============================================
- Coverage 58.73% 58.72% -0.01%
Complexity 7418 7418
============================================
Files 1649 1649
Lines 212369 212382 +13
Branches 6330 6330
============================================
+ Hits 124727 124730 +3
- Misses 83095 83105 +10
Partials 4547 4547
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/38523/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/38523/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.88% <14.28%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3278081039
##########
sdks/go/pkg/beam/runners/prism/internal/stage.go:
##########
@@ -240,7 +240,7 @@ progress:
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 and failing bundle", "bundle", rb, "error", err.Error())
- if b.BundleErr != nil {
+ if b.BundleErr == nil {
b.BundleErr = err
}
Review Comment:
Added a mutex and thread-safe setter and getter.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4501486595
r: @lostluck
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3282519889
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -61,8 +62,18 @@ type B struct {
dataSema atomic.Int32
OutputData engine.TentativeData
- Resp chan *fnpb.ProcessBundleResponse
- Done chan struct{}
+ Resp chan *fnpb.ProcessBundleResponse
+ // DataAbort is closed when the worker responds to the bundle instruction
+ // (with success or failure), signaling ProcessOn to stop streaming data.
+ //
+ // This prevents a deadlock where a worker fails mid-bundle and stops reading
+ // from the data channel while the runner blocks indefinitely attempting to
+ // write remaining elements. Other signals are insufficient to abort immediately:
+ // - ctx.Done() only triggers on global timeouts/cancellations, which is too late.
+ // - wk.StoppedChan is only closed when tearing down the worker pool, which does
+ // not happen while the runner is waiting on the current bundle to finish.
+ DataAbort chan struct{}
+ mu sync.Mutex
BundleErr error
Review Comment:
We should make this bundleErr to soft-indicate that we should be using SetErr and GetErr. The rename force catches any missing locations.
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -98,18 +109,38 @@ func (b *B) LogValue() slog.Value {
slog.String("stage", b.PBDID))
}
+// SetErr sets the bundle error if it is not already set.
+func (b *B) SetErr(err error) {
Review Comment:
Consider having this return a boolean to indicate whether the value was set or not. That way the caller can optionally log the error. (eg. "Error for bundle %v already set, logging dropped followup error: %v").
I can't imagine we would need to log all the locations where this is happening.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3277558302
##########
sdks/go/pkg/beam/runners/prism/internal/stage.go:
##########
@@ -240,7 +240,7 @@ progress:
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 and failing bundle", "bundle", rb, "error", err.Error())
- if b.BundleErr != nil {
+ if b.BundleErr == nil {
b.BundleErr = err
}
Review Comment:
This is true. If the bundle split error happens at the same time as the bundle finishing (and providing the bundle response), there will be a data race on the interface pointer.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4501498483
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "lostluck (via GitHub)" <gi...@apache.org>.
lostluck commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3277561954
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -102,7 +104,11 @@ func (b *B) Respond(resp *fnpb.InstructionResponse) {
return
}
b.responded = true
+ if b.Done != nil {
+ close(b.Done)
+ }
if resp.GetError() != "" {
+ slog.Error("DEBUG: Prism received bundle error from worker response", "bundle", resp.GetInstructionId())
b.BundleErr = fmt.Errorf("bundle %v %v failed:%v", resp.GetInstructionId(), b.PBDID, resp.GetError())
Review Comment:
Also a real error. Don't trigger something that might cause a read before the write it's expected to read. Violates the "happens before" type relationship.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3276336584
##########
sdks/go/pkg/beam/runners/prism/internal/stage.go:
##########
@@ -240,7 +240,7 @@ progress:
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 and failing bundle", "bundle", rb, "error", err.Error())
- if b.BundleErr != nil {
+ if b.BundleErr == nil {
b.BundleErr = err
}
Review Comment:
Accessing and setting `b.BundleErr` here is a data race, as this field is also modified in the worker's response handler (`Respond`) in a different goroutine. Since the `B` struct does not appear to have internal synchronization for this field, you should use a mutex or other synchronization primitive to ensure thread-safe access.
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -102,7 +104,11 @@ func (b *B) Respond(resp *fnpb.InstructionResponse) {
return
}
b.responded = true
+ if b.Done != nil {
+ close(b.Done)
+ }
if resp.GetError() != "" {
+ slog.Error("DEBUG: Prism received bundle error from worker response", "bundle", resp.GetInstructionId())
b.BundleErr = fmt.Errorf("bundle %v %v failed:%v", resp.GetInstructionId(), b.PBDID, resp.GetError())
Review Comment:
There is a race condition here: `b.Done` is closed before `b.BundleErr` is set. A consumer waiting on `b.Done` (e.g., in `ProcessOn`) might see the channel closure and proceed to check `b.BundleErr` before it has been assigned the actual error from the worker. This could lead to a failed bundle being incorrectly treated as successful. Additionally, the log message contains a "DEBUG:" prefix while being at the `Error` level, which is confusing and should be cleaned up.
```suggestion
if resp.GetError() != "" {
slog.Error("Prism received bundle error from worker response", "bundle", resp.GetInstructionId())
b.BundleErr = fmt.Errorf("bundle %v %v failed:%v", resp.GetInstructionId(), b.PBDID, resp.GetError())
}
if b.Done != nil {
close(b.Done)
}
if resp.GetError() != "" {
```
##########
sdks/go/pkg/beam/runners/prism/internal/testdofns_test.go:
##########
@@ -404,3 +405,37 @@ func (fn *selfCheckpointingDoFn) ProcessElement(rt *sdf.LockRTracker, _ []byte,
}
}
}
+
+type errorSplitTracker struct {
+ *offsetrange.Tracker
+}
+
+func (t *errorSplitTracker) TrySplit(fraction float64) (any, any, error) {
+ return nil, nil, fmt.Errorf("intentional split error from tracker")
+}
+
+type slowFailSDF struct{}
+
+func (fn *slowFailSDF) CreateInitialRestriction(config SourceConfig) offsetrange.Restriction {
+ return offsetrange.Restriction{Start: 0, End: config.NumElements}
+}
+
+func (fn *slowFailSDF) SplitRestriction(config SourceConfig, rest offsetrange.Restriction) []offsetrange.Restriction {
+ return rest.EvenSplits(config.InitialSplits)
+}
+
+func (fn *slowFailSDF) RestrictionSize(_ SourceConfig, rest offsetrange.Restriction) float64 {
+ return rest.Size()
+}
+
+func (fn *slowFailSDF) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker {
+ return sdf.NewLockRTracker(&errorSplitTracker{Tracker: offsetrange.NewTracker(rest)})
+}
+
+func (fn *slowFailSDF) ProcessElement(ctx context.Context, rt *sdf.LockRTracker, config SourceConfig, emit func(int64)) error {
+ fmt.Println("DEBUG: slowFailSDF.ProcessElement invoked")
Review Comment:
This `fmt.Println` appears to be a leftover debug statement and should be removed to keep the test output clean.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #38523:
URL: https://github.com/apache/beam/pull/38523#discussion_r3283266226
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -61,8 +62,18 @@ type B struct {
dataSema atomic.Int32
OutputData engine.TentativeData
- Resp chan *fnpb.ProcessBundleResponse
- Done chan struct{}
+ Resp chan *fnpb.ProcessBundleResponse
+ // DataAbort is closed when the worker responds to the bundle instruction
+ // (with success or failure), signaling ProcessOn to stop streaming data.
+ //
+ // This prevents a deadlock where a worker fails mid-bundle and stops reading
+ // from the data channel while the runner blocks indefinitely attempting to
+ // write remaining elements. Other signals are insufficient to abort immediately:
+ // - ctx.Done() only triggers on global timeouts/cancellations, which is too late.
+ // - wk.StoppedChan is only closed when tearing down the worker pool, which does
+ // not happen while the runner is waiting on the current bundle to finish.
+ DataAbort chan struct{}
+ mu sync.Mutex
BundleErr error
Review Comment:
Done.
##########
sdks/go/pkg/beam/runners/prism/internal/worker/bundle.go:
##########
@@ -98,18 +109,38 @@ func (b *B) LogValue() slog.Value {
slog.String("stage", b.PBDID))
}
+// SetErr sets the bundle error if it is not already set.
+func (b *B) SetErr(err error) {
Review Comment:
Done.
--
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
Re: [PR] [Prism] Fix DEADLINE_EXCEEDED errors caused by worker failures [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #38523:
URL: https://github.com/apache/beam/pull/38523#issuecomment-4511104871
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `63.82979%` with `17 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 58.73%. Comparing base ([`e83a1d5`](https://app.codecov.io/gh/apache/beam/commit/e83a1d5bad9ed96550adff9aa1df40287a24f97f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a7364c4`](https://app.codecov.io/gh/apache/beam/commit/a7364c462a935d850165694a751bf01a5a9d3b62?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 21 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...o/pkg/beam/runners/prism/internal/worker/bundle.go](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&filepath=sdks%2Fgo%2Fpkg%2Fbeam%2Frunners%2Fprism%2Finternal%2Fworker%2Fbundle.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL3dvcmtlci9idW5kbGUuZ28=) | 57.14% | [15 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [sdks/go/pkg/beam/runners/prism/internal/stage.go](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&filepath=sdks%2Fgo%2Fpkg%2Fbeam%2Frunners%2Fprism%2Finternal%2Fstage.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL3ByaXNtL2ludGVybmFsL3N0YWdlLmdv) | 83.33% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/beam/pull/38523?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #38523 +/- ##
=========================================
Coverage 58.72% 58.73%
Complexity 7418 7418
=========================================
Files 1649 1649
Lines 212354 212351 -3
Branches 6330 6330
=========================================
+ Hits 124712 124717 +5
+ Misses 83097 83089 -8
Partials 4545 4545
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/38523/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [go](https://app.codecov.io/gh/apache/beam/pull/38523/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `28.75% <63.82%> (+0.03%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/38523?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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