Reimplement Pub/Sub Lite's I/O using UnboundedSource.

[dpcollins-google]

SDF is not well supported by the default mode of dataflow, and UnboundedSource is supported by all modes.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• [ ] Choose reviewer(s) and mention them in a comment (R: @username).
• [ ] Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• [ ] Update CHANGES.md with noteworthy changes.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[dpcollins-google]

R: @chamikaramj

[github-actions[bot]]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[codecov[bot]]

Codecov Report

Merging #22612 (17fb9c0) into master (54b0784) will increase coverage by 0.01%. The diff coverage is 63.15%.

:exclamation: Current head 17fb9c0 differs from pull request most recent head 6ccf85d. Consider uploading reports for the commit 6ccf85d to get more accurate results

@@            Coverage Diff             @@
##           master   #22612      +/-   ##
==========================================
+ Coverage   74.19%   74.20%   +0.01%     
==========================================
  Files         706      708       +2     
  Lines       93229    93462     +233     
==========================================
+ Hits        69168    69355     +187     
- Misses      22793    22832      +39     
- Partials     1268     1275       +7     

Flag	Coverage Δ
go	51.52% <56.97%> (+0.02%)	:arrow_up:
python	83.61% <72.56%> (+0.04%)	:arrow_up:

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
sdks/go/pkg/beam/core/runtime/exec/input.go	51.96% <ø> (+0.99%)	:arrow_up:
sdks/go/pkg/beam/runners/direct/direct.go	65.76% <0.00%> (-0.60%)	:arrow_down:
sdks/go/pkg/beam/validate.go	42.10% <0.00%> (ø)
...on/apache_beam/examples/complete/juliaset/setup.py	0.00% <0.00%> (ø)
...m/examples/inference/pytorch_image_segmentation.py	0.00% <ø> (ø)
...ython/apache_beam/examples/kafkataxi/kafka_taxi.py	0.00% <ø> (ø)
sdks/python/apache_beam/examples/sql_taxi.py	0.00% <ø> (ø)
...python/apache_beam/examples/wordcount_xlang_sql.py	0.00% <ø> (ø)
sdks/python/apache_beam/io/azure/blobstorageio.py	26.88% <0.00%> (ø)
...thon/apache_beam/ml/inference/pytorch_inference.py	0.00% <0.00%> (ø)
... and 32 more

:mega: Codecov can now indicate which changes are the most critical in Pull Requests. Learn more

[chamikaramj]

Probably someone from Dataflow streaming team should take a quick look as well. @scwhittle will you be able to check ?

[chamikaramj]

Run PostCommit_Java_Dataflow

[chamikaramj]

Run PostCommit_Java_DataflowV2

[chamikaramj]

Thanks Sam. I think we can merge this when tests pass. LGTM.

Based on an offline discussion with Daniel, we'll be adding a transform override in a follow-up PR to keep SDF version the default while overriding with UnboundedSource-based version for production (Runner v1) Dataflow.

[dpcollins-google]

Run Java_GCP_IO_Direct PreCommit

[chamikaramj]

This failure might be related ? https://ci-beam.apache.org/job/beam_PreCommit_Java_GCP_IO_Direct_Phrase/2/testReport/junit/org.apache.beam.sdk.io.gcp.pubsublite/ReadWriteIT/testReadWrite/

[chamikaramj]

Run PostCommit_Java_Dataflow

[dpcollins-google]

Run PostCommit_Java_Dataflow

[dpcollins-google]

Run PostCommit_Java_DataflowV2

[chamikaramj]

Run Java_GCP_IO_Direct PreCommit

[dpcollins-google]

Run PostCommit_Java_DataflowV2

[dpcollins-google]

Run PostCommit_Java_Dataflow

[chamikaramj]

Pls fix spotless.

@@ -18,7 +18,6 @@ 17:22:48 package·org.apache.beam.sdk.io.gcp.pubsublite.internal; 17:22:48
17:22:48 import·static·org.junit.Assert.assertEquals; 17:22:48 -import·static·org.junit.Assert.assertThrows; 17:22:48 import·static·org.mockito.Mockito.verify;

[dpcollins-google]

Run PostCommit_Java_DataflowV2

[dpcollins-google]

Run PostCommit_Java_Dataflow

[dpcollins-google]

Pls fix spotless.

@@ -18,7 +18,6 @@ 17:22:48 package·org.apache.beam.sdk.io.gcp.pubsublite.internal; 17:22:48 17:22:48 import·static·org.junit.Assert.assertEquals; 17:22:48 -import·static·org.junit.Assert.assertThrows; 17:22:48 import·static·org.mockito.Mockito.verify;

Done.