m-trieu opened a new pull request, #28835:
URL: https://github.com/apache/beam/pull/28835
Add interfaces/classes for direct path:
**ProcessWorkItemClient**
Exposed to `ProcessWorkItem` to give access to route GetData and CommitWork
stream RPCs to the same workers where GetWork was called (currently
`StreamingDataflowWorker#process`).
**ProcessWorkItem**
Replaces `WorkItemReceiver`, same but takes and exposes
`ProcessWorkItemClient` instead of `WorkItem`. Since `ProcessWorkItemClient`
needs a way to get data for work, refresh work, get side input data, and commit
work, the place where its created (`GrpcGetWorkStream`) needs to be modified to
accept GetDataStream (for keyed/state data), GetDataStream (global side input
data), and a CommitWorkStream.
**GetWorkBudget**
A struct to model `item` and `byte` budgets for how much work a user worker
can handle. This is passed in `GetWorkRequest`(s) to Windmill to control how
many items/bytes of `Work` is returned.
**GetWorkBudgetDistributor**
Given a set of `WindmillStreamSender`(s) and `GetWorkBudget`, distributes
the budgets to the `WindmillStreamSender`(s) in some manner.
**WindmillStreamSender**
When the `Grpc*Stream`(s) are created, they immediately start the underlying
grpc stream (`startStream` is called, and has protected access). To be able to
assign budgets and get the streams ready to be started (similar to
`GetWorkClientSender`), `WindmillStreamSender` wraps the 3 WorkItem API RPC
streams, and exposes a `startStream`, and `closeAllStreams` to manage the
underlying streams. Once the streams are started they are cached (via thread
safe memoization). Once certain endpoints are stale, the `closeAllStreams`
gives a way to close all of the underlying streams un a `WindmillStreamSender`
instance. This will also be an interface to adjust budget in the
`GetWorkStream` as well as track windmill/backend worker `backlog` in the
future.
**DispatcherClient**
Manages/vends out stubs and the dispatcher
Thread safe via synchronization on reads and writes.
Add **StreamingEngineClient**
Future changes need still:
- Have `GrpcGetWorkStream` accept a `GrpcGetDataStream` and
`GrpcCommitWorkStream` so that it can construct a `ProcessWorkItemClient` and
pass it onto the `ProcessWorkItem` (replaces current behavior where`WorkItem`
being passed to the `WorkItemReceiver`).
- Integrate with `StreamingDataflowWorker`, might be worth having 2
different implementations of `StreamingDataflowWorker` since current
`MetricTrackingWindmillServer` is used for `GetData` (keyed and global) fetches.
- `GetWorkBudgetDistributor` implementation that distributes the budget
evenly
- A way to trigger budget rebalances in `StreamingEngineClient`. Once we
receive a change in the endpoint mappings, we need a way to redistribute the
budget to the new worker set. I have been looking for a publish/subscribe or
observable pattern to do this.
- Need to figure out a way to batch commits since they need to go to the
same origin worker
R: @scwhittle
------------------------
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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit
[https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more
information about GitHub Actions CI or the [workflows
README](https://github.com/apache/beam/blob/master/.github/workflows/README.md)
to see a list of phrases to trigger workflows.
--
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]
