[FLINK-4670] [cluster management] Add watch mechanism on current RPC framework

[beyond1920]

This pr aims at adding watch mechanism on current RPC framework, so RPC gateway could be watched to make sure the rpc server is running just like previous DeathWatch in akka.
There are following main differences:

1. Add watch and unwatch method to RpcEndpoint class
2. Implement WatchOperationHandler in AkkaInvocationHandler, so it could handle watch and unwatch operation

[tillrohrmann]

The linked JIRA issue seems to be wrong.

I'm not completely sure whether we want to add this functionality to the RpcService or not. At the moment it is easy to implement because we can rely on Akka's death watch functionality. But future implementations might work differently. Thus, we would be bound to reimplement the death watch functionality.

Furthermore, once a target actor system has been marked as "dead" by the deathwatch it will be quarantined. If it was marked wrongly, then there is no way for the target actor system to talk to the watching system and vice versa.

What's the benefit over having a separate service for the heartbeat like it is done in #2435?

[beyond1920]

Hi, till. Thanks for your review.
First sorry to bind to the wrong jira link, already correct it.
About heartbeatManager, i thought we could separate HeartbeatManager into two part, each part focus only on one function.

1. Death watch to monitor liveness, it could register watch, unregister watch, and notify the dead targets. just like deathwatch in akka. Although now I implement it on Akka's dead watch functionality, but interface is defined. It's fine to give another implementation of it in future.
2. Payload deliver periodically. But because data sync behavior is different in different component, e.g, TM,JM,RM, so I'm not sure it is good to extract the logic into a common logic.