{"action":"created","comment":{"url":"https://api.github.com/repos/apache/flink/pulls/comments/35216181","id":35216181,"diff_hunk":"@@ -19,34 +19,30 @@\n package org.apache.flink.runtime\n \n import _root_.akka.actor.Actor\n-\n-/**\n- * Mixin to add debug message logging\n- */\n-trait ActorLogMessages {\n-  that: Actor with ActorSynchronousLogging =>\n-\n-  override def receive: Receive = new Actor.Receive {\n-    private val _receiveWithLogMessages = receiveWithLogMessages\n-\n-    override def isDefinedAt(x: Any): Boolean = _receiveWithLogMessages.isDefinedAt(x)\n-\n-    override def apply(x: Any): Unit = {\n-      if (!log.isDebugEnabled) {\n-        _receiveWithLogMessages(x)\n-      }\n-      else {\n-        log.debug(s\"Received message $x at ${that.self.path} from ${that.sender()}.\")\n-\n-        val start = System.nanoTime()\n-\n-        _receiveWithLogMessages(x)\n-\n-        val duration = (System.nanoTime() - start) / 1000000\n-        log.debug(s\"Handled message $x in $duration ms from ${that.sender()}.\")\n-      }\n-    }\n+import grizzled.slf4j.Logger\n+\n+/** Base trait for Flink's actors.\n+  *\n+  * The message handling logic is defined in the handleMessage method. This allows to mixin\n+  * stackable traits which change the message receiving behaviour.\n+  */\n+trait FlinkActor extends Actor {\n+  lazy val log = Logger(getClass)","path":"flink-runtime/src/main/scala/org/apache/flink/runtime/FlinkActor.scala","position":39,"original_position":39,"commit_id":"5b66bffdcf625f596331a6fa23ac6cac9da36722","original_commit_id":"5b66bffdcf625f596331a6fa23ac6cac9da36722","user":{"login":"StephanEwen","id":1727146,"avatar_url":"https://avatars.githubusercontent.com/u/1727146?v=3","gravatar_id":"","url":"https://api.github.com/users/StephanEwen","html_url":"https://github.com/StephanEwen","followers_url":"https://api.github.com/users/StephanEwen/followers","following_url":"https://api.github.com/users/StephanEwen/following{/other_user}","gists_url":"https://api.github.com/users/StephanEwen/gists{/gist_id}","starred_url":"https://api.github.com/users/StephanEwen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/StephanEwen/subscriptions","organizations_url":"https://api.github.com/users/StephanEwen/orgs","repos_url":"https://api.github.com/users/StephanEwen/repos","events_url":"https://api.github.com/users/StephanEwen/events{/privacy}","received_events_url":"https://api.github.com/users/StephanEwen/received_events","type":"User","site_admin":false},"body":"I think abstract variables just give us function calls instead. Can it be a regular variable?\r\nWhen is the `getClass()` method here actually called?","created_at":"2015-07-22T14:10:19Z","updated_at":"2015-07-22T14:10:19Z","html_url":"https://github.com/apache/flink/pull/917#discussion_r35216181","pull_request_url":"https://api.github.com/repos/apache/flink/pulls/917","_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/comments/35216181"},"html":{"href":"https://github.com/apache/flink/pull/917#discussion_r35216181"},"pull_request":{"href":"https://api.github.com/repos/apache/flink/pulls/917"}}},"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/917","id":40106432,"html_url":"https://github.com/apache/flink/pull/917","diff_url":"https://github.com/apache/flink/pull/917.diff","patch_url":"https://github.com/apache/flink/pull/917.patch","issue_url":"https://api.github.com/repos/apache/flink/issues/917","number":917,"state":"open","locked":false,"title":"[FLINK-2332] [runtime] Adds leader session IDs and registration session IDs","user":{"login":"tillrohrmann","id":5756858,"avatar_url":"https://avatars.githubusercontent.com/u/5756858?v=3","gravatar_id":"","url":"https://api.github.com/users/tillrohrmann","html_url":"https://github.com/tillrohrmann","followers_url":"https://api.github.com/users/tillrohrmann/followers","following_url":"https://api.github.com/users/tillrohrmann/following{/other_user}","gists_url":"https://api.github.com/users/tillrohrmann/gists{/gist_id}","starred_url":"https://api.github.com/users/tillrohrmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tillrohrmann/subscriptions","organizations_url":"https://api.github.com/users/tillrohrmann/orgs","repos_url":"https://api.github.com/users/tillrohrmann/repos","events_url":"https://api.github.com/users/tillrohrmann/events{/privacy}","received_events_url":"https://api.github.com/users/tillrohrmann/received_events","type":"User","site_admin":false},"body":"## Registration Session IDs\r\n\r\nIntroduces 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. \r\n\r\nIn 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.\r\n\r\n## Leader Session IDs\r\n\r\nIn 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. \r\n\r\nMessages 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.\r\n\r\nIn 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.\r\n\r\nThe `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.\r\n\r\nIn 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.\r\n\r\nIn 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.\r\n\r\n## ActorGateway refactoring\r\n\r\nIn 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. \r\n\r\n`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. \r\n\r\n## Style formatting\r\n\r\nThe PR also contains some Scala style harmonisation of old Scala code.\r\n\r\n## TL;DR\r\n\r\nIn 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.","created_at":"2015-07-16T08:42:10Z","updated_at":"2015-07-22T14:10:19Z","closed_at":null,"merged_at":null,"merge_commit_sha":"6448012b0a10b0c3d890e6018cdf0d5cbed7a9b3","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/flink/pulls/917/commits","review_comments_url":"https://api.github.com/repos/apache/flink/pulls/917/comments","review_comment_url":"https://api.github.com/repos/apache/flink/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/flink/issues/917/comments","statuses_url":"https://api.github.com/repos/apache/flink/statuses/5b66bffdcf625f596331a6fa23ac6cac9da36722","head":{"label":"tillrohrmann:introduceLeaderSessionID","ref":"introduceLeaderSessionID","sha":"5b66bffdcf625f596331a6fa23ac6cac9da36722","user":{"login":"tillrohrmann","id":5756858,"avatar_url":"https://avatars.githubusercontent.com/u/5756858?v=3","gravatar_id":"","url":"https://api.github.com/users/tillrohrmann","html_url":"https://github.com/tillrohrmann","followers_url":"https://api.github.com/users/tillrohrmann/followers","following_url":"https://api.github.com/users/tillrohrmann/following{/other_user}","gists_url":"https://api.github.com/users/tillrohrmann/gists{/gist_id}","starred_url":"https://api.github.com/users/tillrohrmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tillrohrmann/subscriptions","organizations_url":"https://api.github.com/users/tillrohrmann/orgs","repos_url":"https://api.github.com/users/tillrohrmann/repos","events_url":"https://api.github.com/users/tillrohrmann/events{/privacy}","received_events_url":"https://api.github.com/users/tillrohrmann/received_events","type":"User","site_admin":false},"repo":{"id":20681108,"name":"flink","full_name":"tillrohrmann/flink","owner":{"login":"tillrohrmann","id":5756858,"avatar_url":"https://avatars.githubusercontent.com/u/5756858?v=3","gravatar_id":"","url":"https://api.github.com/users/tillrohrmann","html_url":"https://github.com/tillrohrmann","followers_url":"https://api.github.com/users/tillrohrmann/followers","following_url":"https://api.github.com/users/tillrohrmann/following{/other_user}","gists_url":"https://api.github.com/users/tillrohrmann/gists{/gist_id}","starred_url":"https://api.github.com/users/tillrohrmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tillrohrmann/subscriptions","organizations_url":"https://api.github.com/users/tillrohrmann/orgs","repos_url":"https://api.github.com/users/tillrohrmann/repos","events_url":"https://api.github.com/users/tillrohrmann/events{/privacy}","received_events_url":"https://api.github.com/users/tillrohrmann/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/tillrohrmann/flink","description":"Mirror of Apache Flink","fork":true,"url":"https://api.github.com/repos/tillrohrmann/flink","forks_url":"https://api.github.com/repos/tillrohrmann/flink/forks","keys_url":"https://api.github.com/repos/tillrohrmann/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/tillrohrmann/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/tillrohrmann/flink/teams","hooks_url":"https://api.github.com/repos/tillrohrmann/flink/hooks","issue_events_url":"https://api.github.com/repos/tillrohrmann/flink/issues/events{/number}","events_url":"https://api.github.com/repos/tillrohrmann/flink/events","assignees_url":"https://api.github.com/repos/tillrohrmann/flink/assignees{/user}","branches_url":"https://api.github.com/repos/tillrohrmann/flink/branches{/branch}","tags_url":"https://api.github.com/repos/tillrohrmann/flink/tags","blobs_url":"https://api.github.com/repos/tillrohrmann/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/tillrohrmann/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/tillrohrmann/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/tillrohrmann/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/tillrohrmann/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/tillrohrmann/flink/languages","stargazers_url":"https://api.github.com/repos/tillrohrmann/flink/stargazers","contributors_url":"https://api.github.com/repos/tillrohrmann/flink/contributors","subscribers_url":"https://api.github.com/repos/tillrohrmann/flink/subscribers","subscription_url":"https://api.github.com/repos/tillrohrmann/flink/subscription","commits_url":"https://api.github.com/repos/tillrohrmann/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/tillrohrmann/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/tillrohrmann/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/tillrohrmann/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/tillrohrmann/flink/contents/{+path}","compare_url":"https://api.github.com/repos/tillrohrmann/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/tillrohrmann/flink/merges","archive_url":"https://api.github.com/repos/tillrohrmann/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/tillrohrmann/flink/downloads","issues_url":"https://api.github.com/repos/tillrohrmann/flink/issues{/number}","pulls_url":"https://api.github.com/repos/tillrohrmann/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/tillrohrmann/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/tillrohrmann/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/tillrohrmann/flink/labels{/name}","releases_url":"https://api.github.com/repos/tillrohrmann/flink/releases{/id}","created_at":"2014-06-10T10:19:35Z","updated_at":"2015-04-21T08:56:37Z","pushed_at":"2015-07-16T08:40:14Z","git_url":"git://github.com/tillrohrmann/flink.git","ssh_url":"git@github.com:tillrohrmann/flink.git","clone_url":"https://github.com/tillrohrmann/flink.git","svn_url":"https://github.com/tillrohrmann/flink","homepage":null,"size":53722,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"8261ed5438278331c9cd760463273ad94d4bd410","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":20587599,"name":"flink","full_name":"apache/flink","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/flink","description":"Mirror of Apache Flink","fork":false,"url":"https://api.github.com/repos/apache/flink","forks_url":"https://api.github.com/repos/apache/flink/forks","keys_url":"https://api.github.com/repos/apache/flink/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/flink/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/flink/teams","hooks_url":"https://api.github.com/repos/apache/flink/hooks","issue_events_url":"https://api.github.com/repos/apache/flink/issues/events{/number}","events_url":"https://api.github.com/repos/apache/flink/events","assignees_url":"https://api.github.com/repos/apache/flink/assignees{/user}","branches_url":"https://api.github.com/repos/apache/flink/branches{/branch}","tags_url":"https://api.github.com/repos/apache/flink/tags","blobs_url":"https://api.github.com/repos/apache/flink/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/flink/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/flink/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/flink/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/flink/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/flink/languages","stargazers_url":"https://api.github.com/repos/apache/flink/stargazers","contributors_url":"https://api.github.com/repos/apache/flink/contributors","subscribers_url":"https://api.github.com/repos/apache/flink/subscribers","subscription_url":"https://api.github.com/repos/apache/flink/subscription","commits_url":"https://api.github.com/repos/apache/flink/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/flink/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/flink/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/flink/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/flink/contents/{+path}","compare_url":"https://api.github.com/repos/apache/flink/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/flink/merges","archive_url":"https://api.github.com/repos/apache/flink/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/flink/downloads","issues_url":"https://api.github.com/repos/apache/flink/issues{/number}","pulls_url":"https://api.github.com/repos/apache/flink/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/flink/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/flink/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/flink/labels{/name}","releases_url":"https://api.github.com/repos/apache/flink/releases{/id}","created_at":"2014-06-07T07:00:10Z","updated_at":"2015-07-22T09:15:18Z","pushed_at":"2015-07-22T12:54:45Z","git_url":"git://github.com/apache/flink.git","ssh_url":"git@github.com:apache/flink.git","clone_url":"https://github.com/apache/flink.git","svn_url":"https://github.com/apache/flink","homepage":null,"size":2564597,"stargazers_count":513,"watchers_count":513,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":340,"mirror_url":"git://git.apache.org/incubator-flink.git","open_issues_count":61,"forks":340,"open_issues":61,"watchers":513,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/flink/pulls/917"},"html":{"href":"https://github.com/apache/flink/pull/917"},"issue":{"href":"https://api.github.com/repos/apache/flink/issues/917"},"comments":{"href":"https://api.github.com/repos/apache/flink/issues/917/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/flink/pulls/917/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/flink/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/flink/pulls/917/commits"},"statuses":{"href":"https://api.github.com/repos/apache/flink/statuses/5b66bffdcf625f596331a6fa23ac6cac9da36722"}}}}