johannaojeling opened a new issue, #29210: URL: https://github.com/apache/beam/issues/29210
### What happened? # Problem In SDFs, `ProcessSizedElementsAndRestrictions.GetProgress()` can return `math.NA()` if done and remaining are both 0: https://github.com/apache/beam/blob/714badb821b7acfb9801a69dd0aff08bc84f6b4f/sdks/go/pkg/beam/core/runtime/exec/sdf.go#L925-L926 ```go d, r := n.rt.GetProgress() // d, r := 0, 0 frac := d / (d + r) // frac := 0 / (0 + 0) ``` This causes downstream problems if using an `offsetrange.GrowableTracker` with an unbounded `offsetrange.Restriction`, as `math.NA()` is then being passed as the fraction to `TrySplit()`. The function returns a primary restriction with its `End` smaller than its `Start`. This in turn leads to a negative return value when `Size()` is invoked, which causes the pipeline to fail. # Example use case Reading messages from a stream fails at checkpointing if no new messages have been published and read since the last checkpoint. ## Given - 5 messages have been published on stream - no new messages are published since the pipeline starts executing - the restriction is based on the sequence number of a message - the read is configured to start from seq no 1 - the estimated end (exclusive) is the seq no of the last published message + 1 - the SDF initiates a self-checkpoint with process continuation resume when there are no new messages to process ## Initial state Rest: {Start: 1, End: math.MaxInt64} Tracker: {attempted: -1, claimed: 0} Estimated end: 6 ## At checkpoint 1 Rest: {Start: 1, End: math.MaxInt64} Tracker: {attempted: 5, claimed: 5} Estimated end: 6 → fraction: 0.5 → splitPt: 6 → primary: {Start: 1, End: 6} → residual: {Start: 6, End: math.MaxInt64} ## At checkpoint 2 Rest: {Start: 6, End: math.MaxInt64} Tracker: {attempted: -1, claimed: 5} Estimated end: 6 → fraction: math.NA() → splitPt: 5 → primary: {Start: 6, End: 5} → residual: {Start: 5, End: math.MaxInt64} → primary.Size() < 0 → error <details> <summary>Logs</summary> ``` 2023/10/31 07:48:33 WARN SDK Error from split, aborting splits bundle.ID=inst004 bundle.stage=stage-003 bundle.watermark=1969-12-31T23:59:59.999Z error="split[inst004] error from SDK: unable to split inst004: \tAttempting split in ProcessSizedElementsAndRestrictions\n\tSDF.ProcessSizedElementsAndRestrictions[natsio.readFn] UID:4 Out:[2]\nsize returned expected to be non-negative but received -1." 2023/10/31 07:48:33 ERROR unable to split inst004: Attempting split in ProcessSizedElementsAndRestrictions SDF.ProcessSizedElementsAndRestrictions[natsio.readFn] UID:4 Out:[2] ``` </details> ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [ ] Component: Python SDK - [ ] Component: Java SDK - [X] Component: Go SDK - [ ] Component: Typescript SDK - [ ] Component: IO connector - [ ] Component: Beam YAML - [ ] Component: Beam examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Samza Runner - [ ] Component: Twister2 Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
