You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/03/26 07:26:09 UTC
[GitHub] [beam] robertwb opened a new pull request #11235: [WIP] Pull bundle
descriptors.
robertwb opened a new pull request #11235: [WIP] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] 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://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401946646
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
+ BeamFnApi.GetProcessBundleDescriptorRequest request,
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) {
+ final String workerId = headerAccessor.getSdkWorkerId();
+ if (Strings.isNullOrEmpty(workerId)) {
Review comment:
We should check that we are not closed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r404183803
##########
File path: sdks/python/apache_beam/runners/worker/worker_id_interceptor.py
##########
@@ -48,17 +49,23 @@ def __init__(self, worker_id=None):
if worker_id:
self._worker_id = worker_id
+ def intercept_unary_unary(self, continuation, client_call_details, request):
+ return self._intercept(continuation, client_call_details, request)
Review comment:
Removing the worker id means that process bundle descriptors need to be unique per control server instance.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401948665
##########
File path: sdks/python/apache_beam/runners/worker/worker_id_interceptor.py
##########
@@ -48,17 +49,23 @@ def __init__(self, worker_id=None):
if worker_id:
self._worker_id = worker_id
+ def intercept_unary_unary(self, continuation, client_call_details, request):
Review comment:
You might as well and implement the UnaryStreamClientInterceptor and StreamUnaryClientInterceptor now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r404241747
##########
File path: runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolServiceTest.java
##########
@@ -120,4 +121,16 @@ public void onCompleted() {
latch.await();
assertThat(sawComplete.get(), is(true));
}
+
+ @Test
+ public void testUnknownBundle() throws Exception {
+ BeamFnApi.GetProcessBundleDescriptorRequest request =
+ BeamFnApi.GetProcessBundleDescriptorRequest.newBuilder()
+ .setProcessBundleDescriptorId("missing")
+ .build();
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver = mock(StreamObserver.class);
+ controlService.getProcessBundleDescriptor(request, responseObserver);
+
+ verify(responseObserver).onError(any(StatusException.class));
Review comment:
Validate the "not found" shows up in the status message and/or status code to ensure that we got the right StatusException
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401946210
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
+ BeamFnApi.GetProcessBundleDescriptorRequest request,
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) {
+ final String workerId = headerAccessor.getSdkWorkerId();
+ if (Strings.isNullOrEmpty(workerId)) {
+ // TODO(BEAM-4149): Enforce proper worker id.
+ LOG.warn("No worker_id header provided in getProcessBundleDescriptor request");
+ }
+
+ LOG.info("getProcessBundleDescriptor request with id {}", workerId);
Review comment:
```suggestion
LOG.info("getProcessBundleDescriptor: descriptor {} worker id {}", request.getProcessBundleDescriptorId(), workerId);
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401954623
##########
File path: sdks/python/apache_beam/runners/worker/sdk_worker.py
##########
@@ -112,7 +113,11 @@ def __init__(self,
self._state_handler_factory = GrpcStateHandlerFactory(
self._state_cache, credentials)
self._profiler_factory = profiler_factory
- self._fns = {} # type: Dict[str, beam_fn_api_pb2.ProcessBundleDescriptor]
+ self._fns = KeyedDefaultDict(
+ lambda id: self._control_stub.GetProcessBundleDescriptor(
+ beam_fn_api_pb2.GetProcessBundleDescriptorRequest(
Review comment:
What appears in the default dict if GetProcessBundleDescriptor raises an error because the call fails?
Does it get propagated to the caller and it fails processing the bundle (is anything left in the map)?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401952609
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
Review comment:
Please add failure tests to FnApiControlClientPoolService for NOT_FOUND and missing worker id.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on issue #11235: [BEAM-9618] Pull bundle
descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on issue #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#issuecomment-609943402
Fixed typo and strengthened test.
I actually did think about GCing based on worker ID, but don't think this'll be a problem in practice as the lifetimes tend to be the same (and it is still possible to do if the server wants to keep track of what bundle processor descriptors are used by which clients, even if they don't send worker IDs, which is also being fixed).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on issue #11235: [BEAM-9618] Pull bundle
descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on issue #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#issuecomment-608674803
@lukecwik Was there anything more you wanted me to do on this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on issue #11235: [BEAM-9618] Pull bundle
descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on issue #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#issuecomment-604762162
R: @lukecwik this is rebased and should be ready for review. I will remove commit cebab89 before merging.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11235: [WIP]
Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11235: [WIP] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r398805859
##########
File path: model/fn-execution/src/main/proto/beam_fn_api.proto
##########
@@ -81,6 +81,13 @@ service BeamFnControl {
returns (
// A stream of instructions requested of the SDK to be performed.
stream InstructionRequest) {}
+
+ rpc GetProcessBundleDescriptor(GetProcessBundleDescriptorRequest) returns (
Review comment:
The current Control must be a stream because (1) it is server-initiated requests on a client-initiated connection and (2) in streaming pipelines, it we want it to be high-volume, low-latency, possibly pipelined. However, using streams in this way vs. "standard" RPC methods comes at a cost (in both code and conceptual complexity).
As far as ordering, it is preferable to have a protocol that is not sensitive to this kind of thing. (In fact, this eliminates the one need for ordering in our Control stream, where bundles needed to be registered before they were used (and this didn't fit well with pattern of dispatching instructions to threadpool-backed handlers in the SDKs which required extra locking/waiting anyway). Progress and splitting requests were already asynchronous.)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401943249
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
+ BeamFnApi.GetProcessBundleDescriptorRequest request,
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) {
+ final String workerId = headerAccessor.getSdkWorkerId();
+ if (Strings.isNullOrEmpty(workerId)) {
+ // TODO(BEAM-4149): Enforce proper worker id.
+ LOG.warn("No worker_id header provided in getProcessBundleDescriptor request");
Review comment:
Since this is a new method, we should force all users to supply it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401945553
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
+ BeamFnApi.GetProcessBundleDescriptorRequest request,
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) {
+ final String workerId = headerAccessor.getSdkWorkerId();
+ if (Strings.isNullOrEmpty(workerId)) {
+ // TODO(BEAM-4149): Enforce proper worker id.
+ LOG.warn("No worker_id header provided in getProcessBundleDescriptor request");
+ }
+
+ LOG.info("getProcessBundleDescriptor request with id {}", workerId);
+
+ FnApiControlClient client;
+ synchronized (lock) {
+ client = vendedClients.get(workerId);
+ }
+ LOG.warn("getProcessBundleDescriptor client {}", client);
+ LOG.warn("getProcessBundleDescriptor id {}", request.getProcessBundleDescriptorId());
+ LOG.warn(
+ "getProcessBundleDescriptor result {}",
+ client.getProcessBundleDescriptor(request.getProcessBundleDescriptorId()));
+ responseObserver.onNext(
+ client.getProcessBundleDescriptor(request.getProcessBundleDescriptorId()));
Review comment:
Please return the NOT_FOUND status error if the process bundle descriptor is null.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401946261
##########
File path: runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientPoolService.java
##########
@@ -105,12 +105,38 @@ public static FnApiControlClientPoolService offeringClientsToPool(
return newClient.asResponseObserver();
}
+ @Override
+ public void getProcessBundleDescriptor(
+ BeamFnApi.GetProcessBundleDescriptorRequest request,
+ StreamObserver<BeamFnApi.ProcessBundleDescriptor> responseObserver) {
+ final String workerId = headerAccessor.getSdkWorkerId();
+ if (Strings.isNullOrEmpty(workerId)) {
+ // TODO(BEAM-4149): Enforce proper worker id.
+ LOG.warn("No worker_id header provided in getProcessBundleDescriptor request");
+ }
+
+ LOG.info("getProcessBundleDescriptor request with id {}", workerId);
+
+ FnApiControlClient client;
+ synchronized (lock) {
+ client = vendedClients.get(workerId);
+ }
+ LOG.warn("getProcessBundleDescriptor client {}", client);
+ LOG.warn("getProcessBundleDescriptor id {}", request.getProcessBundleDescriptorId());
+ LOG.warn(
+ "getProcessBundleDescriptor result {}",
+ client.getProcessBundleDescriptor(request.getProcessBundleDescriptorId()));
Review comment:
Did you intend to keep all these logs?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r402086684
##########
File path: sdks/python/apache_beam/runners/worker/sdk_worker.py
##########
@@ -112,7 +113,11 @@ def __init__(self,
self._state_handler_factory = GrpcStateHandlerFactory(
self._state_cache, credentials)
self._profiler_factory = profiler_factory
- self._fns = {} # type: Dict[str, beam_fn_api_pb2.ProcessBundleDescriptor]
+ self._fns = KeyedDefaultDict(
+ lambda id: self._control_stub.GetProcessBundleDescriptor(
+ beam_fn_api_pb2.GetProcessBundleDescriptorRequest(
Review comment:
The error gets raised and nothing gets into the map.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r404242047
##########
File path: runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/FnApiControlClientTest.java
##########
@@ -175,4 +179,23 @@ public void testOnCloseMultipleListener() throws Exception {
verify(mockConsumer1).accept(client);
verify(mockConsumer2).accept(client);
}
+
+ @Test
+ public void testRegisterNundle() {
Review comment:
```suggestion
public void testRegisterBundle() {
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r404183803
##########
File path: sdks/python/apache_beam/runners/worker/worker_id_interceptor.py
##########
@@ -48,17 +49,23 @@ def __init__(self, worker_id=None):
if worker_id:
self._worker_id = worker_id
+ def intercept_unary_unary(self, continuation, client_call_details, request):
+ return self._intercept(continuation, client_call_details, request)
Review comment:
Removing the worker id means that process bundle descriptors need to be unique per control server instance.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235: [WIP]
Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [WIP] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r398918303
##########
File path: model/fn-execution/src/main/proto/beam_fn_api.proto
##########
@@ -81,6 +81,13 @@ service BeamFnControl {
returns (
// A stream of instructions requested of the SDK to be performed.
stream InstructionRequest) {}
+
+ rpc GetProcessBundleDescriptor(GetProcessBundleDescriptorRequest) returns (
Review comment:
sg
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb merged pull request #11235: [BEAM-9618] Pull bundle
descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb merged pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401955503
##########
File path: sdks/python/apache_beam/runners/worker/worker_id_interceptor.py
##########
@@ -35,7 +35,8 @@ class _ClientCallDetails(collections.namedtuple(
pass
-class WorkerIdInterceptor(grpc.StreamStreamClientInterceptor):
+class WorkerIdInterceptor(grpc.UnaryUnaryClientInterceptor,
Review comment:
test?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] robertwb commented on issue #11235: [BEAM-9618] Pull bundle
descriptors.
Posted by GitBox <gi...@apache.org>.
robertwb commented on issue #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#issuecomment-610017389
Run Python PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235: [WIP]
Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [WIP] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r398689421
##########
File path: model/fn-execution/src/main/proto/beam_fn_api.proto
##########
@@ -81,6 +81,13 @@ service BeamFnControl {
returns (
// A stream of instructions requested of the SDK to be performed.
stream InstructionRequest) {}
+
+ rpc GetProcessBundleDescriptor(GetProcessBundleDescriptorRequest) returns (
Review comment:
I would have expected this to be a separate stream containing SDK requests similar to how Control works or made Control bidirectional for requests and responses by making a singular `Instruction` field that is streamed both ways that contains both Request and Response fields.
The advantage of using a stream is that it may be important to provide ordering over SDK requests/responses. Putting it into the Control stream allows for a total ordering between all Runner and SDK requests/responses (to solve the instruction id collision issue, we can assign servers the "s" prefix and clients the "c" prefix).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401952191
##########
File path: sdks/python/apache_beam/runners/worker/worker_id_interceptor.py
##########
@@ -48,17 +49,23 @@ def __init__(self, worker_id=None):
if worker_id:
self._worker_id = worker_id
+ def intercept_unary_unary(self, continuation, client_call_details, request):
+ return self._intercept(continuation, client_call_details, request)
Review comment:
Hmm, each RPC will need to pass the worker_id which is a minor argument for bidi stream but not worth the complexity of it compared to the unary RPC. We also don't expect this to happen a lot.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [beam] lukecwik commented on a change in pull request #11235:
[BEAM-9618] Pull bundle descriptors.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #11235: [BEAM-9618] Pull bundle descriptors.
URL: https://github.com/apache/beam/pull/11235#discussion_r401948041
##########
File path: sdks/python/apache_beam/runners/worker/sdk_worker_test.py
##########
@@ -115,8 +115,9 @@ def _check_fn_registration_multi_request(self, *args):
{item.id: item
for item in process_bundle_descriptors})
- def test_fn_registration(self):
- self._check_fn_registration_multi_request((1, 4), (4, 4))
+
+# def test_fn_registration(self):
+# self._check_fn_registration_multi_request((1, 4), (4, 4))
Review comment:
if we can, we should test the registration lookup failure case.
The success case will be easily be covered by other tests.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services