Add Lineage metrics to PubsubIO

[Abacn]

Covers

UnboundedReader
PubsubBoundedWriter
PubsubUnboundedWriter (PubsubSink / PubsubDynamicSink)

Please add a meaningful description for your change here

---

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

• 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 or the workflows README to see a list of phrases to trigger workflows.

[Abacn]

tested on integration test (rebased on #32022)

PubsubReadIT:

after

PipelineResult job = pipeline.run();

print

(job.metrics().queryMetrics(MetricsFilter.builder().addNameFilter(MetricNameFilter.named(LINEAGE_NAMESPACE, SOURCE_METRIC_NAME)).build()).getStringSets());

get

StringSetResult{stringSet=[pubsub:topic:pubsub-public-data.taxirides-realtime]}}

PubsubWriteIT:

replace to use SINK_METRIC_NAME, query gets

StringSetResult{stringSet=[pubsub:topic:apache-beam-testing.pubsub-write-1722451794792

[Abacn]

R: @rohitsinha54

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

[rohitsinha54]

Would it be possible to update any of the PubSub integration tests which runs in dataflow runner to assert for lineage metrics?

[Abacn]

We can and should do this after runner support cleared. currently Pubsub integration tests are running on both direct runner / legacy Dataflow / Dataflow v2. When I add asserts currently it will only pass on Legacy runner suite (Direct runner suite will also pass after #32022 in)

[Abacn]

a couple of unit tests failed due to that the fake topic/subscription does not follow the expected topic/subscription format, direct runner integration tests all passed however