{"action":"created","comment":{"url":"https://api.github.com/repos/radargun/radargun/pulls/comments/44323577","id":44323577,"diff_hunk":"@@ -0,0 +1,213 @@\n+package org.radargun.stages.mapreducestream;\n+\n+import org.radargun.DistStageAck;\n+import org.radargun.StageResult;\n+import org.radargun.config.Property;\n+import org.radargun.config.Stage;\n+import org.radargun.reporting.Report;\n+import org.radargun.stages.AbstractDistStage;\n+import org.radargun.stages.cache.RandomDataStage;\n+import org.radargun.state.SlaveState;\n+import org.radargun.stats.DataOperationStats;\n+import org.radargun.stats.DefaultStatistics;\n+import org.radargun.stats.Statistics;\n+import org.radargun.traits.CacheInformation;\n+import org.radargun.traits.Clustered;\n+import org.radargun.traits.InjectTrait;\n+import org.radargun.traits.Streamable;\n+import org.radargun.utils.Projections;\n+import org.radargun.utils.TimeService;\n+import org.radargun.utils.Utils;\n+\n+import java.util.Collections;\n+import java.util.HashMap;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.concurrent.TimeUnit;\n+\n+/**\n+ * @author Roman Macor &lt;rmacor@redhat.com&gt;\n+ */\n+@Stage(doc = \"Stage which executes a MapReduce Task against all keys in the cache.\")\n+public class MapReduceStreamStage extends AbstractDistStage {\n+\n+   @Property(optional = true, doc = \"Boolean value that determines if the \"\n+           + \"final results of the MapReduceTask are written to the log of the \"\n+           + \"first slave node. The default is false.\")\n+   private boolean printResult = false;\n+\n+   @Property(doc = \"The number of times to execute the Map/Reduce task. The default is 10.\")\n+   private int numExecutions = 10;\n+\n+   @Property(doc = \"Name of the cache where map-reduce task should be\" + \"executed. Default is the default cache.\")\n+   private String cacheName;\n+\n+   @Property(doc = \"The name of the key in the MasterState object that returns the total number of \"\n+           + \"bytes processed by the Map/Reduce task. The default is RandomDataStage.RANDOMDATA_TOTALBYTES_KEY.\")\n+   private String totalBytesKey = RandomDataStage.RANDOMDATA_TOTALBYTES_KEY;\n+\n+   @Property(optional = false, doc = \"Fully qualified class name of the StreamConsumer implementation.\")\n+   private String streamOperationClass;\n+\n+   @InjectTrait\n+   private Streamable streamable;\n+\n+   @InjectTrait(dependency = InjectTrait.Dependency.MANDATORY)\n+   private Clustered clustered;\n+\n+   @InjectTrait(dependency = InjectTrait.Dependency.MANDATORY)\n+   private CacheInformation cacheInformation;\n+\n+   private Streamable.StreamConsumer consumer;\n+\n+   private Map<Object, Object> payloadMap = null;\n+\n+   private void configureMapReduceTask() {\n+      try {\n+         consumer = Utils.instantiate(streamOperationClass);\n+\n+      } catch (Exception e) {\n+         throw new IllegalArgumentException(\"Could not instantiate MapReduce implementation class: \" + streamOperationClass, e);\n+      }\n+   }\n+\n+   @Override\n+   public StageResult processAckOnMaster(List<DistStageAck> acks) {\n+      StageResult result = super.processAckOnMaster(acks);\n+      if (result.isError())\n+         return result;\n+\n+      Report report = masterState.getReport();\n+      Report.Test test = report.createTest(\"Map_Reduce_Stream_Stage\", null, true);\n+      int testIteration = test.getIterations().size();\n+\n+      Map<Integer, Report.SlaveResult> numberOfResultKeysResult = new HashMap<>();\n+      Map<Integer, Report.SlaveResult> durationsResult = new HashMap<>();\n+\n+      for (MapReduceStreamAck ack : Projections.instancesOf(acks, MapReduceStreamAck.class)) {\n+         if (ack.stats != null) {\n+            DataOperationStats opStats = (DataOperationStats) ack.stats.getOperationsStats().get(Streamable.STREAMABLE.name);\n+\n+            opStats.setTotalBytes((Long) masterState.get(totalBytesKey));\n+            test.addStatistics(testIteration, ack.getSlaveIndex(), Collections.singletonList(ack.stats));\n+            durationsResult.put(ack.getSlaveIndex(), new Report.SlaveResult(opStats.getResponseTimes(), false));\n+            test.addResult(testIteration, new Report.TestResult(\"Map/Reduce durations\", durationsResult, \"\", false));\n+         }\n+         if (ack.numberOfResultKeys != null) {\n+            numberOfResultKeysResult.put(ack.getSlaveIndex(), new Report.SlaveResult(ack.numberOfResultKeys, false));\n+            test.addResult(testIteration, new Report.TestResult(\"Key count in Map/Reduce result map\",\n+                    numberOfResultKeysResult, \"\", false));\n+         }\n+      }\n+\n+      return StageResult.SUCCESS;\n+   }\n+\n+   @Override\n+   public DistStageAck executeOnSlave() {\n+      if (!isServiceRunning()) {\n+         return errorResponse(\"Service is not runnning\", null);\n+      }\n+\n+      if (slaveState.getSlaveIndex() == 0) {\n+         DistStageAck result = null;\n+         Map<Object, Object> prevPayloadMap = null;\n+\n+         Statistics stats = new DefaultStatistics(new DataOperationStats());\n+         stats.begin();\n+\n+         for (int i = 0; i < numExecutions; i++) {\n+            prevPayloadMap = payloadMap;\n+\n+            result = executeMapReduceStreamTask(stats);\n+\n+            if (prevPayloadMap != null) {\n+               if (prevPayloadMap.keySet().equals(payloadMap.keySet())\n+                       && prevPayloadMap.entrySet().equals(payloadMap.entrySet())) {\n+                  log.info(i + \": Got the same results for two Map/Reduce runs\");\n+               } else {\n+                  log.error(i + \": Did not get the same results for two Map/Reduce runs\");\n+                  break;\n+               }\n+            }\n+\n+            if (result.isError() && exitOnFailure) {\n+               break;\n+            }\n+         }\n+         stats.end();\n+\n+         log.info(\"Executed Map reduce stream stage on slave\");\n+         return result;\n+\n+      } else {\n+         return new MapReduceStreamAck(slaveState);\n+      }\n+   }\n+\n+   private DistStageAck executeMapReduceStreamTask(Statistics stats) {\n+      long durationNanos;\n+      long start;\n+\n+      configureMapReduceTask();\n+\n+      MapReduceStreamAck ack = new MapReduceStreamAck(slaveState);\n+      try {\n+         start = TimeService.nanoTime();\n+\n+         consumer.accept(streamable.stream(cacheName));\n+\n+         Object result = consumer.getResult();\n+         if (result instanceof Map) {","path":"core/src/main/java/org/radargun/stages/mapreducestream/MapReduceStreamStage.java","position":161,"original_position":161,"commit_id":"c6ff0b25a74a8e31d12c1189c4245b56c9fbe51b","original_commit_id":"c6ff0b25a74a8e31d12c1189c4245b56c9fbe51b","user":{"login":"rvansa","id":2167869,"avatar_url":"https://avatars.githubusercontent.com/u/2167869?v=3","gravatar_id":"","url":"https://api.github.com/users/rvansa","html_url":"https://github.com/rvansa","followers_url":"https://api.github.com/users/rvansa/followers","following_url":"https://api.github.com/users/rvansa/following{/other_user}","gists_url":"https://api.github.com/users/rvansa/gists{/gist_id}","starred_url":"https://api.github.com/users/rvansa/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rvansa/subscriptions","organizations_url":"https://api.github.com/users/rvansa/orgs","repos_url":"https://api.github.com/users/rvansa/repos","events_url":"https://api.github.com/users/rvansa/events{/privacy}","received_events_url":"https://api.github.com/users/rvansa/received_events","type":"User","site_admin":false},"body":"Not really elegant. If you want to result be just reported, it should be the consumer (or rather the Function) that processes it to something presentable.","created_at":"2015-11-09T20:04:50Z","updated_at":"2015-11-09T20:04:50Z","html_url":"https://github.com/radargun/radargun/pull/282#discussion_r44323577","pull_request_url":"https://api.github.com/repos/radargun/radargun/pulls/282","_links":{"self":{"href":"https://api.github.com/repos/radargun/radargun/pulls/comments/44323577"},"html":{"href":"https://github.com/radargun/radargun/pull/282#discussion_r44323577"},"pull_request":{"href":"https://api.github.com/repos/radargun/radargun/pulls/282"}}},"pull_request":{"url":"https://api.github.com/repos/radargun/radargun/pulls/282","id":47285532,"html_url":"https://github.com/radargun/radargun/pull/282","diff_url":"https://github.com/radargun/radargun/pull/282.diff","patch_url":"https://github.com/radargun/radargun/pull/282.patch","issue_url":"https://api.github.com/repos/radargun/radargun/issues/282","number":282,"state":"open","locked":false,"title":"Adding new Map/Reduce stage which uses DistributedCacheStream API","user":{"login":"rmacor","id":8191057,"avatar_url":"https://avatars.githubusercontent.com/u/8191057?v=3","gravatar_id":"","url":"https://api.github.com/users/rmacor","html_url":"https://github.com/rmacor","followers_url":"https://api.github.com/users/rmacor/followers","following_url":"https://api.github.com/users/rmacor/following{/other_user}","gists_url":"https://api.github.com/users/rmacor/gists{/gist_id}","starred_url":"https://api.github.com/users/rmacor/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rmacor/subscriptions","organizations_url":"https://api.github.com/users/rmacor/orgs","repos_url":"https://api.github.com/users/rmacor/repos","events_url":"https://api.github.com/users/rmacor/events{/privacy}","received_events_url":"https://api.github.com/users/rmacor/received_events","type":"User","site_admin":false},"body":"Adds new Map/Reduce stage (currently only Word Count example) that uses DistributedCacheStream API.","created_at":"2015-10-09T15:28:31Z","updated_at":"2015-11-09T20:04:50Z","closed_at":null,"merged_at":null,"merge_commit_sha":"2249e87577f76da9733199912b0c57497b841589","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/radargun/radargun/pulls/282/commits","review_comments_url":"https://api.github.com/repos/radargun/radargun/pulls/282/comments","review_comment_url":"https://api.github.com/repos/radargun/radargun/pulls/comments{/number}","comments_url":"https://api.github.com/repos/radargun/radargun/issues/282/comments","statuses_url":"https://api.github.com/repos/radargun/radargun/statuses/c6ff0b25a74a8e31d12c1189c4245b56c9fbe51b","head":{"label":"rmacor:distributedCacheStream-performanceTests","ref":"distributedCacheStream-performanceTests","sha":"c6ff0b25a74a8e31d12c1189c4245b56c9fbe51b","user":{"login":"rmacor","id":8191057,"avatar_url":"https://avatars.githubusercontent.com/u/8191057?v=3","gravatar_id":"","url":"https://api.github.com/users/rmacor","html_url":"https://github.com/rmacor","followers_url":"https://api.github.com/users/rmacor/followers","following_url":"https://api.github.com/users/rmacor/following{/other_user}","gists_url":"https://api.github.com/users/rmacor/gists{/gist_id}","starred_url":"https://api.github.com/users/rmacor/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rmacor/subscriptions","organizations_url":"https://api.github.com/users/rmacor/orgs","repos_url":"https://api.github.com/users/rmacor/repos","events_url":"https://api.github.com/users/rmacor/events{/privacy}","received_events_url":"https://api.github.com/users/rmacor/received_events","type":"User","site_admin":false},"repo":{"id":39938988,"name":"radargun","full_name":"rmacor/radargun","owner":{"login":"rmacor","id":8191057,"avatar_url":"https://avatars.githubusercontent.com/u/8191057?v=3","gravatar_id":"","url":"https://api.github.com/users/rmacor","html_url":"https://github.com/rmacor","followers_url":"https://api.github.com/users/rmacor/followers","following_url":"https://api.github.com/users/rmacor/following{/other_user}","gists_url":"https://api.github.com/users/rmacor/gists{/gist_id}","starred_url":"https://api.github.com/users/rmacor/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rmacor/subscriptions","organizations_url":"https://api.github.com/users/rmacor/orgs","repos_url":"https://api.github.com/users/rmacor/repos","events_url":"https://api.github.com/users/rmacor/events{/privacy}","received_events_url":"https://api.github.com/users/rmacor/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/rmacor/radargun","description":"The RadarGun benchmarking framework for data grids and distributed caches","fork":true,"url":"https://api.github.com/repos/rmacor/radargun","forks_url":"https://api.github.com/repos/rmacor/radargun/forks","keys_url":"https://api.github.com/repos/rmacor/radargun/keys{/key_id}","collaborators_url":"https://api.github.com/repos/rmacor/radargun/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/rmacor/radargun/teams","hooks_url":"https://api.github.com/repos/rmacor/radargun/hooks","issue_events_url":"https://api.github.com/repos/rmacor/radargun/issues/events{/number}","events_url":"https://api.github.com/repos/rmacor/radargun/events","assignees_url":"https://api.github.com/repos/rmacor/radargun/assignees{/user}","branches_url":"https://api.github.com/repos/rmacor/radargun/branches{/branch}","tags_url":"https://api.github.com/repos/rmacor/radargun/tags","blobs_url":"https://api.github.com/repos/rmacor/radargun/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/rmacor/radargun/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/rmacor/radargun/git/refs{/sha}","trees_url":"https://api.github.com/repos/rmacor/radargun/git/trees{/sha}","statuses_url":"https://api.github.com/repos/rmacor/radargun/statuses/{sha}","languages_url":"https://api.github.com/repos/rmacor/radargun/languages","stargazers_url":"https://api.github.com/repos/rmacor/radargun/stargazers","contributors_url":"https://api.github.com/repos/rmacor/radargun/contributors","subscribers_url":"https://api.github.com/repos/rmacor/radargun/subscribers","subscription_url":"https://api.github.com/repos/rmacor/radargun/subscription","commits_url":"https://api.github.com/repos/rmacor/radargun/commits{/sha}","git_commits_url":"https://api.github.com/repos/rmacor/radargun/git/commits{/sha}","comments_url":"https://api.github.com/repos/rmacor/radargun/comments{/number}","issue_comment_url":"https://api.github.com/repos/rmacor/radargun/issues/comments{/number}","contents_url":"https://api.github.com/repos/rmacor/radargun/contents/{+path}","compare_url":"https://api.github.com/repos/rmacor/radargun/compare/{base}...{head}","merges_url":"https://api.github.com/repos/rmacor/radargun/merges","archive_url":"https://api.github.com/repos/rmacor/radargun/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/rmacor/radargun/downloads","issues_url":"https://api.github.com/repos/rmacor/radargun/issues{/number}","pulls_url":"https://api.github.com/repos/rmacor/radargun/pulls{/number}","milestones_url":"https://api.github.com/repos/rmacor/radargun/milestones{/number}","notifications_url":"https://api.github.com/repos/rmacor/radargun/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/rmacor/radargun/labels{/name}","releases_url":"https://api.github.com/repos/rmacor/radargun/releases{/id}","created_at":"2015-07-30T08:09:35Z","updated_at":"2015-07-30T08:09:37Z","pushed_at":"2015-11-09T09:39:26Z","git_url":"git://github.com/rmacor/radargun.git","ssh_url":"git@github.com:rmacor/radargun.git","clone_url":"https://github.com/rmacor/radargun.git","svn_url":"https://github.com/rmacor/radargun","homepage":"http://radargun.github.com/radargun/","size":5350,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"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":"radargun:master","ref":"master","sha":"b7cabd19e4c67dd2374ddf16450b75b8f7eb0b6d","user":{"login":"radargun","id":517796,"avatar_url":"https://avatars.githubusercontent.com/u/517796?v=3","gravatar_id":"","url":"https://api.github.com/users/radargun","html_url":"https://github.com/radargun","followers_url":"https://api.github.com/users/radargun/followers","following_url":"https://api.github.com/users/radargun/following{/other_user}","gists_url":"https://api.github.com/users/radargun/gists{/gist_id}","starred_url":"https://api.github.com/users/radargun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/radargun/subscriptions","organizations_url":"https://api.github.com/users/radargun/orgs","repos_url":"https://api.github.com/users/radargun/repos","events_url":"https://api.github.com/users/radargun/events{/privacy}","received_events_url":"https://api.github.com/users/radargun/received_events","type":"Organization","site_admin":false},"repo":{"id":1156136,"name":"radargun","full_name":"radargun/radargun","owner":{"login":"radargun","id":517796,"avatar_url":"https://avatars.githubusercontent.com/u/517796?v=3","gravatar_id":"","url":"https://api.github.com/users/radargun","html_url":"https://github.com/radargun","followers_url":"https://api.github.com/users/radargun/followers","following_url":"https://api.github.com/users/radargun/following{/other_user}","gists_url":"https://api.github.com/users/radargun/gists{/gist_id}","starred_url":"https://api.github.com/users/radargun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/radargun/subscriptions","organizations_url":"https://api.github.com/users/radargun/orgs","repos_url":"https://api.github.com/users/radargun/repos","events_url":"https://api.github.com/users/radargun/events{/privacy}","received_events_url":"https://api.github.com/users/radargun/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/radargun/radargun","description":"The RadarGun benchmarking framework for data grids and distributed caches","fork":false,"url":"https://api.github.com/repos/radargun/radargun","forks_url":"https://api.github.com/repos/radargun/radargun/forks","keys_url":"https://api.github.com/repos/radargun/radargun/keys{/key_id}","collaborators_url":"https://api.github.com/repos/radargun/radargun/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/radargun/radargun/teams","hooks_url":"https://api.github.com/repos/radargun/radargun/hooks","issue_events_url":"https://api.github.com/repos/radargun/radargun/issues/events{/number}","events_url":"https://api.github.com/repos/radargun/radargun/events","assignees_url":"https://api.github.com/repos/radargun/radargun/assignees{/user}","branches_url":"https://api.github.com/repos/radargun/radargun/branches{/branch}","tags_url":"https://api.github.com/repos/radargun/radargun/tags","blobs_url":"https://api.github.com/repos/radargun/radargun/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/radargun/radargun/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/radargun/radargun/git/refs{/sha}","trees_url":"https://api.github.com/repos/radargun/radargun/git/trees{/sha}","statuses_url":"https://api.github.com/repos/radargun/radargun/statuses/{sha}","languages_url":"https://api.github.com/repos/radargun/radargun/languages","stargazers_url":"https://api.github.com/repos/radargun/radargun/stargazers","contributors_url":"https://api.github.com/repos/radargun/radargun/contributors","subscribers_url":"https://api.github.com/repos/radargun/radargun/subscribers","subscription_url":"https://api.github.com/repos/radargun/radargun/subscription","commits_url":"https://api.github.com/repos/radargun/radargun/commits{/sha}","git_commits_url":"https://api.github.com/repos/radargun/radargun/git/commits{/sha}","comments_url":"https://api.github.com/repos/radargun/radargun/comments{/number}","issue_comment_url":"https://api.github.com/repos/radargun/radargun/issues/comments{/number}","contents_url":"https://api.github.com/repos/radargun/radargun/contents/{+path}","compare_url":"https://api.github.com/repos/radargun/radargun/compare/{base}...{head}","merges_url":"https://api.github.com/repos/radargun/radargun/merges","archive_url":"https://api.github.com/repos/radargun/radargun/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/radargun/radargun/downloads","issues_url":"https://api.github.com/repos/radargun/radargun/issues{/number}","pulls_url":"https://api.github.com/repos/radargun/radargun/pulls{/number}","milestones_url":"https://api.github.com/repos/radargun/radargun/milestones{/number}","notifications_url":"https://api.github.com/repos/radargun/radargun/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/radargun/radargun/labels{/name}","releases_url":"https://api.github.com/repos/radargun/radargun/releases{/id}","created_at":"2010-12-10T13:35:42Z","updated_at":"2015-10-29T22:58:54Z","pushed_at":"2015-11-09T09:39:26Z","git_url":"git://github.com/radargun/radargun.git","ssh_url":"git@github.com:radargun/radargun.git","clone_url":"https://github.com/radargun/radargun.git","svn_url":"https://github.com/radargun/radargun","homepage":"http://radargun.github.com/radargun/","size":8350,"stargazers_count":63,"watchers_count":63,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":45,"mirror_url":null,"open_issues_count":30,"forks":45,"open_issues":30,"watchers":63,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/radargun/radargun/pulls/282"},"html":{"href":"https://github.com/radargun/radargun/pull/282"},"issue":{"href":"https://api.github.com/repos/radargun/radargun/issues/282"},"comments":{"href":"https://api.github.com/repos/radargun/radargun/issues/282/comments"},"review_comments":{"href":"https://api.github.com/repos/radargun/radargun/pulls/282/comments"},"review_comment":{"href":"https://api.github.com/repos/radargun/radargun/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/radargun/radargun/pulls/282/commits"},"statuses":{"href":"https://api.github.com/repos/radargun/radargun/statuses/c6ff0b25a74a8e31d12c1189c4245b56c9fbe51b"}}}}