Abacn commented on PR #41013: URL: https://github.com/apache/beam/pull/41013#issuecomment-6842626300
> 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 -- 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]
