{"action":"created","comment":{"url":"https://api.github.com/repos/apache/storm/pulls/comments/62433019","id":62433019,"diff_hunk":"@@ -0,0 +1,434 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one\n+ * or more contributor license agreements.  See the NOTICE file\n+ * distributed with this work for additional information\n+ * regarding copyright ownership.  The ASF licenses this file\n+ * to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance\n+ * with the License.  You may obtain a copy of the License at\n+ *\n+ * http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.apache.storm.scheduler;\n+\n+import java.util.ArrayList;\n+import java.util.Collection;\n+import java.util.Collections;\n+import java.util.Comparator;\n+import java.util.HashMap;\n+import java.util.HashSet;\n+import java.util.LinkedList;\n+import java.util.List;\n+import java.util.Map;\n+import java.util.Set;\n+import java.util.TreeMap;\n+\n+import org.apache.storm.utils.Utils;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+import org.apache.storm.Config;\n+import org.apache.storm.scheduler.Cluster;\n+import org.apache.storm.scheduler.ExecutorDetails;\n+import org.apache.storm.scheduler.IScheduler;\n+import org.apache.storm.scheduler.SchedulerAssignment;\n+import org.apache.storm.scheduler.Topologies;\n+import org.apache.storm.scheduler.TopologyDetails;\n+import org.apache.storm.scheduler.WorkerSlot;\n+\n+public class IsolationScheduler implements IScheduler {\n+    private final static Logger LOG = LoggerFactory.getLogger(IsolationScheduler.class);\n+\n+    private Map conf;\n+\n+    @Override\n+    public void prepare(Map conf) {\n+        this.conf = conf;\n+    }\n+\n+    // for each isolated topology:\n+    //   compute even distribution of executors -> workers on the number of workers specified for the topology\n+    //   compute distribution of workers to machines\n+    // determine host -> list of [slot, topology id, executors]\n+    // iterate through hosts and: a machine is good if:\n+    //   1. only running workers from one isolated topology\n+    //   2. all workers running on it match one of the distributions of executors for that topology\n+    //   3. matches one of the # of workers\n+    // blacklist the good hosts and remove those workers from the list of need to be assigned workers\n+    // otherwise unassign all other workers for isolated topologies if assigned\n+\n+    public Set<ExecutorDetails> removeElemFromSet(Set<Set<ExecutorDetails>> executorsSets) {\n+        if (executorsSets.iterator().hasNext()) {\n+            Set<ExecutorDetails> elem = executorsSets.iterator().next();\n+            executorsSets.remove(elem);\n+            return elem;\n+        } else {\n+            return null;\n+        }\n+    }\n+\n+    // get host -> all assignable worker slots for non-blacklisted machines (assigned or not assigned)\n+    // will then have a list of machines that need to be assigned (machine -> [topology, list of list of executors])\n+    // match each spec to a machine (who has the right number of workers), free everything else on that machine and assign those slots (do one topology at a time)\n+    // blacklist all machines who had production slots defined\n+    // log isolated topologies who weren't able to get enough slots / machines\n+    // run default scheduler on isolated topologies that didn't have enough slots + non-isolated topologies on remaining machines\n+    // set blacklist to what it was initially\n+    @Override\n+    public void schedule(Topologies topologies, Cluster cluster) {\n+        Set<String> origBlacklist = cluster.getBlacklistedHosts();\n+        List<TopologyDetails> isoTopologies = isolatedTopologies(topologies.getTopologies());\n+        Set<String> isoIds = isolatedTopoplogyIds(isoTopologies);\n+        Map<String, Set<Set<ExecutorDetails>>> topologyWorkerSpecs = topologyWorkerSpecs(isoTopologies);\n+        Map<String, Map<Integer, Integer>> topologyMachineDistributions = topologyMachineDistributions(isoTopologies);\n+        Map<String, List<AssignmentInfo>> hostAssignments = hostAssignments(cluster);\n+\n+        Set<Map.Entry<String, List<AssignmentInfo>>> hostAssignEntries = hostAssignments.entrySet();\n+        for (Map.Entry<String, List<AssignmentInfo>> entry : hostAssignEntries) {\n+            List<AssignmentInfo> assInfos = entry.getValue();\n+            String topologyId = assInfos.get(0).getTopologyId();\n+            Map<Integer, Integer> distribution = topologyMachineDistributions.get(topologyId);\n+            Set<Set<ExecutorDetails>> workerSpecs = topologyWorkerSpecs.get(topologyId);\n+            int numWorkers = assInfos.size();\n+\n+            if (isoIds.contains(topologyId)\n+                    && checkAssignmentTopology(assInfos, topologyId)\n+                    && (distribution != null && distribution.containsKey(numWorkers))\n+                    && checkAssignmentWorkerSpecs(assInfos, workerSpecs)) {\n+                decrementDistribution(distribution, numWorkers);\n+                for (AssignmentInfo ass : assInfos) {\n+                    workerSpecs.remove(ass.getExecutors());\n+                }\n+                cluster.blacklistHost(entry.getKey());\n+            } else {\n+                for (AssignmentInfo ass : assInfos) {\n+                    if (isoIds.contains(ass.getTopologyId())) {\n+                        cluster.freeSlot(ass.getWorkerSlot());\n+                    }\n+                }\n+            }\n+        }\n+\n+        Map<String, Set<WorkerSlot>> hostUsedSlots = hostUsedSlots(cluster);\n+        LinkedList<HostAssignableSlots> hss = hostAssignableSlots(cluster);\n+        List<String> failedTopologyIds = new ArrayList<String>();\n+        for (Map.Entry<String, Set<Set<ExecutorDetails>>> entry : topologyWorkerSpecs.entrySet()) {\n+            String topologyId = entry.getKey();\n+            Set<Set<ExecutorDetails>> executorSet = entry.getValue();\n+            if (executorSet != null && executorSet.size() > 0) {\n+                failedTopologyIds.add(topologyId);\n+            }\n+            List<Integer> workerNum = distributionSortedAmts(topologyMachineDistributions.get(topologyId));\n+            for (Integer num : workerNum) {\n+                HostAssignableSlots hostSlots = hss.peek();\n+                List<WorkerSlot> slot = hostSlots != null ? hostSlots.getWorkerSlots() : null;\n+\n+                if (slot != null && slot.size() >= num.intValue()) {\n+                    hss.poll();\n+                    cluster.freeSlots(hostUsedSlots.get(hostSlots.getHostName()));\n+                    for (WorkerSlot tmpSlot : slot.subList(0, num)) {\n+                        Set<ExecutorDetails> executor = removeElemFromSet(executorSet);\n+                        cluster.assign(tmpSlot, topologyId, executor);\n+                    }\n+                    cluster.blacklistHost(hostSlots.getHostName());\n+                }\n+            }\n+        }\n+\n+        if (failedTopologyIds.size() > 0) {\n+            LOG.warn(\"Unable to isolate topologies \" + failedTopologyIds\n+                    + \". No machine had enough worker slots to run the remaining workers for these topologies. \"\n+                    + \"Clearing all other resources and will wait for enough resources for \"\n+                    + \"isolated topologies before allocating any other resources.\");\n+            // clear workers off all hosts that are not blacklisted\n+            Map<String, Set<WorkerSlot>> usedSlots = hostUsedSlots(cluster);\n+            Set<Map.Entry<String, Set<WorkerSlot>>> entries = usedSlots.entrySet();\n+            for (Map.Entry<String, Set<WorkerSlot>> entry : entries) {\n+                if (!cluster.isBlacklistedHost(entry.getKey())) {\n+                    cluster.freeSlots(entry.getValue());\n+                }\n+            }\n+        } else {\n+            // run default scheduler on non-isolated topologies\n+            Set<String> allocatedTopologies = allocatedTopologies(topologyWorkerSpecs);\n+            Topologies leftOverTopologies = leftoverTopologies(topologies, allocatedTopologies);\n+            DefaultScheduler.defaultSchedule(leftOverTopologies, cluster);\n+        }\n+        cluster.setBlacklistedHosts(origBlacklist);\n+    }\n+\n+    private List<TopologyDetails> isolatedTopologies(Collection<TopologyDetails> topologies) {\n+        Object isoMachines = conf.get(Config.ISOLATION_SCHEDULER_MACHINES);","path":"storm-core/src/jvm/org/apache/storm/scheduler/IsolationScheduler.java","position":166,"original_position":166,"commit_id":"b29adfe1b4119ebcc8f0aa185f4c65e634b08768","original_commit_id":"b29adfe1b4119ebcc8f0aa185f4c65e634b08768","user":{"login":"abhishekagarwal87","id":1477457,"avatar_url":"https://avatars.githubusercontent.com/u/1477457?v=3","gravatar_id":"","url":"https://api.github.com/users/abhishekagarwal87","html_url":"https://github.com/abhishekagarwal87","followers_url":"https://api.github.com/users/abhishekagarwal87/followers","following_url":"https://api.github.com/users/abhishekagarwal87/following{/other_user}","gists_url":"https://api.github.com/users/abhishekagarwal87/gists{/gist_id}","starred_url":"https://api.github.com/users/abhishekagarwal87/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/abhishekagarwal87/subscriptions","organizations_url":"https://api.github.com/users/abhishekagarwal87/orgs","repos_url":"https://api.github.com/users/abhishekagarwal87/repos","events_url":"https://api.github.com/users/abhishekagarwal87/events{/privacy}","received_events_url":"https://api.github.com/users/abhishekagarwal87/received_events","type":"User","site_admin":false},"body":"can we make this Map<String, Number> isoMachines","created_at":"2016-05-08T14:50:53Z","updated_at":"2016-05-08T14:50:53Z","html_url":"https://github.com/apache/storm/pull/1359#discussion_r62433019","pull_request_url":"https://api.github.com/repos/apache/storm/pulls/1359","_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/comments/62433019"},"html":{"href":"https://github.com/apache/storm/pull/1359#discussion_r62433019"},"pull_request":{"href":"https://api.github.com/repos/apache/storm/pulls/1359"}}},"pull_request":{"url":"https://api.github.com/repos/apache/storm/pulls/1359","id":67635274,"html_url":"https://github.com/apache/storm/pull/1359","diff_url":"https://github.com/apache/storm/pull/1359.diff","patch_url":"https://github.com/apache/storm/pull/1359.patch","issue_url":"https://api.github.com/repos/apache/storm/issues/1359","number":1359,"state":"open","locked":false,"title":"STORM-1239: port backtype.storm.scheduler.IsolationScheduler to java","user":{"login":"vesense","id":6711230,"avatar_url":"https://avatars.githubusercontent.com/u/6711230?v=3","gravatar_id":"","url":"https://api.github.com/users/vesense","html_url":"https://github.com/vesense","followers_url":"https://api.github.com/users/vesense/followers","following_url":"https://api.github.com/users/vesense/following{/other_user}","gists_url":"https://api.github.com/users/vesense/gists{/gist_id}","starred_url":"https://api.github.com/users/vesense/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vesense/subscriptions","organizations_url":"https://api.github.com/users/vesense/orgs","repos_url":"https://api.github.com/users/vesense/repos","events_url":"https://api.github.com/users/vesense/events{/privacy}","received_events_url":"https://api.github.com/users/vesense/received_events","type":"User","site_admin":false},"body":"","created_at":"2016-04-24T12:00:11Z","updated_at":"2016-05-08T14:50:53Z","closed_at":null,"merged_at":null,"merge_commit_sha":"28d79793c0494fef57ba266148990d4dba4ae891","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/storm/pulls/1359/commits","review_comments_url":"https://api.github.com/repos/apache/storm/pulls/1359/comments","review_comment_url":"https://api.github.com/repos/apache/storm/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/storm/issues/1359/comments","statuses_url":"https://api.github.com/repos/apache/storm/statuses/b29adfe1b4119ebcc8f0aa185f4c65e634b08768","head":{"label":"vesense:STORM-1239","ref":"STORM-1239","sha":"b29adfe1b4119ebcc8f0aa185f4c65e634b08768","user":{"login":"vesense","id":6711230,"avatar_url":"https://avatars.githubusercontent.com/u/6711230?v=3","gravatar_id":"","url":"https://api.github.com/users/vesense","html_url":"https://github.com/vesense","followers_url":"https://api.github.com/users/vesense/followers","following_url":"https://api.github.com/users/vesense/following{/other_user}","gists_url":"https://api.github.com/users/vesense/gists{/gist_id}","starred_url":"https://api.github.com/users/vesense/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vesense/subscriptions","organizations_url":"https://api.github.com/users/vesense/orgs","repos_url":"https://api.github.com/users/vesense/repos","events_url":"https://api.github.com/users/vesense/events{/privacy}","received_events_url":"https://api.github.com/users/vesense/received_events","type":"User","site_admin":false},"repo":{"id":30167243,"name":"storm","full_name":"vesense/storm","owner":{"login":"vesense","id":6711230,"avatar_url":"https://avatars.githubusercontent.com/u/6711230?v=3","gravatar_id":"","url":"https://api.github.com/users/vesense","html_url":"https://github.com/vesense","followers_url":"https://api.github.com/users/vesense/followers","following_url":"https://api.github.com/users/vesense/following{/other_user}","gists_url":"https://api.github.com/users/vesense/gists{/gist_id}","starred_url":"https://api.github.com/users/vesense/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/vesense/subscriptions","organizations_url":"https://api.github.com/users/vesense/orgs","repos_url":"https://api.github.com/users/vesense/repos","events_url":"https://api.github.com/users/vesense/events{/privacy}","received_events_url":"https://api.github.com/users/vesense/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/vesense/storm","description":"Mirror of Apache Storm","fork":true,"url":"https://api.github.com/repos/vesense/storm","forks_url":"https://api.github.com/repos/vesense/storm/forks","keys_url":"https://api.github.com/repos/vesense/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/vesense/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/vesense/storm/teams","hooks_url":"https://api.github.com/repos/vesense/storm/hooks","issue_events_url":"https://api.github.com/repos/vesense/storm/issues/events{/number}","events_url":"https://api.github.com/repos/vesense/storm/events","assignees_url":"https://api.github.com/repos/vesense/storm/assignees{/user}","branches_url":"https://api.github.com/repos/vesense/storm/branches{/branch}","tags_url":"https://api.github.com/repos/vesense/storm/tags","blobs_url":"https://api.github.com/repos/vesense/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/vesense/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/vesense/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/vesense/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/vesense/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/vesense/storm/languages","stargazers_url":"https://api.github.com/repos/vesense/storm/stargazers","contributors_url":"https://api.github.com/repos/vesense/storm/contributors","subscribers_url":"https://api.github.com/repos/vesense/storm/subscribers","subscription_url":"https://api.github.com/repos/vesense/storm/subscription","commits_url":"https://api.github.com/repos/vesense/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/vesense/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/vesense/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/vesense/storm/issues/comments{/number}","contents_url":"https://api.github.com/repos/vesense/storm/contents/{+path}","compare_url":"https://api.github.com/repos/vesense/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/vesense/storm/merges","archive_url":"https://api.github.com/repos/vesense/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/vesense/storm/downloads","issues_url":"https://api.github.com/repos/vesense/storm/issues{/number}","pulls_url":"https://api.github.com/repos/vesense/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/vesense/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/vesense/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/vesense/storm/labels{/name}","releases_url":"https://api.github.com/repos/vesense/storm/releases{/id}","deployments_url":"https://api.github.com/repos/vesense/storm/deployments","created_at":"2015-02-02T02:07:13Z","updated_at":"2016-01-13T01:35:05Z","pushed_at":"2016-05-02T11:06:00Z","git_url":"git://github.com/vesense/storm.git","ssh_url":"git@github.com:vesense/storm.git","clone_url":"https://github.com/vesense/storm.git","svn_url":"https://github.com/vesense/storm","homepage":null,"size":105285,"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":"4e3c5bd5427cfad76f04d0c6e68dc321d516547e","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":14135470,"name":"storm","full_name":"apache/storm","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/storm","description":"Mirror of Apache Storm","fork":false,"url":"https://api.github.com/repos/apache/storm","forks_url":"https://api.github.com/repos/apache/storm/forks","keys_url":"https://api.github.com/repos/apache/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/storm/teams","hooks_url":"https://api.github.com/repos/apache/storm/hooks","issue_events_url":"https://api.github.com/repos/apache/storm/issues/events{/number}","events_url":"https://api.github.com/repos/apache/storm/events","assignees_url":"https://api.github.com/repos/apache/storm/assignees{/user}","branches_url":"https://api.github.com/repos/apache/storm/branches{/branch}","tags_url":"https://api.github.com/repos/apache/storm/tags","blobs_url":"https://api.github.com/repos/apache/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/storm/languages","stargazers_url":"https://api.github.com/repos/apache/storm/stargazers","contributors_url":"https://api.github.com/repos/apache/storm/contributors","subscribers_url":"https://api.github.com/repos/apache/storm/subscribers","subscription_url":"https://api.github.com/repos/apache/storm/subscription","commits_url":"https://api.github.com/repos/apache/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/storm/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/storm/contents/{+path}","compare_url":"https://api.github.com/repos/apache/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/storm/merges","archive_url":"https://api.github.com/repos/apache/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/storm/downloads","issues_url":"https://api.github.com/repos/apache/storm/issues{/number}","pulls_url":"https://api.github.com/repos/apache/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/storm/labels{/name}","releases_url":"https://api.github.com/repos/apache/storm/releases{/id}","deployments_url":"https://api.github.com/repos/apache/storm/deployments","created_at":"2013-11-05T08:00:14Z","updated_at":"2016-05-08T14:03:16Z","pushed_at":"2016-05-06T12:05:28Z","git_url":"git://github.com/apache/storm.git","ssh_url":"git@github.com:apache/storm.git","clone_url":"https://github.com/apache/storm.git","svn_url":"https://github.com/apache/storm","homepage":null,"size":169219,"stargazers_count":3136,"watchers_count":3136,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":2340,"mirror_url":"git://git.apache.org/storm.git","open_issues_count":127,"forks":2340,"open_issues":127,"watchers":3136,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/1359"},"html":{"href":"https://github.com/apache/storm/pull/1359"},"issue":{"href":"https://api.github.com/repos/apache/storm/issues/1359"},"comments":{"href":"https://api.github.com/repos/apache/storm/issues/1359/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/storm/pulls/1359/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/storm/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/storm/pulls/1359/commits"},"statuses":{"href":"https://api.github.com/repos/apache/storm/statuses/b29adfe1b4119ebcc8f0aa185f4c65e634b08768"}}}}