You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2015/10/31 18:57:36 UTC
[1/3] storm git commit: STORM-1152 Change map keySet iteration to
entrySet iteration for efficiency
Repository: storm
Updated Branches:
refs/heads/master 3a724ee3c -> 72fc3beb4
STORM-1152 Change map keySet iteration to entrySet iteration for efficiency
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/bd0e8a01
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/bd0e8a01
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/bd0e8a01
Branch: refs/heads/master
Commit: bd0e8a01bba16f156c544df107345a7f9703dc57
Parents: 0bba2ba
Author: Suresh Srinivas <su...@yahoo-inc.com>
Authored: Fri Oct 30 11:45:08 2015 -0700
Committer: Suresh Srinivas <su...@yahoo-inc.com>
Committed: Fri Oct 30 14:36:42 2015 -0700
----------------------------------------------------------------------
.../jvm/backtype/storm/scheduler/Cluster.java | 64 ++++++--------------
.../backtype/storm/scheduler/Topologies.java | 6 +-
.../authorizer/DRPCSimpleACLAuthorizer.java | 8 +--
.../authorizer/ImpersonationAuthorizer.java | 7 ++-
.../serialization/SerializationFactory.java | 6 +-
.../storm/task/GeneralTopologyContext.java | 5 +-
...uePartitionedTransactionalSpoutExecutor.java | 5 +-
.../backtype/storm/utils/TransferDrainer.java | 17 +++---
.../src/jvm/backtype/storm/utils/Utils.java | 19 +++---
.../OpaquePartitionedTridentSpoutExecutor.java | 10 +--
.../topology/TridentTopologyBuilder.java | 23 +++----
11 files changed, 74 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
index dd15999..f604013 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
@@ -68,8 +68,9 @@ public class Cluster {
this.assignments.putAll(assignments);
this.status = new HashMap<String, String>();
this.hostToId = new HashMap<String, List<String>>();
- for (String nodeId : supervisors.keySet()) {
- SupervisorDetails supervisor = supervisors.get(nodeId);
+ for (Map.Entry<String, SupervisorDetails> entry : supervisors.entrySet()) {
+ String nodeId = entry.getKey();
+ SupervisorDetails supervisor = entry.getValue();
String host = supervisor.getHost();
if (!this.hostToId.containsKey(host)) {
this.hostToId.put(host, new ArrayList<String>());
@@ -108,10 +109,7 @@ public class Cluster {
}
/**
- * Gets all the topologies which needs scheduling.
- *
- * @param topologies
- * @return
+ * @return all the topologies which needs scheduling.
*/
public List<TopologyDetails> needsSchedulingTopologies(Topologies topologies) {
List<TopologyDetails> ret = new ArrayList<TopologyDetails>();
@@ -136,19 +134,12 @@ public class Cluster {
public boolean needsScheduling(TopologyDetails topology) {
int desiredNumWorkers = topology.getNumWorkers();
int assignedNumWorkers = this.getAssignedNumWorkers(topology);
-
- if (desiredNumWorkers > assignedNumWorkers) {
- return true;
- }
-
- return this.getUnassignedExecutors(topology).size() > 0;
+ return desiredNumWorkers > assignedNumWorkers || this.getUnassignedExecutors(topology).size() > 0;
}
/**
- * Gets a executor -> component-id map which needs scheduling in this topology.
- *
* @param topology
- * @return
+ * @return a executor -> component-id map which needs scheduling in this topology.
*/
public Map<ExecutorDetails, String> getNeedsSchedulingExecutorToComponents(TopologyDetails topology) {
Collection<ExecutorDetails> allExecutors = new HashSet(topology.getExecutors());
@@ -163,16 +154,15 @@ public class Cluster {
}
/**
- * Gets a component-id -> executors map which needs scheduling in this topology.
- *
* @param topology
- * @return
+ * @return a component-id -> executors map which needs scheduling in this topology.
*/
public Map<String, List<ExecutorDetails>> getNeedsSchedulingComponentToExecutors(TopologyDetails topology) {
Map<ExecutorDetails, String> executorToComponents = this.getNeedsSchedulingExecutorToComponents(topology);
Map<String, List<ExecutorDetails>> componentToExecutors = new HashMap<String, List<ExecutorDetails>>();
- for (ExecutorDetails executor : executorToComponents.keySet()) {
- String component = executorToComponents.get(executor);
+ for (Map.Entry<ExecutorDetails, String> entry : executorToComponents.entrySet()) {
+ ExecutorDetails executor = entry.getKey();
+ String component = entry.getValue();
if (!componentToExecutors.containsKey(component)) {
componentToExecutors.put(component, new ArrayList<ExecutorDetails>());
}
@@ -186,9 +176,6 @@ public class Cluster {
/**
* Get all the used ports of this supervisor.
- *
- * @param cluster
- * @return
*/
public Set<Integer> getUsedPorts(SupervisorDetails supervisor) {
Map<String, SchedulerAssignment> assignments = this.getAssignments();
@@ -207,9 +194,6 @@ public class Cluster {
/**
* Return the available ports of this supervisor.
- *
- * @param cluster
- * @return
*/
public Set<Integer> getAvailablePorts(SupervisorDetails supervisor) {
Set<Integer> usedPorts = this.getUsedPorts(supervisor);
@@ -228,9 +212,6 @@ public class Cluster {
/**
* Return all the available slots on this supervisor.
- *
- * @param cluster
- * @return
*/
public List<WorkerSlot> getAvailableSlots(SupervisorDetails supervisor) {
Set<Integer> ports = this.getAvailablePorts(supervisor);
@@ -274,10 +255,8 @@ public class Cluster {
}
/**
- * Gets the number of workers assigned to this topology.
- *
* @param topology
- * @return
+ * @return the number of workers assigned to this topology.
*/
public int getAssignedNumWorkers(TopologyDetails topology) {
SchedulerAssignment assignment = this.getAssignmentById(topology.getId());
@@ -317,9 +296,7 @@ public class Cluster {
}
/**
- * Gets all the available slots in the cluster.
- *
- * @return
+ * @return all the available worker slots in the cluster.
*/
public List<WorkerSlot> getAvailableSlots() {
List<WorkerSlot> slots = new ArrayList<WorkerSlot>();
@@ -367,10 +344,8 @@ public class Cluster {
}
/**
- * Checks the specified slot is occupied.
- *
* @param slot the slot be to checked.
- * @return
+ * @return true if the specified slot is occupied.
*/
public boolean isSlotOccupied(WorkerSlot slot) {
for (SchedulerAssignment assignment : this.assignments.values()) {
@@ -467,13 +442,14 @@ public class Cluster {
DNSToSwitchMapping topographyMapper = (DNSToSwitchMapping) Utils.newInstance(clazz);
Map<String, String> resolvedSuperVisors = topographyMapper.resolve(supervisorHostNames);
- for (String hostName : resolvedSuperVisors.keySet()) {
- String rack = resolvedSuperVisors.get(hostName);
-
- if (!networkTopography.containsKey(rack)) {
- networkTopography.put(rack, new ArrayList<String>());
- }
+ for (Map.Entry<String, String> entry : resolvedSuperVisors.entrySet()) {
+ String hostName = entry.getKey();
+ String rack = entry.getValue();
List<String> nodesForRack = networkTopography.get(rack);
+ if (nodesForRack == null) {
+ nodesForRack = new ArrayList<String>();
+ networkTopography.put(rack, nodesForRack);
+ }
nodesForRack.add(hostName);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
index 409ecb3..443bf3f 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
@@ -34,9 +34,9 @@ public class Topologies {
this.topologies.putAll(topologies);
this.nameToId = new HashMap<String, String>(topologies.size());
- for (String topologyId : topologies.keySet()) {
- TopologyDetails topology = topologies.get(topologyId);
- this.nameToId.put(topology.getName(), topologyId);
+ for (Map.Entry<String, TopologyDetails> entry : topologies.entrySet()) {
+ TopologyDetails topology = entry.getValue();
+ this.nameToId.put(topology.getName(), entry.getKey());
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
index 8ce26ed..5e83b9f 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
@@ -37,7 +37,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase {
- public static Logger LOG =
+ public static final Logger LOG =
LoggerFactory.getLogger(DRPCSimpleACLAuthorizer.class);
public static final String CLIENT_USERS_KEY = "client.users";
@@ -75,15 +75,15 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase {
(Map<String,Map<String,?>>)
conf.get(Config.DRPC_AUTHORIZER_ACL);
- for (String function : confAcl.keySet()) {
- Map<String,?> val = confAcl.get(function);
+ for (Map.Entry<String, Map<String, ?>> entry : confAcl.entrySet()) {
+ Map<String,?> val = entry.getValue();
Collection<String> clientUsers =
val.containsKey(CLIENT_USERS_KEY) ?
(Collection<String>) val.get(CLIENT_USERS_KEY) : null;
String invocationUser =
val.containsKey(INVOCATION_USER_KEY) ?
(String) val.get(INVOCATION_USER_KEY) : null;
- acl.put(function,
+ acl.put(entry.getKey(),
new AclFunctionEntry(clientUsers, invocationUser));
}
} else if (!_permitWhenMissingFunctionEntry) {
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/security/auth/authorizer/ImpersonationAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/ImpersonationAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/ImpersonationAuthorizer.java
index c1573a6..0cfd488 100644
--- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/ImpersonationAuthorizer.java
+++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/ImpersonationAuthorizer.java
@@ -43,9 +43,10 @@ public class ImpersonationAuthorizer implements IAuthorizer {
Map<String, Map<String, List<String>>> userToHostAndGroup = (Map<String, Map<String, List<String>>>) conf.get(Config.NIMBUS_IMPERSONATION_ACL);
if (userToHostAndGroup != null) {
- for (String user : userToHostAndGroup.keySet()) {
- Set<String> groups = ImmutableSet.copyOf(userToHostAndGroup.get(user).get("groups"));
- Set<String> hosts = ImmutableSet.copyOf(userToHostAndGroup.get(user).get("hosts"));
+ for (Map.Entry<String, Map<String, List<String>>> entry : userToHostAndGroup.entrySet()) {
+ String user = entry.getKey();
+ Set<String> groups = ImmutableSet.copyOf(entry.getValue().get("groups"));
+ Set<String> hosts = ImmutableSet.copyOf(entry.getValue().get("hosts"));
userImpersonationACL.put(user, new ImpersonationACL(user, groups, hosts));
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java b/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java
index 5549b1d..2a829d1 100644
--- a/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java
+++ b/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java
@@ -81,10 +81,10 @@ public class SerializationFactory {
kryoFactory.preRegister(k, conf);
boolean skipMissing = (Boolean) conf.get(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS);
- for(String klassName: registrations.keySet()) {
- String serializerClassName = registrations.get(klassName);
+ for(Map.Entry<String, String> entry: registrations.entrySet()) {
+ String serializerClassName = entry.getValue();
try {
- Class klass = Class.forName(klassName);
+ Class klass = Class.forName(entry.getKey());
Class serializerClass = null;
if(serializerClassName!=null)
serializerClass = Class.forName(serializerClassName);
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java b/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java
index 53141b1..54de390 100644
--- a/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java
+++ b/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java
@@ -141,11 +141,12 @@ public class GeneralTopologyContext implements JSONAware {
Map<String, Map<String, Grouping>> ret = new HashMap<String, Map<String, Grouping>>();
for(String otherComponentId: getComponentIds()) {
Map<GlobalStreamId, Grouping> inputs = getComponentCommon(otherComponentId).get_inputs();
- for(GlobalStreamId id: inputs.keySet()) {
+ for(Map.Entry<GlobalStreamId, Grouping> entry: inputs.entrySet()) {
+ GlobalStreamId id = entry.getKey();
if(id.get_componentId().equals(componentId)) {
Map<String, Grouping> curr = ret.get(id.get_streamId());
if(curr==null) curr = new HashMap<String, Grouping>();
- curr.put(otherComponentId, inputs.get(id));
+ curr.put(otherComponentId, entry.getValue());
ret.put(id.get_streamId(), curr);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java b/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
index aabcb7a..3be5d37 100644
--- a/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
+++ b/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java
@@ -120,8 +120,9 @@ public class OpaquePartitionedTransactionalSpoutExecutor implements ICommitterTr
public void commit(TransactionAttempt attempt) {
BigInteger txid = attempt.getTransactionId();
Map<Integer, Object> metas = _cachedMetas.remove(txid);
- for(Integer partition: metas.keySet()) {
- Object meta = metas.get(partition);
+ for(Entry<Integer, Object> entry: metas.entrySet()) {
+ Integer partition = entry.getKey();
+ Object meta = entry.getValue();
_partitionStates.get(partition).overrideState(txid, meta);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/utils/TransferDrainer.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/TransferDrainer.java b/storm-core/src/jvm/backtype/storm/utils/TransferDrainer.java
index 5a111ce..feb0328 100644
--- a/storm-core/src/jvm/backtype/storm/utils/TransferDrainer.java
+++ b/storm-core/src/jvm/backtype/storm/utils/TransferDrainer.java
@@ -20,6 +20,7 @@ package backtype.storm.utils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.Map;
import backtype.storm.messaging.IConnection;
import backtype.storm.messaging.TaskMessage;
@@ -30,18 +31,19 @@ public class TransferDrainer {
private HashMap<Integer, ArrayList<ArrayList<TaskMessage>>> bundles = new HashMap();
public void add(HashMap<Integer, ArrayList<TaskMessage>> taskTupleSetMap) {
- for (Integer task : taskTupleSetMap.keySet()) {
- addListRefToMap(this.bundles, task, taskTupleSetMap.get(task));
+ for (Map.Entry<Integer, ArrayList<TaskMessage>> entry : taskTupleSetMap.entrySet()) {
+ addListRefToMap(this.bundles, entry.getKey(), entry.getValue());
}
}
public void send(HashMap<Integer, String> taskToNode, HashMap<String, IConnection> connections) {
HashMap<String, ArrayList<ArrayList<TaskMessage>>> bundleMapByDestination = groupBundleByDestination(taskToNode);
- for (String hostPort : bundleMapByDestination.keySet()) {
+ for (Map.Entry<String, ArrayList<ArrayList<TaskMessage>>> entry : bundleMapByDestination.entrySet()) {
+ String hostPort = entry.getKey();
IConnection connection = connections.get(hostPort);
if (null != connection) {
- ArrayList<ArrayList<TaskMessage>> bundle = bundleMapByDestination.get(hostPort);
+ ArrayList<ArrayList<TaskMessage>> bundle = entry.getValue();
Iterator<TaskMessage> iter = getBundleIterator(bundle);
if (null != iter && iter.hasNext()) {
connection.send(iter);
@@ -98,15 +100,12 @@ public class TransferDrainer {
@Override
public boolean hasNext() {
- if (offset < size) {
- return true;
- }
- return false;
+ return offset < size;
}
@Override
public TaskMessage next() {
- TaskMessage msg = null;
+ TaskMessage msg;
if (iter.hasNext()) {
msg = iter.next();
} else {
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/backtype/storm/utils/Utils.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/Utils.java b/storm-core/src/jvm/backtype/storm/utils/Utils.java
index 8660739..8c100e1 100644
--- a/storm-core/src/jvm/backtype/storm/utils/Utils.java
+++ b/storm-core/src/jvm/backtype/storm/utils/Utils.java
@@ -186,14 +186,14 @@ public class Utils {
public static <T> String join(Iterable<T> coll, String sep) {
Iterator<T> it = coll.iterator();
- String ret = "";
+ StringBuilder ret = new StringBuilder();
while(it.hasNext()) {
- ret = ret + it.next();
+ ret.append(it.next());
if(it.hasNext()) {
- ret = ret + sep;
+ ret.append(sep);
}
}
- return ret;
+ return ret.toString();
}
public static void sleep(long millis) {
@@ -327,10 +327,9 @@ public class Utils {
private static Object normalizeConf(Object conf) {
if (conf == null) return new HashMap();
if (conf instanceof Map) {
- Map confMap = new HashMap((Map) conf);
- for (Object key : confMap.keySet()) {
- Object val = confMap.get(key);
- confMap.put(key, normalizeConf(val));
+ Map<Object, Object> confMap = new HashMap((Map) conf);
+ for (Map.Entry<Object, Object> entry : confMap.entrySet()) {
+ confMap.put(entry.getKey(), normalizeConf(entry.getValue()));
}
return confMap;
} else if (conf instanceof List) {
@@ -426,8 +425,8 @@ public class Utils {
public static <K, V> Map<V, K> reverseMap(Map<K, V> map) {
Map<V, K> ret = new HashMap<V, K>();
- for (K key : map.keySet()) {
- ret.put(map.get(key), key);
+ for (Map.Entry<K, V> entry : map.entrySet()) {
+ ret.put(entry.getValue(), entry.getKey());
}
return ret;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java b/storm-core/src/jvm/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
index 9cd98ee..cb19b7c 100644
--- a/storm-core/src/jvm/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
+++ b/storm-core/src/jvm/storm/trident/spout/OpaquePartitionedTridentSpoutExecutor.java
@@ -119,8 +119,9 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
prevCached = new HashMap<>();
}
- for(String id: _partitionStates.keySet()) {
- EmitterPartitionState s = _partitionStates.get(id);
+ for(Entry<String, EmitterPartitionState> e: _partitionStates.entrySet()) {
+ String id = e.getKey();
+ EmitterPartitionState s = e.getValue();
s.rotatingState.removeState(tx.getTransactionId());
Object lastMeta = prevCached.get(id);
if(lastMeta==null) lastMeta = s.rotatingState.getLastState();
@@ -162,9 +163,8 @@ public class OpaquePartitionedTridentSpoutExecutor implements ICommitterTridentS
Long txid = attempt.getTransactionId();
Map<String, Object> metas = _cachedMetas.remove(txid);
- for(String partitionId: metas.keySet()) {
- Object meta = metas.get(partitionId);
- _partitionStates.get(partitionId).rotatingState.overrideState(txid, meta);
+ for(Entry<String, Object> entry: metas.entrySet()) {
+ _partitionStates.get(entry.getKey()).rotatingState.overrideState(txid, entry.getValue());
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/bd0e8a01/storm-core/src/jvm/storm/trident/topology/TridentTopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/storm/trident/topology/TridentTopologyBuilder.java b/storm-core/src/jvm/storm/trident/topology/TridentTopologyBuilder.java
index 226a11a..15b5296 100644
--- a/storm-core/src/jvm/storm/trident/topology/TridentTopologyBuilder.java
+++ b/storm-core/src/jvm/storm/trident/topology/TridentTopologyBuilder.java
@@ -102,7 +102,7 @@ public class TridentTopologyBuilder {
}
Map<GlobalStreamId, String> fleshOutStreamBatchIds(boolean includeCommitStream) {
- Map<GlobalStreamId, String> ret = new HashMap<GlobalStreamId, String>(_batchIds);
+ Map<GlobalStreamId, String> ret = new HashMap<>(_batchIds);
Set<String> allBatches = new HashSet(_batchIds.values());
for(String b: allBatches) {
ret.put(new GlobalStreamId(masterCoordinator(b), MasterBatchCoordinator.BATCH_STREAM_ID), b);
@@ -134,8 +134,8 @@ public class TridentTopologyBuilder {
Map<GlobalStreamId, String> batchIdsForSpouts = fleshOutStreamBatchIds(false);
Map<GlobalStreamId, String> batchIdsForBolts = fleshOutStreamBatchIds(true);
- Map<String, List<String>> batchesToCommitIds = new HashMap<String, List<String>>();
- Map<String, List<ITridentSpout>> batchesToSpouts = new HashMap<String, List<ITridentSpout>>();
+ Map<String, List<String>> batchesToCommitIds = new HashMap<>();
+ Map<String, List<ITridentSpout>> batchesToSpouts = new HashMap<>();
for(String id: _spouts.keySet()) {
TransactionalSpoutComponent c = _spouts.get(id);
@@ -196,8 +196,9 @@ public class TridentTopologyBuilder {
}
}
- for(String batch: batchesToCommitIds.keySet()) {
- List<String> commitIds = batchesToCommitIds.get(batch);
+ for(Map.Entry<String, List<String>> entry: batchesToCommitIds.entrySet()) {
+ String batch = entry.getKey();
+ List<String> commitIds = entry.getValue();
builder.setSpout(masterCoordinator(batch), new MasterBatchCoordinator(commitIds, batchesToSpouts.get(batch)));
}
@@ -233,9 +234,9 @@ public class TridentTopologyBuilder {
}
Map<String, Set<String>> batchToComponents = getBoltBatchToComponentSubscriptions(id);
- for(String b: batchToComponents.keySet()) {
- for(String comp: batchToComponents.get(b)) {
- d.directGrouping(comp, TridentBoltExecutor.COORD_STREAM(b));
+ for(Map.Entry<String, Set<String>> entry: batchToComponents.entrySet()) {
+ for(String comp: entry.getValue()) {
+ d.directGrouping(comp, TridentBoltExecutor.COORD_STREAM(entry.getKey()));
}
}
@@ -248,8 +249,8 @@ public class TridentTopologyBuilder {
}
private void markBatchGroups(String component, Map<String, String> batchGroups) {
- for(String stream: batchGroups.keySet()) {
- _batchIds.put(new GlobalStreamId(component, stream), batchGroups.get(stream));
+ for(Map.Entry<String, String> entry: batchGroups.entrySet()) {
+ _batchIds.put(new GlobalStreamId(component, entry.getKey()), entry.getValue());
}
}
@@ -291,7 +292,7 @@ public class TridentTopologyBuilder {
private static class Component {
public ITridentBatchBolt bolt;
public Integer parallelism;
- public List<InputDeclaration> declarations = new ArrayList<InputDeclaration>();
+ public List<InputDeclaration> declarations = new ArrayList<>();
public List<Map<String, Object>> componentConfs = new ArrayList<>();
public Set<String> committerBatches;
[3/3] storm git commit: STORM-1152. Change map keySet iteration to
entrySet iteration for efficiency.
Posted by sr...@apache.org.
STORM-1152. Change map keySet iteration to entrySet iteration for efficiency.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/72fc3beb
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/72fc3beb
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/72fc3beb
Branch: refs/heads/master
Commit: 72fc3beb45000c56af7e1863e95f6c8bc8ec9f75
Parents: fe08a9f
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sat Oct 31 10:56:54 2015 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sat Oct 31 10:56:54 2015 -0700
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/72fc3beb/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d641080..f259073 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
## 0.11.0
+ * STORM-1152: Change map keySet iteration to entrySet iteration for efficiency.
* STORM-1147: Storm JDBCBolt should add validation to ensure either insertQuery or table name is specified and not both.
* STORM-1151: Batching in DisruptorQueue
* STORM-350: Update disruptor to latest version (3.3.2)
[2/3] storm git commit: Merge branch 'STORM-1152' of
https://github.com/sureshms/incubator-storm into STORM-1152
Posted by sr...@apache.org.
Merge branch 'STORM-1152' of https://github.com/sureshms/incubator-storm into STORM-1152
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/fe08a9f0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/fe08a9f0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/fe08a9f0
Branch: refs/heads/master
Commit: fe08a9f0bc5505e16fcc768c2421b8e54b1db5f3
Parents: 3a724ee bd0e8a0
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sat Oct 31 10:51:06 2015 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sat Oct 31 10:51:06 2015 -0700
----------------------------------------------------------------------
.../jvm/backtype/storm/scheduler/Cluster.java | 64 ++++++--------------
.../backtype/storm/scheduler/Topologies.java | 6 +-
.../authorizer/DRPCSimpleACLAuthorizer.java | 8 +--
.../authorizer/ImpersonationAuthorizer.java | 7 ++-
.../serialization/SerializationFactory.java | 6 +-
.../storm/task/GeneralTopologyContext.java | 5 +-
...uePartitionedTransactionalSpoutExecutor.java | 5 +-
.../backtype/storm/utils/TransferDrainer.java | 17 +++---
.../src/jvm/backtype/storm/utils/Utils.java | 19 +++---
.../OpaquePartitionedTridentSpoutExecutor.java | 10 +--
.../topology/TridentTopologyBuilder.java | 23 +++----
11 files changed, 74 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/fe08a9f0/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --cc storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
index d676feb,f604013..b07576f
--- a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
@@@ -72,10 -67,10 +72,11 @@@ public class Cluster
this.assignments = new HashMap<String, SchedulerAssignmentImpl>(assignments.size());
this.assignments.putAll(assignments);
this.status = new HashMap<String, String>();
+ this.resources = new HashMap<String, Double[]>();
this.hostToId = new HashMap<String, List<String>>();
- for (String nodeId : supervisors.keySet()) {
- SupervisorDetails supervisor = supervisors.get(nodeId);
+ for (Map.Entry<String, SupervisorDetails> entry : supervisors.entrySet()) {
+ String nodeId = entry.getKey();
+ SupervisorDetails supervisor = entry.getValue();
String host = supervisor.getHost();
if (!this.hostToId.containsKey(host)) {
this.hostToId.put(host, new ArrayList<String>());