Hi Till, Thanks for your reply. I agree point 3 and 4 in your email worth a separated thread to discuss. Let me answer your questions and concerns in point 1 and 2 respectively.
1.Lifecycle of LeaderServer and requirement to implement it LeaderServer starts on cluster entrypoint and its lifecycle is bound to the lifecycle of cluster entrypoint. That is, when the cluster entrypoint starts, a LeaderServer also starts; and LeaderServer gets shut down when the cluster entrypoint gets shut down. This is because we need to provide services discovery during the cluster is running. For implementation part, conceptually it is a service running on cluster entrypoint which holds in memory services information and can be communicatied with. As our internal specific implementation, LeaderServer is an actor running on the actor system running on cluster entrypoint, which is referred as `commonRpcService`. It is just another unfenced rpc endpoint and required no extra changes to the existing interfaces. Apart from LeaderServer, there is another concept in this implementation, the LeaderClient. LeaderClient forwards register request from election service and retrieval service; forwards leader changed message from LeaderServer. As our specific implementation, LeaderClient is an actor and runs on cluster entrypoint, task manager and cluster client. (1). cluster entrypoint The lifecycle of LeaderClient is like LeaderServer. (2). task manager The lifecycle of LeaderClient is bound to the lifecycle of task manager. Specifically, it runs on `rpcService` starts on task manager runner and stops when the service gets shut down. (3). cluster client The lifecycle of LeaderClient is bound to the ClusterClient. With our codebase, only RestClusterClient should do the adaptation. When start ClientHAService based on LeaderClient, it starts a dedicated rpc service on which the LeaderClient runs. The service as well as the LeaderClient gets shut down on RestClusterClient closed, where ClientHAService#close called. It is a transparent implementation inside a specific ClientHAService; thus also, no changes to the existing interfaces. 2. The proposal to replace existing non-ha services Well, I see your concerns on replace existing stable services hurriedly with a new implementation. Here I list the pros and cons of this replacement. If we agree that it does good I can provide an neat and full featured implementation for preview and see concretely what we add and what we gain. For integration, we can then first integrate with MiniCluster and later. pros: + We don't need to pass the address of job manager among slot request. With the new implementation retriever running on task manager registers itself on the LeaderServer which has a global static address. And the retriever retrieves the address of job manager based on JobID. It is not only unify the interfaces #getJobManagerLeaderRetriever, but reduce the cost on job manager switched. Currently, when job manager lost leadership, slots offered to the old job manager are unaware of it immediately. They don't get released until heartbeat from job manager timeout. With LeaderServer based implementation, LeaderServer notifies LeaderClient once job manager lost leadership. + We have a unified implementation in non-ha scenario It can be regarded as a location transparent embedded implementation. + We have a unified view of high-availability services LeaderServer based implementation follows the same view of ZooKeeper based implementation. Since these high-availability services don't have natural difference from one to the other, we can instead naturally handle them under a unified view. I know that we should still configure the address of LeaderServer, but now it is more like connect string of ZooKeeper, instead of address of internal component. In fact, then we can deprecate configuration of job manager port and auto detect a port as we do in ZooKeeper based scenario. cons: - Overhead on client side The overhead of transmitting job manager address is lower enough so I don't list it as valid pros. Correspondingly, messages between actors among existing actor system are regarded as significant overhead. The visible overhead is that we start a dedicated actor system in RestClusterClient. It is due to the implementation of LeaderServer & LeaderClient based on akka. Conceptually it can be any services but we always introduce some overhead. **contrast to current implementations** LeaderServer based implementation can be regarded as a location transparent embedded implementation. Thus there isn't too many contrasts. Also, embedded implementation is used only in MiniCluster scenario where an actor system is already running, so there isn't significant performance concerns. As for pre-configured implementation, named StandaloneHaServices, I agree that it is deadly simple. But apart from the benefit of unification, it is gradually unrealistic to require users pre-configure the port of job manager, especially on cloud native scenario. Although the specific implementation couple the address and port of LeaderServer and that of job manager, it is not a fundamental constraint. Thus, LeaderServer based implementation is more flexible for evolution. Best, tison. Till Rohrmann <trohrm...@apache.org> 于2019年9月9日周一 下午5:37写道： > Hi Tison, > > thanks for starting this discussion. I think your mail includes multiple > points which are worth being treated separately (might even make sense to > have separate discussion threads). Please correct me if I understood things > wrongly: > > 1. Adding new non-ha HAServices: > > Based on your description I could see the "ZooKeeper-light" non-ha > HAServices implementation work. Would any changes to the existing > interfaces be needed? How would the LeaderServer integrate in the lifecycle > of the cluster entrypoint? > > 2. Replacing existing non-ha HAServices with LeaderServer implementation: > > I'm not sure whether we need to enforce that every non-ha HAServices > implementation works as you've described. I think it is pretty much an > implementation detail whether the services talk to a LeaderServer or are > being started with a pre-configured address. I also think that it is fair > to have different implementations with different characteristics and usage > scenarios. As you've said the EmbeddedHaServices are targeted for single > process cluster setups and they are only used by the MiniCluster. > > What I like about the StandaloneHaServices is that they are dead simple > (apart from the configuration). With a new implementation based on the > LeaderServer, the client side implementation becomes much more complex > because now one needs to handle all kind of network issues properly. > Moreover, it adds more complexity to the system because it starts a new > distributed component which needs to be managed. I could see that once the > new implementation has matured enough that it might replace the > EmbeddedHaServices. But I wouldn't start with removing them. > > You are right that due to the fact that we don't know the JM address before > it's being started that we need to send the address with every slot > request. Moreover we have the method #getJobManagerLeaderRetriever(JobID, > defaultJMAddress) on the HAServices. While this is not super nice, I don't > think that this is a fundamental problem at the moment. What we pay is a > couple of extra bytes we need to send over the network. > > Configuration-wise, I'm not so sure whether we gain too much by replacing > the StandaloneHaServices with the LeaderServer based implementation. For > the new implementation one needs to configure a static address as well at > cluster start-up time. The only benefit I can see is that we don't need to > send the JM address to the RM and TMs. But as I've said, I don't think that > this is a big problem for which we need to introduce new HAServices. > Instead I could see that we might be able to remove it once the > LeaderServer HAServices implementation has proven to be stable. > > 3. Configuration of HAServices: > > I agree that Flink's address and port configuration is not done > consistently. I might make sense to group the address and port > configuration under the ha service configuration section. Maybe it makes > also sense to rename ha services into ServiceDiscovery because it also > works in the non-ha case. it could be possible to only configure address > and port if one is using the non-ha services, for example. However, this > definitely deserves a separate discussion and design because one needs to > check where exactly the respective configuration options are being used. > > I think improving the configuration of HAServices is actually orthogonal to > introducing the LeaderServer HAServices implementation and could also be > done for the existing HAServices. > > 4. Clean up of HAServices implementations: > > You are right that some of the existing HAServices implementations are > "dead code" at the moment. They are the result of some implementation ideas > which haven't been completed. I would suggest to start a separate > discussion to discuss what to do with them. > > Cheers, > Till > > On Mon, Sep 9, 2019 at 9:16 AM Zili Chen <wander4...@gmail.com> wrote: > > > Hi devs, > > > > I'd like to start a discussion thread on the topic how we provide > > retrieval services in non-high-availability scenario. To clarify > > terminology, non-high-availability scenario refers to > > StandaloneHaServices and EmbeddedHaServices. > > > > ***The problem*** > > > > We notice that retrieval services of current StandaloneHAServices > > (pre-configured) and EmbeddedHAServices(in-memory) has their > > respective problems. > > > > For pre-configured scenario, we now have a > > getJobManagerLeaderRetriever(JobID, defaultJMAddress) method > > to workaround the problem that it is impossible to configure JM > > address previously. The parameter defaultJMAddress is not in use in > > any other defaultJMAddress with any other high-availability mode. > > Also in MiniCluster scenario and anywhere else leader address > > pre-configure becomes impossible, StandaloneHAServices cannot be used. > > > > For in-memory case, it is clearly that it doesn't fit any distributed > > scenario. > > > > ***The proposal*** > > > > In order to address the inconsistency between pre-configured retrieval > > services and zookeeper based retrieval services, we reconsider the > > promises provided by "non-high-availability" and regard it as > > similar services as zookeeper based one except it doesn't tolerate > > node failure. Thus, we implement a service acts like a standalone > > zookeeper cluster, named LeaderServer. > > > > A leader server is an actor runs on jobmanager actor system and reacts > > to leader contender register and leader retriever request. If > > jobmanager fails, the leader server associated fails, too, where > > "non-high-availability" stands. > > > > In order to communicate with leader server, we start leader client per > > high-availability services(JM, TM, ClusterClient). When leader > > election service starts, it registers the contender to leader server > > via leader client(by akka communication); when leader retriever > > starts, it registers itself to leader server via leader client. > > > > Leader server handles leader election internally just like Embedded > > implementation, and notify retrievers with new leader information > > when there is new leader elected. > > > > In this way, we unify the view of retrieval services in all scenario: > > > > 1. Configure a name services to communicate with. In zookeeper mode > > it is zookeeper and in non-high-availability mode it is leader server. > > 2. Any retrieval request is sent to the name services and is handled > > by that services. > > > > Apart from a unified view, there are other advantages: > > > > + We need not to use a special method > > getJobManagerLeaderRetriever(JobID, defaultJMAddress), instead, use > > getJobManagerLeaderRetriever(JobID). And so that we need not include > > JobManager address in slot request which might become stale during > > transmission. > > > > + Separated configuration concerns on launch and retrieval. JobManager > > address & port, REST address & port is only configured when launch > > a cluster(even in YARN scenario, no need to configure). And when > > retrieval requested, configure the connect info to name services(zk > > or leader server). > > > > + Embedded implementation could be also included in this abstraction > > without any regression on multiple leader simulation for test purpose. > > Actually, leader server acts as a limited standalone zookeeper > > cluster. And thus, from where this proposal comes from, when we > > refactor metadata storage with transaction store proposed in > > FLINK-10333, we only take care of zookeeper implementation and a > > unified non-high-availability implementation. > > > > ***Clean up*** > > > > It is also noticed that there are several stale & unimplemented > > high-availability services implementations which I'd like to remove for > > a clean codebase work on this thread and FLINK-10333. They are: > > > > - YarnHighAvailabilityServices > > - AbstractYarnNonHaServices > > - YarnIntraNonHaMasterServices > > - YarnPreConfiguredMasterNonHaServices > > - SingleLeaderElectionService > > - FsNegativeRunningJobsRegistry > > > > Any feedback is appreciated. > > > > Best, > > tison. > > >
