StateWatcher watches and reports changed Pipeline State

[damondouglas]

This PR closes #32032 with a package private StateWatcher. It sends a GetJobStateRequest to the state stream endpoint of the Job Management service and reports any changes in State to listeners, invoking onStateChanged.

---

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.

[damondouglas]

R: @Abacn or @ahmedabu98

[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

[Abacn]

I tend to think a dedicate Listener and watcher isn't necessary. One just need to periodically polls current state in pipelineResult.waitUntilFinish(), and polls state once everytime pipelineResult.getStete() called. This is what existing implementation doing:

Existing runners (direct, dataflow) track pipeline states in following way

getState() either make a call to query state, or return terminal state if it is already at terminal state

DIrect runner:

beam/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java

Line 325 in 56aa17b

if (this.state == State.RUNNING) {

Dataflow runner:

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java

Line 507 in 56aa17b

return getStateWithRetriesOrUnknownOnException(

For waitUntilFinish, it does polls pipeline state periodically until at a terminal state

Direct runner

beam/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ExecutorServiceParallelExecutor.java

Line 268 in 56aa17b

update = visibleUpdates.tryNext(Duration.millis(1L));

Dataflow runner

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java

Line 325 in 56aa17b

BackOffAdapter.toGcpBackOff(STATUS_BACKOFF_FACTORY.withMaxRetries(0).backoff()),

We can simplify pretty much this PR inside delegate.waitUntilFinish

[damondouglas]

@Abacn this is different in that the runner needs to launch an underlying Job Management service written in a different language and shut it down gracefully. The reason for introducing this PR is that relying on existing tools in Beam do not properly shutdown the gRPC channels, the code is too nested to safely refactor. This has caused errors in relying on proper shutdown, even when calling waitUntilFinish. The result is that the underlying process remains running without owndership and the only way to end the process is to use lsof, for example followed by kill.

Please review this assuming it is needed.

[Abacn]

Thanks for the explanation. I didn't aware of the core context/consideration of this change involves gracefully closing gRPC channel. The change itself is clear and well organized as well as the test. This LGTM