pkuzmickas opened a new pull request, #41291: URL: https://github.com/apache/beam/pull/41291
`PubsubUnboundedSource` always returns `true` from `requiresDeduping()`, so runners that dedupe the output of unbounded sources (Flink, Spark, Direct, and the SDF-based `Read.Unbounded`) key every message by its record id and shuffle all messages before any other processing. Pipelines that tolerate duplicates can't turn this off, although Dataflow offers at-least-once streaming for the same trade-off. On the Flink runner the shuffle has a large cost for heavy pipelines: the read can't be fused with the transforms after it, unaligned checkpoints store the shuffle's in-flight buffers, and restores replay them. In a Beam 2.72 / Flink 2.3 job doing ~1,500 messages/s on 480 subtasks, turning deduplication off (with a patched source) gave one fused stage, checkpoints of ~10 s with no in-flight data instead of ~2.5 GB, and no throughput loss. Changes: - `PubsubIO.Read.withoutDeduplication()` (default unchanged: dedupe). - `PubsubUnboundedSource` gets a `deduplicate` field (package-private constructor, `getDeduplicate()`); `PubsubSource.requiresDeduping()` returns it. - Javadoc says what's traded: messages Pub/Sub redelivers (expired ack deadlines, restores) reach the pipeline again. Dataflow reads Pub/Sub natively and doesn't use the setting. Not included: the setting isn't part of the portable `PubSubReadPayload`, so portable runners still dedupe. Happy to add that if this direction is OK. Testing: `PubsubUnboundedSourceTest.dedupesByDefault` / `noDedupingWhenTurnedOff`, `PubsubIOTest.testReadWithoutDeduplication`; `PubsubUnboundedSourceTest`, `PubsubIOTest` and `PubSubReadPayloadTranslationTest` pass. -- 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]
