[FLINK-4382] Buffer rpc calls until the RpcEndpoint has been started

[tillrohrmann]

This PR allows the AkkaRpcActor to stash messages until the corresponding RcpEndpoint
has been started. When receiving a Processing.START message, the AkkaRpcActor
unstashes all messages and starts processing rpcs. When receiving a Processing.STOP
message, it will stop processing messages and stash incoming messages again.

This PR is based on #2357.

R @StephanEwen

[StephanEwen]

Not completely sure about the message stashing. Why not drop the messages? That would be safer (less to clean up, no overflow).

Is this mainly to not loose the messages from the "self gateway"? Because for all remote messages, it should be not necessary, it should simply appear as if the actor/endpoint started a few milliseconds later.

It may in the common case not be much of an issue, as the period during which messages are stashed is small, but I think it is simply not necessary.

[tillrohrmann]

Dropping is also a solution. I think this will also happen implicitly if the mailbox grows too big.

Stashing has the advantage that we try (on a best effort basis) to answer early requests. That way the rpc calls won't have to time out on the sending side before being retried or alternative measures are taken. Of course this can happen if the start up takes too long on the server side. Thus, I would imagine that this approach is a little bit more responsive wrt early rpcs.

So I don't think that stashing actually hurts.

[tillrohrmann]

The stashing is a feature which comes for free with Akka. Of course the sending side should not rely on this behaviour when implementing its failure treatment. But in cases where we have a race condition between the start-up of the RpcEndpoint and a RpcGateway it should result into a slightly more forgiving receiver behaviour.

I will rebase the PR and if Travis gives green light I would like to merge this PR to the flip-6 feature branch if nobody objects.