Merge pull request #3214 from RuiLi8080/MINOR-Nimbus
[MINOR] remove some unused fields and correct comment typo
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index 121dbdb..1989d1b 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -2044,20 +2044,17 @@
* compute a topology-id -> alive executors map.
*
* @param existingAssignment the current assignments
- * @param topologies the current topologies
* @param topologyToExecutors the executors for the current topologies
* @param scratchTopologyId the topology being rebalanced and should be excluded
* @return the map of topology id to alive executors
*/
private Map<String, Set<List<Integer>>> computeTopologyToAliveExecutors(Map<String, Assignment> existingAssignment,
- Topologies topologies,
Map<String, Set<List<Integer>>> topologyToExecutors,
String scratchTopologyId) {
Map<String, Set<List<Integer>>> ret = new HashMap<>();
for (Entry<String, Assignment> entry : existingAssignment.entrySet()) {
String topoId = entry.getKey();
Assignment assignment = entry.getValue();
- TopologyDetails td = topologies.getById(topoId);
Set<List<Integer>> allExecutors = topologyToExecutors.get(topoId);
Set<List<Integer>> aliveExecutors;
if (topoId.equals(scratchTopologyId)) {
@@ -2226,8 +2223,8 @@
updateAllHeartbeats(existingAssignments, topoToExec, zkHeartbeatTopologies);
- Map<String, Set<List<Integer>>> topoToAliveExecutors = computeTopologyToAliveExecutors(existingAssignments, topologies,
- topoToExec, scratchTopologyId);
+ Map<String, Set<List<Integer>>> topoToAliveExecutors = computeTopologyToAliveExecutors(existingAssignments, topoToExec,
+ scratchTopologyId);
Map<String, Set<Long>> supervisorToDeadPorts = computeSupervisorToDeadPorts(existingAssignments, topoToExec,
topoToAliveExecutors);
Map<String, SchedulerAssignmentImpl> topoToSchedAssignment = computeTopologyToSchedulerAssignment(existingAssignments,
@@ -2373,8 +2370,9 @@
for (Iterator<Entry<String, StormBase>> it = bases.entrySet().iterator(); it.hasNext(); ) {
Entry<String, StormBase> entry = it.next();
String id = entry.getKey();
+ StormBase base = entry.getValue();
try {
- tds.put(id, readTopologyDetails(id, entry.getValue()));
+ tds.put(id, readTopologyDetails(id, base));
} catch (KeyNotFoundException e) {
//A race happened and it is probably not running
it.remove();
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
index 6ab379f..2052eb5 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/BaseResourceAwareStrategy.java
@@ -510,7 +510,7 @@
* this component and then from each neighboring component in sorted order. Do this until there is nothing left to schedule.
*
* @param td The topology the executors belong to
- * @param unassignedExecutors a collection of unassigned executors that need to be unassigned. Should only try to assign executors from
+ * @param unassignedExecutors a collection of unassigned executors that need to be assigned. Should only try to assign executors from
* this list
* @return a list of executors in sorted order
*/
@@ -521,7 +521,7 @@
Map<String, Queue<ExecutorDetails>> compToExecsToSchedule = new HashMap<>();
for (Component component : componentMap.values()) {
- compToExecsToSchedule.put(component.getId(), new LinkedList<ExecutorDetails>());
+ compToExecsToSchedule.put(component.getId(), new LinkedList<>());
for (ExecutorDetails exec : component.getExecs()) {
if (unassignedExecutors.contains(exec)) {
compToExecsToSchedule.get(component.getId()).add(exec);
@@ -533,7 +533,7 @@
sortedComponents.addAll(componentMap.values());
for (Component currComp : sortedComponents) {
- Map<String, Component> neighbors = new HashMap<String, Component>();
+ Map<String, Component> neighbors = new HashMap<>();
for (String compId : Sets.union(currComp.getChildren(), currComp.getParents())) {
neighbors.put(compId, componentMap.get(compId));
}
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
index e6d93db..96d8da9 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/priority/TestFIFOSchedulingPriorityStrategy.java
@@ -52,10 +52,10 @@
config.put(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY, FIFOSchedulingPriorityStrategy.class.getName());
Topologies topologies = new Topologies(
- genTopology("topo-1-jerry", config, 1, 0, 1, 0,Time.currentTimeSecs() - 250,20, "jerry"),
- genTopology("topo-2-bobby", config, 1, 0, 1, 0,Time.currentTimeSecs() - 200,10, "bobby"),
- genTopology("topo-3-bobby", config, 1, 0, 1, 0,Time.currentTimeSecs() - 300,20, "bobby"),
- genTopology("topo-4-derek", config, 1, 0, 1, 0,Time.currentTimeSecs() - 201,29, "derek"));
+ genTopology("topo-1-jerry", config, 1, 0, 1, 0, Time.currentTimeSecs() - 250, 20, "jerry"),
+ genTopology("topo-2-bobby", config, 1, 0, 1, 0, Time.currentTimeSecs() - 200, 10, "bobby"),
+ genTopology("topo-3-bobby", config, 1, 0, 1, 0, Time.currentTimeSecs() - 300, 20, "bobby"),
+ genTopology("topo-4-derek", config, 1, 0, 1, 0, Time.currentTimeSecs() - 201, 29, "derek"));
Cluster cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
ResourceAwareScheduler rs = new ResourceAwareScheduler();
@@ -69,7 +69,7 @@
//new topology needs to be scheduled
//topo-3 should be evicted since its been up the longest
topologies = addTopologies(topologies,
- genTopology("topo-5-derek", config, 1, 0, 1, 0, Time.currentTimeSecs() - 15, 29, "derek"));
+ genTopology("topo-5-derek", config, 1, 0, 1, 0, Time.currentTimeSecs() - 15, 29, "derek"));
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
rs.schedule(topologies, cluster);
@@ -81,7 +81,7 @@
//new topology needs to be scheduled. topo-4 should be evicted. Even though topo-1 from user jerry is older, topo-1 will not be evicted
//since user jerry has enough resource guarantee
topologies = addTopologies(topologies,
- genTopology("topo-6-bobby", config, 1, 0, 1, 0, Time.currentTimeSecs() - 10, 29, "bobby"));
+ genTopology("topo-6-bobby", config, 1, 0, 1, 0, Time.currentTimeSecs() - 10, 29, "bobby"));
cluster = new Cluster(iNimbus, new ResourceMetrics(new StormMetricsRegistry()), supMap, new HashMap<>(), topologies, config);
rs.schedule(topologies, cluster);
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
index 6996ec7..5c28a87 100644
--- a/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestGenericResourceAwareStrategy.java
@@ -300,8 +300,7 @@
assertTopologiesFullyScheduled(cluster, noGpu);
assertTopologiesFullyScheduled(cluster, gpu2);
}
-
-
+
@Test
public void testAntiAffinityWithMultipleTopologies() {
INimbus iNimbus = new INimbusTest();
