You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by wenlong88 <gi...@git.apache.org> on 2016/08/17 14:46:06 UTC
[GitHub] flink pull request #2381: [FLINK-4414][cluster management] Remove restrictio...
GitHub user wenlong88 opened a pull request:
https://github.com/apache/flink/pull/2381
[FLINK-4414][cluster management] Remove restriction on RpcService.getAddress
currently RpcService provide only address of the endpoint, I think rpc service serve both the endpoint create on it and the remote gateway create on it, so it is ok to offer the getAddress to all RpcGateway created on the rpc service including the server and client.
some more changes included, since ther merged in code base:
1. remove restriction of getAddress in RpcService
2. add support of Endpoint inherit and RpcGateway interface inherit
3. add some unittest for AkkaRpcService, remove JobMaster Register Test
4. add error handle method RpcEndpoint, which is a common function for RpcEndpoint
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/alibaba/flink rpc_improvement
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2381.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2381
----
commit 1971f26cc5d0afa4022b3daa93eacd9f9541b072
Author: wenlong.lwl <we...@alibaba-inc.com>
Date: 2016-08-17T14:09:54Z
1. remove restriction of getAddress in RpcService
2. add support of Endpoint inherit and RpcGateway interface inherit
3. add some unittest for AkkaRpcService, remove JobMaster Register Test
4. add error handle method RpcEndpoint, which is a common function for RpcEndpoint
commit bd1f8b1877ca6e7df6e7ed73d70846d6e598c06b
Author: wenlong.lwl <we...@alibaba-inc.com>
Date: 2016-08-17T14:18:23Z
remove error process method since had been added in RpcEndpoint
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2381: [FLINK-4414][cluster management] Remove restrictio...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2381#discussion_r75140948
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java ---
@@ -197,11 +201,11 @@ public void stopService() {
}
@Override
- public String getAddress(RpcGateway selfGateway) {
+ public String getAddress(RpcGateway gateway) {
checkState(!stopped, "RpcService is stopped");
- if (selfGateway instanceof AkkaGateway) {
- ActorRef actorRef = ((AkkaGateway) selfGateway).getRpcEndpoint();
+ if (gateway instanceof AkkaGateway) {
+ ActorRef actorRef = ((AkkaGateway) gateway).getActorRef();
return AkkaUtils.getAkkaURL(actorSystem, actorRef);
--- End diff --
This will be problematic if you pass in a local actor ref which has been created by a different actor system. Because then you will return a wrong remote address.
So for example, the following test would fail:
```
@Test
public void test() {
LinkedBlockingQueue<Object> linkedBlockingQueue = new LinkedBlockingQueue<>();
TestEndpoint testEndpoint = new TestEndpoint(akkaRpcService1,linkedBlockingQueue);
testEndpoint.start();
TestEndpoint testEndpoint2 = new TestEndpoint(akkaRpcService2, linkedBlockingQueue);
testEndpoint2.start();
assertEquals(testEndpoint2.getAddress(), akkaRpcService1.getAddress(testEndpoint2.getSelf()));
assertEquals(testEndpoint.getAddress(), akkaRpcService2.getAddress(testEndpoint.getSelf()));
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2381
Maybe it makes more sense to extends `RpcGateway` interface and offer a getAddress method which returns the remote address of the gateway. We should also add tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on the issue:
https://github.com/apache/flink/pull/2381
ok, I will open a issue for change RpcCompletenessTest to support inheritance of RpcGateway and RpcEndpoint later.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on the issue:
https://github.com/apache/flink/pull/2381
Hi, @tillrohrmann
1. It looks good if you think it is not a bit deal that the interface RpcGateway contains functions defined for specific business logic and functions defined for frameworks logic which right now only contains `getAddress`. I treat it some kind of weird because I think the RpcGateway interface is the business logic protocol of a RPC server, just like what we do in hadoop rpc framework which also defines a interface to define what can do for a RPC server like the RPC abstraction here.
2. I find that in current AkkaRpcService has another problem that if we use a RpcService to get a Endpoint's self address, but the endpoint is not created by the RpcService, we will get a wrong too, just like the example you make. It need to do some check if we want to reserve the function. and you commit solve the problem too.
3. Sorry for the codes left behind in last revert. I have removed it. But I think if the concern in 1 is not a problem, your solution is better, and the PR can be discarded
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2381: [FLINK-4414][cluster management] Remove restrictio...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 closed the pull request at:
https://github.com/apache/flink/pull/2381
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2381
I'm actually not so sure whether it's such a big problem for the annotation check. You simply don't consider the methods which have been declared by `RpcGateway` and you're done. In this [commit](https://github.com/tillrohrmann/flink/commit/fc80d930dbf045592ca52b9e76fcafb260c13707) you can see how I would do it. If this commit fulfils your requirements, then we can merge it.
This PR still contains unrelated changes such as the `RpcCompletenessTest` improvements. This should be treated in a separate PR. The best thing is to either do a force push or to close an re-open a PR in this case (then you get rid of the revert commit).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on the issue:
https://github.com/apache/flink/pull/2381
If there is a `RpcGateway.getAddress`, it will be more friendly in api. But I also think mixing the user defined RPC methods and framework interfaces will be a bit of odd and needs specific processing in annotation check.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on the issue:
https://github.com/apache/flink/pull/2381
Coming back to the initial motivation of this PR, It is because that I tried to get the address from the RpcService, and the RpcService provides a api `getAddress(RpcGateway gateway)` but can noly provide the address of endpoint created on it only.
Another solution just for discuss:
1. change function prototype to `getAddress(RpcEndpoint endpoint)`
2. the AkkaRpcService can adapt this change easily, since we can get actorRef from RpcEndpoint, and do what is done now. Besides, AkkaRpcService need to check whether the actorRef is in the cached actors and raise a exception if not.
In this way, we cannot get address from gateway, but it also make sense since the endpoint is server , which must have a address to visit, and the gateway is a client, providing an address is an optional.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2381
Thanks for your contribution @wenlong88. The PR mixes multiple things together. It would be better if you could separate these changes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on the issue:
https://github.com/apache/flink/pull/2381
@tillrohrmann I agrees that `getAddress` in gateway can provide some facilities, I think that we can treat a special method `getAddress` executed locally just like the timeout argument, which is a special argument annotated with `RpcTimeout`. so just feel free to merge the commit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2381
Alright, then I'll merge the PR and close the issue. You can then close this PR. For the other things you've changed like the `RpcCompletenessTest`, please open a new PR (one PR for each issue).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #2381: [FLINK-4414][cluster management] Remove restriction on Rp...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2381
Hi @wenlong88,
1. I'm actually not so sure whether `RpcGateway.getAddress` method is something framework specific. For any `RpcService` implementation you need some kind of address to connect to a `RpcEndpoint`. The `RpcService` has the method `connect` which requires exactly this information to open a connection. Thus, I think that every `RpcGateway` should be able to tell to which `RpcEndpoint` it is currently connected. So for example, there is currently also no way to open another `RpcGateway` to the same `RpcEndpoint` given a `RpcGateway`. What do you think?
2. You're right, the current implementation has the same problem and should be fixed either way.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #2381: [FLINK-4414][cluster management] Remove restrictio...
Posted by wenlong88 <gi...@git.apache.org>.
Github user wenlong88 commented on a diff in the pull request:
https://github.com/apache/flink/pull/2381#discussion_r75157339
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcService.java ---
@@ -197,11 +201,11 @@ public void stopService() {
}
@Override
- public String getAddress(RpcGateway selfGateway) {
+ public String getAddress(RpcGateway gateway) {
checkState(!stopped, "RpcService is stopped");
- if (selfGateway instanceof AkkaGateway) {
- ActorRef actorRef = ((AkkaGateway) selfGateway).getRpcEndpoint();
+ if (gateway instanceof AkkaGateway) {
+ ActorRef actorRef = ((AkkaGateway) gateway).getActorRef();
return AkkaUtils.getAkkaURL(actorSystem, actorRef);
--- End diff --
yeah, you are right. It is a problem I have missed checking. Added the connected actor to the caches in case of mixed actor system to make current solution work
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---