pkuzmickas opened a new pull request, #41292: URL: https://github.com/apache/beam/pull/41292
Each `PubsubUnboundedSource` reader pulls up to 1000 messages at a time (`PULL_BATCH_SIZE`) and extends their ack deadlines for at most two minutes (`PROCESSING_TIMEOUT`) after the pull. A reader that emits fewer than ~8 messages/s, for example because it's fused with expensive transforms, can't get through a batch in that time: the rest of the batch expires and Pub/Sub redelivers it, over and over. We hit this on the Flink runner with a CPU-heavy pipeline (~3 messages/s per subtask): 60-85% of deliveries expired. With 100-message pulls (patched source) expirations went to zero. Changes: - `PubsubIO.Read.withPullBatchSize(int)` (default unchanged: 1000; must be positive). - `PubsubUnboundedSource` gets a `pullBatchSize` field (package-private constructor, `getPullBatchSize()`), used by the reader's pull. - Dataflow reads Pub/Sub natively and doesn't use the setting. Not included: the setting isn't part of the portable `PubSubReadPayload`. Happy to add that if this direction is OK. Testing: `PubsubUnboundedSourceTest.pullsTheConfiguredBatchSize` (the reader requests the configured size), `PubsubIOTest.testReadPullBatchSize`; `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]
