GitHub user tillrohrmann opened a pull request:
https://github.com/apache/flink/pull/917
[FLINK-2332] [runtime] Adds leader session IDs and registration session IDs
## Registration Session IDs
Introduces registration session IDs for all registration messages. Upon
receiving a registration message, this ID is checked and if not correct, the
message is discarded. That way, it is possible to distinguish old registration
messages which are delayed from valid ones.
In the current implementation, a static registration session ID is assigned
when the `TaskManager` actor is created. However, with support for high
availability, where the leader can change while trying to register at old
leader, it becomes important to distinguish old from new registration messages.
## Leader Session IDs
In order to support high availability, we not only have to distinguish the
old from the new registration messages, but also control messages which are
sent to and from the `JobManager` and the `TaskManager`. In order to filter out
possibly old messages, this PR introduces a leader session ID which denotes the
currently valid messages. However, unlike the registration session ID, leader
session IDs are assigned to messages transparently.
Messages which extend the `RequiresLeaderSessionID` interface will be
wrapped in a `LeaderSessionMessage` which also contains the currently known
leader session ID. At the receiving end, the `LeaderSessionMessages` are
unpacked and the received leader session ID is compared to the currently stored
leader session ID. If both IDs are the same, the wrapped message is processed.
If not, then wrapped message will be discarded.
In order to support this behaviour, the PR introduces a new `FlinkActor`
for Scala actors and a `FlinkUntypedActor` for Java actors. Both actors provide
a `decorateMessage` method which allows sub types of the
`FlinkActor`/`FlinkUntypedActor` to decorate outgoing messages. Therefore, all
implementing classes are supposed to call `decorateMessage` before sending a
message to another actor.
The `FlinkUntypedActor` already comes with support for message logging and
leader session message filtering. Furthermore, its `decorateMessage` method
implementation checks for each message if it's an subtype of
`RequiresLeaderSessionID` or not and if it is the case, then wraps this message
in a `LeaderSessionMessage`. The receive method of this class, will then take
care to unwrap the messages accordingly.
In order to have the same behavior with Scala actors one has to extend the
`FlinkActor` and mixin the `LeaderSessionMessages` and `LoggingMessages`
mixins. They effectively do the same as the `FlinkUntypedActor`, but offer a
better extensibility of the Scala actors in the future.
In case that a `RequiresLeaderSessionID` message is received without being
wrapped in a `LeaderSessionMessage` by a FlinkActor, an exception is thrown,
which effectively terminates the execution of the actor. The reason for this is
that a unwrapped message might leave the system in an inconsistent state if
it's a message from an old leader. Furthermore, since not every
`RequiresLeaderSessionID` message requires a response, it is not possible to
notify the sender of the wrong message about the missing leader session ID.
## ActorGateway refactoring
In order to guarantee the similar wrapping behaviour when one sends
messages outside of an actor, the former `InstanceGateway` has been refactored
to `ActorGateway` and all `ActorRef` interactions have been replaced by
`ActorGateway` instances. Only the web server still uses `ActorRefs`, because
it is about to be refactored anyway (see #677). However, the PR #677 should be
updated accordingly.
`AkkaActorGateway` implements the `ActorGateway` and makes sure that all
`RequiresLeaderSessionID` messages are wrapped correctly in a
`LeaderSessionMessage`. In order to make this happen, the `AkkaActorGateway` is
given the current leader session ID upon creation. For any actor interaction
from outside of an actor, this class should be used. Using this abstraction
will allow us to easily extend the decoration behaviour of messages in the
future, too.
## Style formatting
The PR also contains some Scala style harmonisation of old Scala code.
## TL;DR
In order to support leader session IDs all Flink actors should extend
`FlinkUntypedActor` or `FlinkActor` with `LeaderSessionMessages` and
`LogMessages` mixins. Whenever a message is sent from within an actor, the
message should be decorated by calling `decorateMessage`. When messages are
sent from outside of an actor, the `AkkaActorGateway` should be used instead of
directly using `ActorRef`. That way, we guarantee the proper wrapping of the
messages.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tillrohrmann/flink introduceLeaderSessionID
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/917.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 #917
----
commit 5b66bffdcf625f596331a6fa23ac6cac9da36722
Author: Till Rohrmann <[email protected]>
Date: 2015-07-08T13:37:50Z
[FLINK-2332] [runtime] Adds leader session IDs and registration session IDs
to JobManager and TaskManager messages.
Refactors Flink's actor traits to support stackable traits
Refactored Flink's actors to use a factory method to generate messages
Replaced ActorRef with InstanceGateway in Task, RuntimeEnvironment and
ExecutionGraph
Add comments
Add test cases for registration session ID and leader session ID
Adds IT case to check that a CancelMessage with the wrong leader session ID
is 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 [email protected] or file a JIRA ticket
with INFRA.
---
