[BEAM-5605] Add support for channel splitting to the gRPC read "source" and propagate "split" calls to the downstream receiver #10501
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…e" and propagate "split" calls to the downstream receiver This code mirrors the logic/implementation within https://github.com/apache/beam/blob/16757ef9a6da4d0ac218c6c4d6b19e2a49ccca45/sdks/python/apache_beam/runners/worker/bundle_processor.py#L206 To be able to propagate the split call to the downstream receiver, I collapsed all the harness FnDataReceiver types into two existing implementations and one new implementations. The previous hierarchy was: element counting receiver -> time counting receiver -> multiplexing receiver (possibly the original receiver) The current implementation combined the element counting, time counting and multiplexing into the MultiplexingMetricTrackingFnDataReceiver while for the singleton case into the MetricTrackingFnDataReceiver. To propagate splits, a SplittingMetricTrackingFnDataReceiver was created that extends the MetricTrackingFnDataReceiver. Note, like in Python, there is currently no support for splitting as in https://github.com/apache/beam/blob/c167d8ef99b21148bcab7c37538a6ef2f64864c7/sdks/python/apache_beam/runners/worker/operations.py#L133
lukecwik
force-pushed
the
splittabledofn
branch
3 times, most recently
from
d0eb6dd
to
d541573
Compare
|
Run Java PreCommit |
lukecwik
changed the title
robertwb
approved these changes
Jan 11, 2020
| @@ -131,6 +137,11 @@ | |||
| private final BeamFnDataClient beamFnDataClient; | |||
| private final Coder<WindowedValue<OutputT>> coder; | |||
|
|
|||
| private final Object splittingLock = new Object(); | |||
| // 0-based count of the number of elements | |||
There was a problem hiding this comment.
0 based index of the current element being processed(?).
| private final Object splittingLock = new Object(); | ||
| // 0-based count of the number of elements | ||
| private long index = -1; | ||
| // 0-based count of the number of elements |
There was a problem hiding this comment.
0-based index of the first element to not process. (Or is this the last element to process?)
There was a problem hiding this comment.
first element to not process.
| totalBufferSize = stopIndex; | ||
| } | ||
|
|
||
| // In the case where we have yet to process an element, set the current element progress to 1. |
There was a problem hiding this comment.
Should this be an else clause below?
There was a problem hiding this comment.
Its logically the same where the else clause is the default based upon what we initialize.
|
|
||
| // Compute the amount of "remaining" work that we know of. | ||
| double remainder = totalBufferSize - index - currentElementProgress; | ||
| // Compute the fraction of work that we should "keep". |
There was a problem hiding this comment.
Compute the number of elements that we should "keep."
|
Run Java PreCommit |
1 similar comment
|
Run Java PreCommit |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This code mirrors the logic/implementation within Python.
To be able to propagate the split call to the downstream receiver, I collapsed all the harness
FnDataReceivertypes into two existing implementations and one new implementation.The previous hierarchy was:
The current implementation combined the element counting, time counting and multiplexing into the
MultiplexingMetricTrackingFnDataReceiverwhile for the singleton case into theMetricTrackingFnDataReceiver.To propagate splits, a
SplittingMetricTrackingFnDataReceiverwas created that extends the MetricTrackingFnDataReceiver. Note, like in Python, there is currently no support for splitting a multiplexed consumer.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.See the Contributor Guide for more tips on how to make review process smoother.
Post-Commit Tests Status (on master branch)
Pre-Commit Tests Status (on master branch)
See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.