{"action":"created","comment":{"url":"https://api.github.com/repos/twitter/summingbird/pulls/comments/75226834","id":75226834,"diff_hunk":"@@ -204,49 +202,58 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird\n     }\n     val metrics = getOrElse(stormDag, node, DEFAULT_SPOUT_STORM_METRICS)\n \n-    val registerAllMetrics = Externalizer({ context: TopologyContext =>\n+    val summerOpt: Option[SummerNode[Storm]] = stormDag.dependantsOf(node.asInstanceOf[StormNode]).collect { case s: SummerNode[Storm] => s }.headOption\n+    val stormSpout = summerOpt match {\n+      case Some(s) => {\n+        val builder = BuildSummer(this, stormDag, node, jobID)\n+        val registerAllMetricsAndCounters = createRegistrationFunction(metrics, getCountersForJob(jobID), jobID)\n+        createSpoutToFeedSummer[Any, Any](stormDag, s, builder, tormentaSpout, registerAllMetricsAndCounters)\n+      }\n+      case None => {\n+        val registerAllMetricsAndCounters = createRegistrationFunction(metrics, getCountersForJob(jobID), jobID)\n+        getStormSpout(tormentaSpout, registerAllMetricsAndCounters)\n+      }\n+    }\n+    topologyBuilder.setSpout(nodeName, stormSpout, sourceParallelism)\n+  }\n+\n+  // Tormenta bug when openHook might get lost if it doesn't happen right before the getSpout.\n+  private def getStormSpout[T](spout: Spout[T], register: Externalizer[(TopologyContext) => Unit]) = {\n+    val hookedSpout = spout.openHook(register.get)\n+    hookedSpout.getSpout\n+  }\n+\n+  private def getCountersForJob(jobId: JobId) = JobCounters.getCountersForJob(jobId).getOrElse(Nil)\n+\n+  /*\n+  This function takes the metrics and counters to register to the TopologyContext of a JobID and outputs a function which can be called on openHook.\n+   */\n+  private def createRegistrationFunction(metrics: SpoutStormMetrics, counters: TraversableOnce[(Group, Name)], jobID: JobId): Externalizer[(TopologyContext) => Unit] = {\n+    Externalizer({ context: TopologyContext =>\n       // Register metrics passed in SpoutStormMetrics option.\n       metrics.metrics().foreach {\n         x: StormMetric[IMetric] =>\n           context.registerMetric(x.name, x.metric, x.interval.inSeconds)\n       }\n       // Register summingbird counter metrics.\n-      StormStatProvider.registerMetrics(jobID, context, countersForSpout)\n+      StormStatProvider.registerMetrics(jobID, context, counters.toSeq)\n       SummingbirdRuntimeStats.addPlatformStatProvider(StormStatProvider)\n     })\n-    val hookedTormentaSpout = tormentaSpout.openHook(registerAllMetrics.get)\n-\n-    val summerOpt: Option[SummerNode[Storm]] = stormDag.dependantsOf(node.asInstanceOf[StormNode]).collect { case s: SummerNode[Storm] => s }.headOption\n-\n-    val stormSpout = summerOpt match {\n-      /*\n-         *  As the spout is being followed by a summer, we do not have map-side aggregation handled in the spout.\n-         *  This means Summer is the place where the aggregation should happen.\n-         */\n-      case Some(s) => createSpoutToFeedSummer(stormDag, s, hookedTormentaSpout)\n-      case None => hookedTormentaSpout.getSpout\n-    }\n-    topologyBuilder.setSpout(nodeName, stormSpout, sourceParallelism)\n   }\n \n-  private def createSpoutToFeedSummer(stormDag: Dag[Storm], node: StormNode, spout: Spout[(Timestamp, Any)]) = {\n+  private def createSpoutToFeedSummer[K, V](stormDag: Dag[Storm], node: StormNode, builder: SummerBuilder, spout: Spout[(Timestamp, Any)], fn: Externalizer[(TopologyContext) => Unit]): IRichSpout = {","path":"summingbird-storm/src/main/scala/com/twitter/summingbird/storm/StormPlatform.scala","position":93,"original_position":93,"commit_id":"66eeb2bfe57572f977a6a303dc23b2232915385d","original_commit_id":"66eeb2bfe57572f977a6a303dc23b2232915385d","user":{"login":"pankajroark","id":664454,"avatar_url":"https://avatars.githubusercontent.com/u/664454?v=3","gravatar_id":"","url":"https://api.github.com/users/pankajroark","html_url":"https://github.com/pankajroark","followers_url":"https://api.github.com/users/pankajroark/followers","following_url":"https://api.github.com/users/pankajroark/following{/other_user}","gists_url":"https://api.github.com/users/pankajroark/gists{/gist_id}","starred_url":"https://api.github.com/users/pankajroark/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pankajroark/subscriptions","organizations_url":"https://api.github.com/users/pankajroark/orgs","repos_url":"https://api.github.com/users/pankajroark/repos","events_url":"https://api.github.com/users/pankajroark/events{/privacy}","received_events_url":"https://api.github.com/users/pankajroark/received_events","type":"User","site_admin":false},"body":"Can we move this out to a separate file, like FlatMapBoltProvider?","created_at":"2016-08-17T23:34:11Z","updated_at":"2016-08-17T23:34:11Z","html_url":"https://github.com/twitter/summingbird/pull/682#discussion_r75226834","pull_request_url":"https://api.github.com/repos/twitter/summingbird/pulls/682","_links":{"self":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/comments/75226834"},"html":{"href":"https://github.com/twitter/summingbird/pull/682#discussion_r75226834"},"pull_request":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/682"}}},"pull_request":{"url":"https://api.github.com/repos/twitter/summingbird/pulls/682","id":81182439,"html_url":"https://github.com/twitter/summingbird/pull/682","diff_url":"https://github.com/twitter/summingbird/pull/682.diff","patch_url":"https://github.com/twitter/summingbird/pull/682.patch","issue_url":"https://api.github.com/repos/twitter/summingbird/issues/682","number":682,"state":"open","locked":false,"title":"Adding crush down capability into the SourceNode","user":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"body":"Closes #679\r\nThis PR mainly adds the summing capability to the spout. When FMMergeableWithSource is opted in, the flatMap operation in the spout creates a fanOut. The fanOut is being crushed down by using the capability of AsyncSummers inside the spout and emitting only crushed tuples. The spout-side aggregation metrics are added to the spout for performance analysis. The messageIds are also aggregated along with the tuples and ack/fail back is sent to all the aggregated messageIds. Tuples from different streams are aggregated individually.  ","created_at":"2016-08-12T21:00:11Z","updated_at":"2016-08-17T23:34:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"d5f22d8f6e841347bdb4d6762889b2f9bf0affb9","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/twitter/summingbird/pulls/682/commits","review_comments_url":"https://api.github.com/repos/twitter/summingbird/pulls/682/comments","review_comment_url":"https://api.github.com/repos/twitter/summingbird/pulls/comments{/number}","comments_url":"https://api.github.com/repos/twitter/summingbird/issues/682/comments","statuses_url":"https://api.github.com/repos/twitter/summingbird/statuses/66eeb2bfe57572f977a6a303dc23b2232915385d","head":{"label":"NPraneeth:pnaramsetti/add-aggregation-spout","ref":"pnaramsetti/add-aggregation-spout","sha":"66eeb2bfe57572f977a6a303dc23b2232915385d","user":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"repo":{"id":65408713,"name":"summingbird","full_name":"NPraneeth/summingbird","owner":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/NPraneeth/summingbird","description":"Streaming MapReduce with Scalding and Storm","fork":true,"url":"https://api.github.com/repos/NPraneeth/summingbird","forks_url":"https://api.github.com/repos/NPraneeth/summingbird/forks","keys_url":"https://api.github.com/repos/NPraneeth/summingbird/keys{/key_id}","collaborators_url":"https://api.github.com/repos/NPraneeth/summingbird/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/NPraneeth/summingbird/teams","hooks_url":"https://api.github.com/repos/NPraneeth/summingbird/hooks","issue_events_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/events{/number}","events_url":"https://api.github.com/repos/NPraneeth/summingbird/events","assignees_url":"https://api.github.com/repos/NPraneeth/summingbird/assignees{/user}","branches_url":"https://api.github.com/repos/NPraneeth/summingbird/branches{/branch}","tags_url":"https://api.github.com/repos/NPraneeth/summingbird/tags","blobs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/NPraneeth/summingbird/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/refs{/sha}","trees_url":"https://api.github.com/repos/NPraneeth/summingbird/git/trees{/sha}","statuses_url":"https://api.github.com/repos/NPraneeth/summingbird/statuses/{sha}","languages_url":"https://api.github.com/repos/NPraneeth/summingbird/languages","stargazers_url":"https://api.github.com/repos/NPraneeth/summingbird/stargazers","contributors_url":"https://api.github.com/repos/NPraneeth/summingbird/contributors","subscribers_url":"https://api.github.com/repos/NPraneeth/summingbird/subscribers","subscription_url":"https://api.github.com/repos/NPraneeth/summingbird/subscription","commits_url":"https://api.github.com/repos/NPraneeth/summingbird/commits{/sha}","git_commits_url":"https://api.github.com/repos/NPraneeth/summingbird/git/commits{/sha}","comments_url":"https://api.github.com/repos/NPraneeth/summingbird/comments{/number}","issue_comment_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/comments{/number}","contents_url":"https://api.github.com/repos/NPraneeth/summingbird/contents/{+path}","compare_url":"https://api.github.com/repos/NPraneeth/summingbird/compare/{base}...{head}","merges_url":"https://api.github.com/repos/NPraneeth/summingbird/merges","archive_url":"https://api.github.com/repos/NPraneeth/summingbird/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/NPraneeth/summingbird/downloads","issues_url":"https://api.github.com/repos/NPraneeth/summingbird/issues{/number}","pulls_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls{/number}","milestones_url":"https://api.github.com/repos/NPraneeth/summingbird/milestones{/number}","notifications_url":"https://api.github.com/repos/NPraneeth/summingbird/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/NPraneeth/summingbird/labels{/name}","releases_url":"https://api.github.com/repos/NPraneeth/summingbird/releases{/id}","deployments_url":"https://api.github.com/repos/NPraneeth/summingbird/deployments","created_at":"2016-08-10T18:58:51Z","updated_at":"2016-08-10T18:58:53Z","pushed_at":"2016-08-16T21:21:11Z","git_url":"git://github.com/NPraneeth/summingbird.git","ssh_url":"git@github.com:NPraneeth/summingbird.git","clone_url":"https://github.com/NPraneeth/summingbird.git","svn_url":"https://github.com/NPraneeth/summingbird","homepage":"https://twitter.com/summingbird","size":6230,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"develop"}},"base":{"label":"twitter:develop","ref":"develop","sha":"22db7b0a65f472c42c8f283a7b7773eb68f3d973","user":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?v=3","gravatar_id":"","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"repo":{"id":5957708,"name":"summingbird","full_name":"twitter/summingbird","owner":{"login":"twitter","id":50278,"avatar_url":"https://avatars.githubusercontent.com/u/50278?v=3","gravatar_id":"","url":"https://api.github.com/users/twitter","html_url":"https://github.com/twitter","followers_url":"https://api.github.com/users/twitter/followers","following_url":"https://api.github.com/users/twitter/following{/other_user}","gists_url":"https://api.github.com/users/twitter/gists{/gist_id}","starred_url":"https://api.github.com/users/twitter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/twitter/subscriptions","organizations_url":"https://api.github.com/users/twitter/orgs","repos_url":"https://api.github.com/users/twitter/repos","events_url":"https://api.github.com/users/twitter/events{/privacy}","received_events_url":"https://api.github.com/users/twitter/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/twitter/summingbird","description":"Streaming MapReduce with Scalding and Storm","fork":false,"url":"https://api.github.com/repos/twitter/summingbird","forks_url":"https://api.github.com/repos/twitter/summingbird/forks","keys_url":"https://api.github.com/repos/twitter/summingbird/keys{/key_id}","collaborators_url":"https://api.github.com/repos/twitter/summingbird/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/twitter/summingbird/teams","hooks_url":"https://api.github.com/repos/twitter/summingbird/hooks","issue_events_url":"https://api.github.com/repos/twitter/summingbird/issues/events{/number}","events_url":"https://api.github.com/repos/twitter/summingbird/events","assignees_url":"https://api.github.com/repos/twitter/summingbird/assignees{/user}","branches_url":"https://api.github.com/repos/twitter/summingbird/branches{/branch}","tags_url":"https://api.github.com/repos/twitter/summingbird/tags","blobs_url":"https://api.github.com/repos/twitter/summingbird/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/twitter/summingbird/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/twitter/summingbird/git/refs{/sha}","trees_url":"https://api.github.com/repos/twitter/summingbird/git/trees{/sha}","statuses_url":"https://api.github.com/repos/twitter/summingbird/statuses/{sha}","languages_url":"https://api.github.com/repos/twitter/summingbird/languages","stargazers_url":"https://api.github.com/repos/twitter/summingbird/stargazers","contributors_url":"https://api.github.com/repos/twitter/summingbird/contributors","subscribers_url":"https://api.github.com/repos/twitter/summingbird/subscribers","subscription_url":"https://api.github.com/repos/twitter/summingbird/subscription","commits_url":"https://api.github.com/repos/twitter/summingbird/commits{/sha}","git_commits_url":"https://api.github.com/repos/twitter/summingbird/git/commits{/sha}","comments_url":"https://api.github.com/repos/twitter/summingbird/comments{/number}","issue_comment_url":"https://api.github.com/repos/twitter/summingbird/issues/comments{/number}","contents_url":"https://api.github.com/repos/twitter/summingbird/contents/{+path}","compare_url":"https://api.github.com/repos/twitter/summingbird/compare/{base}...{head}","merges_url":"https://api.github.com/repos/twitter/summingbird/merges","archive_url":"https://api.github.com/repos/twitter/summingbird/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/twitter/summingbird/downloads","issues_url":"https://api.github.com/repos/twitter/summingbird/issues{/number}","pulls_url":"https://api.github.com/repos/twitter/summingbird/pulls{/number}","milestones_url":"https://api.github.com/repos/twitter/summingbird/milestones{/number}","notifications_url":"https://api.github.com/repos/twitter/summingbird/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/twitter/summingbird/labels{/name}","releases_url":"https://api.github.com/repos/twitter/summingbird/releases{/id}","deployments_url":"https://api.github.com/repos/twitter/summingbird/deployments","created_at":"2012-09-25T22:38:35Z","updated_at":"2016-08-17T17:45:17Z","pushed_at":"2016-08-17T23:14:11Z","git_url":"git://github.com/twitter/summingbird.git","ssh_url":"git@github.com:twitter/summingbird.git","clone_url":"https://github.com/twitter/summingbird.git","svn_url":"https://github.com/twitter/summingbird","homepage":"https://twitter.com/summingbird","size":8561,"stargazers_count":1773,"watchers_count":1773,"language":"Scala","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":230,"mirror_url":null,"open_issues_count":136,"forks":230,"open_issues":136,"watchers":1773,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/682"},"html":{"href":"https://github.com/twitter/summingbird/pull/682"},"issue":{"href":"https://api.github.com/repos/twitter/summingbird/issues/682"},"comments":{"href":"https://api.github.com/repos/twitter/summingbird/issues/682/comments"},"review_comments":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/682/comments"},"review_comment":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/twitter/summingbird/pulls/682/commits"},"statuses":{"href":"https://api.github.com/repos/twitter/summingbird/statuses/66eeb2bfe57572f977a6a303dc23b2232915385d"}}}}
{"actor":{"display_login":"pankajroark"}}