damccorm commented on PR #41013: URL: https://github.com/apache/beam/pull/41013#issuecomment-6843268853
> > Thanks for tracking this down @jbonofre — great root-cause analysis! > > A few high-level thoughts and questions: > > > > 1. **Impact on other `UnboundedSource` IOs (AI helped me find these :)):** The core fix in `Read.java` (`UnboundedSourceAsSDFRestrictionTracker.getProgress()` no longer calling `currentRestriction()` -> `getCheckpointMark()`) actually fixes silent bugs in several other `UnboundedSource` connectors when executed as an SDF (e.g., on Dataflow Runner v2): > > > > * [`MqttIO`](https://github.com/apache/beam/blob/master/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java#L515-L520): `MqttCheckpointMark.Preparer.newCheckpoint()` drains `messages` into the returned `MqttCheckpointMark` and resets `messages = new ArrayList<>()` (identical to `JmsIO`), so every `getProgress()` call was dropping MQTT acks. > > * [`SolaceIO` (`UnboundedSolaceReader`)](https://github.com/apache/beam/blob/master/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java#L298-L307): `getCheckpointMark()` copies and clears `receivedMessages` into a new `PendingCheckpoint` keyed by `checkpointId`. Because `SolaceCheckpointMark.finalizeCheckpoint()` only acks its own `checkpointId`, every `getProgress()` call was orphaning messages in `pendingCheckpoints`. > > * [`SqsUnboundedReader`](https://github.com/apache/beam/blob/master/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/sqs/SqsUnboundedReader.java#L428-L433) and [`PubsubUnboundedSource`](https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java#L1020-L1022): Both increment `numInFlightCheckpoints` on every `getCheckpointMark()` call and only decrement it in `finalizeCheckpoint()` (which gates closing the SQS client / PubSub reader), so discarded checkpoints from `getProgress()` permanently leaked `numInFlightCheckpoints`. > > * [`UnboundedReadFromBoundedSource`](https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/UnboundedReadFromBoundedSource.java#L383-L401): `getCheckpointMark()` splits the underlying `residualSource` and re-initializes the reader on every call. > > All these IOs are backed by UnboundedSource so yeah technical they have same bug > > A good followup (not in scope of this PR) is to check if Python UnboundedSource (added recently) has similar issue Yeah - I was mostly curious if other IOs specifically would run into this particular issue (they do). > Sorry for being late. This LGTM and I'm fine with current form or split it if you prefer. PostCommit Dataflow v2 should already have verified its not breaking existing IO (GCP ones at least) I am not really worried about it breaking other IOs, and I do think this is a net positive. I think it maybe misses other potential issues though and we might need slightly broader changes to `Read.java` to fix them (https://github.com/apache/beam/pull/41013#discussion_r4556472965). Specifically, we call currentRestriction 3 times, starting here - https://github.com/apache/beam/blob/d8084e89d334260719129740933c7af309187f09/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java#L610 - The first instance gets the checkpoint to register with bundleFinalizer (which drains the messages read during the bundle into Checkpoint 1). - The second one calls tracker.currentRestriction().getCheckpoint() again instead of reusing the checkpoint variable from line 617, creating Checkpoint 2 and immediately throwing it away. - the Third returns ProcessContinuation.resume(), which causes the harness/runner to immediately call tracker.trySplit(0), which calls currentRestriction() a third time (Checkpoint 3) and returns that as the residual restriction for the next bundle. So bundleFinalizer gets Checkpoint 1, Checkpoint 2 gets dropped, and the runner persists Checkpoint 3. Checkpoints 2 and 3 are likely empty, which is ok for JmsIO with these changes, but problematic for other IOs. As examples: SqsUnboundedReader and PubsubUnboundedSource don't clear their safeToDeleteIds / safeToAckIds until finalizeCheckpoint(), and they unconditionally call numInFlightCheckpoints.incrementAndGet() on every getCheckpointMark() call. Because Checkpoints 2 and 3 are never registered with bundleFinalizer, numInFlightCheckpoints never drops back to 0 (which prevents cleanup from 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
