(refactor) some opportunities to use lambda functions

storm-core/src/jvm/backtype/storm/drpc/DRPCSpout.java

@@ -79,9 +79,9 @@ public void open(Map conf, TopologyContext context, SpoutOutputCollector collect
                 throw new RuntimeException("No DRPC servers configured for topology");   
             }
             if(numTasks < servers.size()) {
-                for(String s: servers) {
+                servers.forEach(s -> {
                     _clients.add(new DRPCInvocationsClient(s, port));
-                }
+                });
             } else {
                 int i = index % servers.size();
                 _clients.add(new DRPCInvocationsClient(servers.get(i), port));
@@ -92,9 +92,9 @@ public void open(Map conf, TopologyContext context, SpoutOutputCollector collect
 
     @Override
     public void close() {
-        for(DRPCInvocationsClient client: _clients) {
+        _clients.forEach(client -> {
             client.close();
-        }
+        });
     }
 
     @Override

storm-core/src/jvm/backtype/storm/generated/ClusterSummary.java

@@ -537,10 +537,9 @@ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache
       oprot.writeFieldBegin(TOPOLOGIES_FIELD_DESC);
       {
         oprot.writeListBegin(new org.apache.thrift7.protocol.TList(org.apache.thrift7.protocol.TType.STRUCT, this.topologies.size()));
-        for (TopologySummary _iter44 : this.topologies)
-        {
+        this.topologies.forEach(_iter44 -> {
           _iter44.write(oprot);
-        }
+        });
         oprot.writeListEnd();
       }
       oprot.writeFieldEnd();

storm-core/src/jvm/backtype/storm/generated/Grouping.java

@@ -379,10 +379,9 @@ protected void writeValue(org.apache.thrift7.protocol.TProtocol oprot) throws or
         List<String> fields = (List<String>)value_;
         {
           oprot.writeListBegin(new org.apache.thrift7.protocol.TList(org.apache.thrift7.protocol.TType.STRING, fields.size()));
-          for (String _iter7 : fields)
-          {
+          fields.forEach(_iter7 -> {
             oprot.writeString(_iter7);
-          }
+          });
           oprot.writeListEnd();
         }
         return;

storm-core/src/jvm/backtype/storm/generated/JavaObject.java

@@ -412,10 +412,9 @@ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache
       oprot.writeFieldBegin(ARGS_LIST_FIELD_DESC);
       {
         oprot.writeListBegin(new org.apache.thrift7.protocol.TList(org.apache.thrift7.protocol.TType.STRUCT, this.args_list.size()));
-        for (JavaObjectArg _iter3 : this.args_list)
-        {
+        this.args_list.forEach(_iter3 -> {
           _iter3.write(oprot);
-        }
+        });
         oprot.writeListEnd();
       }
       oprot.writeFieldEnd();

storm-core/src/jvm/backtype/storm/generated/StreamInfo.java

@@ -410,10 +410,9 @@ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache
       oprot.writeFieldBegin(OUTPUT_FIELDS_FIELD_DESC);
       {
         oprot.writeListBegin(new org.apache.thrift7.protocol.TList(org.apache.thrift7.protocol.TType.STRING, this.output_fields.size()));
-        for (String _iter11 : this.output_fields)
-        {
+        this.output_fields.forEach(_iter11 -> {
           oprot.writeString(_iter11);
-        }
+        });
         oprot.writeListEnd();
       }
       oprot.writeFieldEnd();

storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java

@@ -796,10 +796,9 @@ public void write(org.apache.thrift7.protocol.TProtocol oprot) throws org.apache
       oprot.writeFieldBegin(EXECUTORS_FIELD_DESC);
       {
         oprot.writeListBegin(new org.apache.thrift7.protocol.TList(org.apache.thrift7.protocol.TType.STRUCT, this.executors.size()));
-        for (ExecutorSummary _iter155 : this.executors)
-        {
+        this.executors.forEach(_iter155 -> {
           _iter155.write(oprot);
-        }
+        });
         oprot.writeListEnd();
       }
       oprot.writeFieldEnd();

storm-core/src/jvm/backtype/storm/scheduler/Cluster.java

@@ -218,24 +218,24 @@ public List<WorkerSlot> getAvailableSlots(SupervisorDetails supervisor) {
         Set<Integer> ports = this.getAvailablePorts(supervisor);
         List<WorkerSlot> slots = new ArrayList<WorkerSlot>(ports.size());
 
-        for (Integer port : ports) {
+        ports.forEach(port -> {
             slots.add(new WorkerSlot(supervisor.getId(), port));
-        }
+        });
 
         return slots;
     }
-    
+
     public List<WorkerSlot> getAssignableSlots(SupervisorDetails supervisor) {
         Set<Integer> ports = this.getAssignablePorts(supervisor);
         List<WorkerSlot> slots = new ArrayList<WorkerSlot>(ports.size());
 
-        for (Integer port : ports) {
+        ports.forEach(port -> {
             slots.add(new WorkerSlot(supervisor.getId(), port));
-        }
+        });
 
         return slots;        
     }
-    
+
     /**
      * get the unassigned executors of the topology.
      */
@@ -342,9 +342,9 @@ public void freeSlot(WorkerSlot slot) {
      */
     public void freeSlots(Collection<WorkerSlot> slots) {
         if(slots!=null) {
-            for (WorkerSlot slot : slots) {
+            slots.forEach(slot -> {
                 this.freeSlot(slot);
-            }
+            });
         }
     }
 
@@ -405,9 +405,9 @@ public List<SupervisorDetails> getSupervisorsByHost(String host) {
         List<SupervisorDetails> ret = new ArrayList<SupervisorDetails>();
 
         if (nodeIds != null) {
-            for (String nodeId : nodeIds) {
+            nodeIds.forEach(nodeId -> {
                 ret.add(this.getSupervisorById(nodeId));
-            }
+            });
         }
 
         return ret;

storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java

@@ -55,11 +55,11 @@ public Set<WorkerSlot> getSlots() {
      * @param executors
      */
     public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors) {
-        for (ExecutorDetails executor : executors) {
+        executors.forEach(executor -> {
             this.executorToSlot.put(executor, slot);
-        }
+        });
     }
-    
+
     /**
      * Release the slot occupied by this assignment.
      * @param slot
@@ -74,9 +74,9 @@ public void unassignBySlot(WorkerSlot slot) {
         }
 
         // remove
-        for (ExecutorDetails executor : executors) {
+        executors.forEach(executor -> {
             this.executorToSlot.remove(executor);
-        }
+        });
     }
 
     /**
@@ -107,4 +107,4 @@ public Map<ExecutorDetails, WorkerSlot> getExecutorToSlot() {
     public Set<ExecutorDetails> getExecutors() {
         return this.executorToSlot.keySet();
     }
-}
+}

storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java

@@ -61,9 +61,9 @@ public SupervisorDetails(String id, String host, Object schedulerMeta, Collectio
     private void setAllPorts(Collection<Number> allPorts) {
         this.allPorts = new HashSet<Integer>();
         if(allPorts!=null) {
-            for(Number n: allPorts) {
+            allPorts.forEach(n -> {
                 this.allPorts.add(n.intValue());
-            }
+            });
         }
     }
 

storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java

@@ -73,16 +73,16 @@ public Map<ExecutorDetails, String> getExecutorToComponent() {
 
     public Map<ExecutorDetails, String> selectExecutorToComponent(Collection<ExecutorDetails> executors) {
         Map<ExecutorDetails, String> ret = new HashMap<ExecutorDetails, String>(executors.size());
-        for (ExecutorDetails executor : executors) {
+        executors.forEach(executor -> {
             String compId = this.executorToComponent.get(executor);
             if (compId != null) {
                 ret.put(executor, compId);
             }
-        }
+        });
 
         return ret;
     }
-    
+
     public Collection<ExecutorDetails> getExecutors() {
         return this.executorToComponent.keySet();
     }

storm-core/src/jvm/backtype/storm/tuple/Fields.java

@@ -47,9 +47,9 @@ public Fields(List<String> fields) {
 
     public List<Object> select(Fields selector, List<Object> tuple) {
         List<Object> ret = new ArrayList<Object>(selector.size());
-        for(String s: selector) {
+        selector.forEach(s -> {
             ret.add(tuple.get(_index.get(s)));
-        }
+        });
         return ret;
     }
 

storm-core/src/jvm/backtype/storm/utils/RotatingMap.java

@@ -116,9 +116,7 @@ public Object remove(K key) {
 
     public int size() {
         int size = 0;
-        for(HashMap<K, V> bucket: _buckets) {
-            size+=bucket.size();
-        }
+        _buckets.stream().map(bucket -> bucket.size()).reduce(size, Integer::sum);
         return size;
-    }    
+    }
 }

storm-core/src/jvm/backtype/storm/utils/TimeCacheMap.java

@@ -148,9 +148,7 @@ public Object remove(K key) {
     public int size() {
         synchronized(_lock) {
             int size = 0;
-            for(HashMap<K, V> bucket: _buckets) {
-                size+=bucket.size();
-            }
+            _buckets.stream().map(bucket -> bucket.size()).reduce(size, Integer::sum);
             return size;
         }
     }

storm-core/src/jvm/storm/trident/Stream.java

@@ -365,10 +365,8 @@ private void projectionValidation(Fields projFields) {
         }
 
         Fields allFields = this.getOutputFields();
-        for (String field : projFields) {
-            if (!allFields.contains(field)) {
-                throw new IllegalArgumentException("Trying to select non-existent field: '" + field + "' from stream containing fields fields: <" + allFields + ">");
-            }
-        }
+        projFields.stream().filter(field -> !allFields.contains(field)).forEach(field -> {
+            throw new IllegalArgumentException("Trying to select non-existent field: '" + field + "' from stream containing fields fields: <" + allFields + ">");
+        });
     }
 }

storm-core/src/jvm/storm/trident/spout/RichSpoutBatchTriggerer.java

@@ -150,15 +150,17 @@ public List<Integer> emit(String ignore, List<Object> values, Object msgId) {
             finish.msgId = msgId;
             List<Integer> tasks = _collector.emit(_stream, new ConsList(batchId, values));
             Set<Integer> outTasksSet = new HashSet<Integer>(tasks);
-            for(Integer t: _outputTasks) {
+            _outputTasks.stream().map(t -> {
                 int count = 0;
-                if(outTasksSet.contains(t)) {
+                if (outTasksSet.contains(t)) {
                     count = 1;
                 }
                 long r = _rand.nextLong();
                 _collector.emitDirect(t, _coordStream, new Values(batchId, count), r);
+                return r;
+            }).forEach(r -> {
                 finish.vals.add(r);
-            }
+            });
             _finishConditions.put(batchIdVal, finish);
             return tasks;
         }

storm-core/src/jvm/storm/trident/state/map/CachedMap.java

@@ -58,9 +58,9 @@ public List<T> multiGet(List<List<Object>> keys) {
         }
 
         List<T> ret = new ArrayList<T>(keys.size());
-        for(List<Object> key: keys) {
+        keys.forEach(key -> {
             ret.add(results.get(key));
-        }
+        });
         return ret;
     }
 

storm-core/src/jvm/storm/trident/state/map/MapCombinerAggStateUpdater.java

@@ -57,10 +57,12 @@ public void updateState(MapState map, List<TridentTuple> tuples, TridentCollecto
         List<List<Object>> groups = new ArrayList<List<Object>>(tuples.size());
         List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(tuples.size());
 
-        for(TridentTuple t: tuples) {
+        tuples.stream().map(t -> {
             groups.add(_groupFactory.create(t));
+            return t;
+        }).forEach(t -> {
             updaters.add(new CombinerValueUpdater(_agg,_inputFactory.create(t).getValue(0)));
-        }
+        });
         List<Object> newVals = map.multiUpdate(groups, updaters);
 
         for(int i=0; i<tuples.size(); i++) {

storm-core/src/jvm/storm/trident/state/map/MapReducerAggStateUpdater.java

@@ -67,9 +67,9 @@ public void updateState(MapState map, List<TridentTuple> tuples, TridentCollecto
         }
         List<List<Object>> uniqueGroups = new ArrayList(grouped.keySet());
         List<ValueUpdater> updaters = new ArrayList(uniqueGroups.size());
-        for(List<Object> group: uniqueGroups) {
+        uniqueGroups.forEach(group -> {
             updaters.add(new ReducerValueUpdater(_agg, grouped.get(group)));
-        }
+        });
         List<Object> results = map.multiUpdate(uniqueGroups, updaters);
 
         for(int i=0; i<uniqueGroups.size(); i++) {

storm-core/src/jvm/storm/trident/state/map/OpaqueMap.java

@@ -87,9 +87,9 @@ public List<T> multiUpdate(List<List<Object>> keys, List<ValueUpdater> updaters)
     @Override
     public void multiPut(List<List<Object>> keys, List<T> vals) {
         List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(vals.size());
-        for(T val: vals) {
+        vals.forEach(val -> {
             updaters.add(new ReplaceUpdater<T>(val));
-        }
+        });
         multiUpdate(keys, updaters);
     }
 

storm-core/src/jvm/storm/trident/testing/FeederBatchSpout.java

@@ -140,9 +140,9 @@ public FeederEmitter(int index) {
         public void emitBatch(TransactionAttempt tx, Map<Integer, List<List<Object>>> coordinatorMeta, TridentCollector collector) {
             List<List<Object>> tuples = coordinatorMeta.get(_index);
             if(tuples!=null) {
-                for(List<Object> t: tuples) {
+                tuples.forEach(t -> {
                     collector.emit(t);
-                }
+                });
             }
         }
 

storm-core/src/jvm/storm/trident/testing/MemoryBackingMap.java

@@ -30,9 +30,9 @@ public class MemoryBackingMap implements IBackingMap<Object> {
     @Override
     public List<Object> multiGet(List<List<Object>> keys) {
         List ret = new ArrayList();
-        for(List key: keys) {
+        keys.forEach(key -> {
             ret.add(_vals.get(key));
-        }
+        });
         return ret;
     }
 

storm-core/src/jvm/storm/trident/testing/MemoryMapState.java

@@ -107,15 +107,13 @@ public MemoryMapStateBacking(String id) {
         }
 
         @Override
-        public List<T> multiGet(List<List<Object>> keys) {
+        public List<T> multiGet(List<List<Object>> keys)  {
             List<T> ret = new ArrayList();
-            for (List<Object> key : keys) {
-                ret.add(db.get(key));
-            }
+            keys.forEach(key -> {
+ret.add(db.get(key));
+});
             return ret;
-        }
-
-        @Override
+        }@Override
         public void multiPut(List<List<Object>> keys, List<T> vals) {
             for (int i = 0; i < keys.size(); i++) {
                 List<Object> key = keys.get(i);

storm-core/src/jvm/storm/trident/testing/TuplifyArgs.java

@@ -29,9 +29,8 @@ public class TuplifyArgs extends BaseFunction {
     public void execute(TridentTuple input, TridentCollector collector) {
         String args = input.getString(0);
         List<List<Object>> tuples = (List) JSONValue.parse(args);
-        for(List<Object> tuple: tuples) {
+        tuples.forEach(tuple -> {
             collector.emit(tuple);
-        }
+        });
     }
-
 }

storm-netty/src/jvm/backtype/storm/messaging/netty/Context.java

@@ -59,9 +59,9 @@ public IConnection connect(String storm_id, String host, int port) {
      * terminate this context
      */
     public void term() {
-        for (IConnection conn : connections) {
+        connections.forEach(conn -> {
             conn.close();
-        }
+        });
         connections = null;
     }
 }