Repository: storm Updated Branches: refs/heads/master 5a4e1f85b -> e25f28f71
http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/KeyedRoundRobinQueue.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/KeyedRoundRobinQueue.java b/storm-core/src/jvm/backtype/storm/utils/KeyedRoundRobinQueue.java index 3cb455d..4ecdda6 100644 --- a/storm-core/src/jvm/backtype/storm/utils/KeyedRoundRobinQueue.java +++ b/storm-core/src/jvm/backtype/storm/utils/KeyedRoundRobinQueue.java @@ -28,15 +28,15 @@ import java.util.concurrent.Semaphore; public class KeyedRoundRobinQueue<V> { private final Object _lock = new Object(); private Semaphore _size = new Semaphore(0); - private Map<Object, Queue<V>> _queues = new HashMap<Object, Queue<V>>(); - private List<Object> _keyOrder = new ArrayList<Object>(); + private Map<Object, Queue<V>> _queues = new HashMap<>(); + private List<Object> _keyOrder = new ArrayList<>(); private int _currIndex = 0; public void add(Object key, V val) { synchronized(_lock) { Queue<V> queue = _queues.get(key); if(queue==null) { - queue = new LinkedList<V>(); + queue = new LinkedList<>(); _queues.put(key, queue); _keyOrder.add(key); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/ListDelegate.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/ListDelegate.java b/storm-core/src/jvm/backtype/storm/utils/ListDelegate.java index 1e091f0..d198a72 100644 --- a/storm-core/src/jvm/backtype/storm/utils/ListDelegate.java +++ b/storm-core/src/jvm/backtype/storm/utils/ListDelegate.java @@ -27,7 +27,7 @@ public class ListDelegate implements List<Object> { private List<Object> _delegate; public ListDelegate() { - _delegate = new ArrayList<Object>(); + _delegate = new ArrayList<>(); } public void setDelegate(List<Object> delegate) { @@ -84,12 +84,12 @@ public class ListDelegate implements List<Object> { } @Override - public boolean addAll(Collection<? extends Object> clctn) { + public boolean addAll(Collection<?> clctn) { return _delegate.addAll(clctn); } @Override - public boolean addAll(int i, Collection<? extends Object> clctn) { + public boolean addAll(int i, Collection<?> clctn) { return _delegate.addAll(i, clctn); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/Monitor.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/Monitor.java b/storm-core/src/jvm/backtype/storm/utils/Monitor.java index 36fedc4..41717a9 100644 --- a/storm-core/src/jvm/backtype/storm/utils/Monitor.java +++ b/storm-core/src/jvm/backtype/storm/utils/Monitor.java @@ -20,7 +20,6 @@ package backtype.storm.utils; import backtype.storm.generated.*; import java.util.HashSet; -import java.util.List; import java.util.Map; public class Monitor { @@ -98,7 +97,7 @@ public class Monitor { } private HashSet<String> getComponents(Nimbus.Client client, String topology) throws Exception{ - HashSet<String> components = new HashSet<String>(); + HashSet<String> components = new HashSet<>(); ClusterSummary clusterSummary = client.getClusterInfo(); TopologySummary topologySummary = null; for (TopologySummary ts: clusterSummary.get_topologies()) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java b/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java index a715de2..a9306f7 100644 --- a/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java +++ b/storm-core/src/jvm/backtype/storm/utils/NimbusClient.java @@ -22,13 +22,8 @@ import backtype.storm.Config; import backtype.storm.generated.ClusterSummary; import backtype.storm.generated.Nimbus; import backtype.storm.generated.NimbusSummary; -import backtype.storm.nimbus.ILeaderElector; -import backtype.storm.nimbus.NimbusInfo; import backtype.storm.security.auth.ThriftClient; import backtype.storm.security.auth.ThriftConnectionType; -import clojure.lang.IFn; -import clojure.lang.PersistentArrayMap; -import com.google.common.base.Splitter; import com.google.common.collect.Lists; import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; @@ -38,7 +33,6 @@ import java.util.List; import java.util.Map; public class NimbusClient extends ThriftClient { - public static final String DELIMITER = ":"; private Nimbus.Client _client; private static final Logger LOG = LoggerFactory.getLogger(NimbusClient.class); @@ -56,7 +50,7 @@ public class NimbusClient extends ThriftClient { asUser = (String) conf.get(Config.STORM_DO_AS_USER); } - List<String> seeds = null; + List<String> seeds; if(conf.containsKey(Config.NIMBUS_HOST)) { LOG.warn("Using deprecated config {} for backward compatibility. Please update your storm.yaml so it only has config {}", Config.NIMBUS_HOST, Config.NIMBUS_SEEDS); @@ -67,7 +61,7 @@ public class NimbusClient extends ThriftClient { for (String host : seeds) { int port = Integer.parseInt(conf.get(Config.NIMBUS_THRIFT_PORT).toString()); - ClusterSummary clusterInfo = null; + ClusterSummary clusterInfo; try { NimbusClient client = new NimbusClient(conf, host, port); clusterInfo = client.getClient().getClusterInfo(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/RegisteredGlobalState.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/RegisteredGlobalState.java b/storm-core/src/jvm/backtype/storm/utils/RegisteredGlobalState.java index 48053fc..159a7c1 100644 --- a/storm-core/src/jvm/backtype/storm/utils/RegisteredGlobalState.java +++ b/storm-core/src/jvm/backtype/storm/utils/RegisteredGlobalState.java @@ -27,7 +27,7 @@ import java.util.UUID; * tuples. */ public class RegisteredGlobalState { - private static HashMap<String, Object> _states = new HashMap<String, Object>(); + private static HashMap<String, Object> _states = new HashMap<>(); private static final Object _lock = new Object(); public static Object globalLock() { @@ -50,9 +50,7 @@ public class RegisteredGlobalState { public static Object getState(String id) { synchronized(_lock) { - Object ret = _states.get(id); - //System.out.println("State: " + ret.toString()); - return ret; + return _states.get(id); } } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/RotatingMap.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/RotatingMap.java b/storm-core/src/jvm/backtype/storm/utils/RotatingMap.java index 2b8d66b..03d1bf8 100644 --- a/storm-core/src/jvm/backtype/storm/utils/RotatingMap.java +++ b/storm-core/src/jvm/backtype/storm/utils/RotatingMap.java @@ -53,7 +53,7 @@ public class RotatingMap<K, V> { if(numBuckets<2) { throw new IllegalArgumentException("numBuckets must be >= 2"); } - _buckets = new LinkedList<HashMap<K, V>>(); + _buckets = new LinkedList<>(); for(int i=0; i<numBuckets; i++) { _buckets.add(new HashMap<K, V>()); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/ServiceRegistry.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/ServiceRegistry.java b/storm-core/src/jvm/backtype/storm/utils/ServiceRegistry.java index 724bc3e..d5c2eca 100644 --- a/storm-core/src/jvm/backtype/storm/utils/ServiceRegistry.java +++ b/storm-core/src/jvm/backtype/storm/utils/ServiceRegistry.java @@ -22,7 +22,7 @@ import java.util.UUID; // this class should be combined with RegisteredGlobalState public class ServiceRegistry { - private static HashMap<String, Object> _services = new HashMap<String, Object>(); + private static HashMap<String, Object> _services = new HashMap<>(); private static final Object _lock = new Object(); public static String registerService(Object service) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/StormBoundedExponentialBackoffRetry.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/StormBoundedExponentialBackoffRetry.java b/storm-core/src/jvm/backtype/storm/utils/StormBoundedExponentialBackoffRetry.java index 5422c2e..be6f0e9 100644 --- a/storm-core/src/jvm/backtype/storm/utils/StormBoundedExponentialBackoffRetry.java +++ b/storm-core/src/jvm/backtype/storm/utils/StormBoundedExponentialBackoffRetry.java @@ -42,8 +42,9 @@ public class StormBoundedExponentialBackoffRetry extends BoundedExponentialBacko public StormBoundedExponentialBackoffRetry(int baseSleepTimeMs, int maxSleepTimeMs, int maxRetries) { super(baseSleepTimeMs, maxSleepTimeMs, maxRetries); expRetriesThreshold = 1; - while ((1 << (expRetriesThreshold + 1)) < ((maxSleepTimeMs - baseSleepTimeMs) / 2)) + while ((1 << (expRetriesThreshold + 1)) < ((maxSleepTimeMs - baseSleepTimeMs) / 2)) { expRetriesThreshold++; + } LOG.debug("The baseSleepTimeMs [{}] the maxSleepTimeMs [{}] the maxRetries [{}]", baseSleepTimeMs, maxSleepTimeMs, maxRetries); if (baseSleepTimeMs > maxSleepTimeMs) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/VersionInfo.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/VersionInfo.java b/storm-core/src/jvm/backtype/storm/utils/VersionInfo.java index 1740e18..f3586bb 100644 --- a/storm-core/src/jvm/backtype/storm/utils/VersionInfo.java +++ b/storm-core/src/jvm/backtype/storm/utils/VersionInfo.java @@ -86,7 +86,7 @@ public class VersionInfo { } - private static VersionInfo COMMON_VERSION_INFO = new VersionInfo("storm-core"); + private static final VersionInfo COMMON_VERSION_INFO = new VersionInfo("storm-core"); public static String getVersion() { return COMMON_VERSION_INFO._getVersion(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/TridentTopology.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/TridentTopology.java b/storm-core/src/jvm/storm/trident/TridentTopology.java index 8164717..58c83e4 100644 --- a/storm-core/src/jvm/storm/trident/TridentTopology.java +++ b/storm-core/src/jvm/storm/trident/TridentTopology.java @@ -87,16 +87,17 @@ import storm.trident.util.TridentUtils; // all operations have finishBatch and can optionally be committers public class TridentTopology { - //TODO: add a method for drpc stream, needs to know how to automatically do returnresults, etc + //TODO: add a method for drpc stream, needs to know how to automatically do return results, etc // is it too expensive to do a batch per drpc request? - DefaultDirectedGraph<Node, IndexedEdge> _graph; - Map<String, List<Node>> _colocate = new HashMap(); - UniqueIdGen _gen; + final DefaultDirectedGraph<Node, IndexedEdge> _graph; + final Map<String, List<Node>> _colocate; + final UniqueIdGen _gen; public TridentTopology() { - _graph = new DefaultDirectedGraph(new ErrorEdgeFactory()); - _gen = new UniqueIdGen(); + this(new DefaultDirectedGraph<Node, IndexedEdge>(new ErrorEdgeFactory()), + new HashMap<String, List<Node>>(), + new UniqueIdGen()); } private TridentTopology(DefaultDirectedGraph<Node, IndexedEdge> graph, Map<String, List<Node>> colocate, UniqueIdGen gen) { @@ -195,7 +196,7 @@ public class TridentTopology { } public Stream multiReduce(List<Fields> inputFields, List<Stream> streams, MultiReducer function, Fields outputFields) { - List<String> names = new ArrayList<String>(); + List<String> names = new ArrayList<>(); for(Stream s: streams) { if(s._name!=null) { names.add(s._name); @@ -206,9 +207,9 @@ public class TridentTopology { } public Stream multiReduce(List<Fields> inputFields, List<GroupedStream> groupedStreams, GroupedMultiReducer function, Fields outputFields) { - List<Fields> fullInputFields = new ArrayList<Fields>(); - List<Stream> streams = new ArrayList<Stream>(); - List<Fields> fullGroupFields = new ArrayList<Fields>(); + List<Fields> fullInputFields = new ArrayList<>(); + List<Stream> streams = new ArrayList<>(); + List<Fields> fullGroupFields = new ArrayList<>(); for(int i=0; i<groupedStreams.size(); i++) { GroupedStream gs = groupedStreams.get(i); Fields groupFields = gs.getGroupFields(); @@ -270,10 +271,10 @@ public class TridentTopology { completeDRPC(graph, _colocate, _gen); - List<SpoutNode> spoutNodes = new ArrayList<SpoutNode>(); + List<SpoutNode> spoutNodes = new ArrayList<>(); // can be regular nodes (static state) or processor nodes - Set<Node> boltNodes = new HashSet<Node>(); + Set<Node> boltNodes = new HashSet<>(); for(Node n: graph.vertexSet()) { if(n instanceof SpoutNode) { spoutNodes.add((SpoutNode) n); @@ -283,7 +284,7 @@ public class TridentTopology { } - Set<Group> initialGroups = new HashSet<Group>(); + Set<Group> initialGroups = new HashSet<>(); for(List<Node> colocate: _colocate.values()) { Group g = new Group(graph, colocate); boltNodes.removeAll(colocate); @@ -301,7 +302,7 @@ public class TridentTopology { // add identity partitions between groups - for(IndexedEdge<Node> e: new HashSet<IndexedEdge>(graph.edgeSet())) { + for(IndexedEdge<Node> e: new HashSet<>(graph.edgeSet())) { if(!(e.source instanceof PartitionNode) && !(e.target instanceof PartitionNode)) { Group g1 = grouper.nodeGroup(e.source); Group g2 = grouper.nodeGroup(e.target); @@ -325,7 +326,7 @@ public class TridentTopology { // this is because can't currently merge splitting logic into a spout // not the most kosher algorithm here, since the grouper indexes are being trounced via the adding of nodes to random groups, but it // works out - List<Node> forNewGroups = new ArrayList<Node>(); + List<Node> forNewGroups = new ArrayList<>(); for(Group g: mergedGroups) { for(PartitionNode n: extraPartitionInputs(g)) { Node idNode = makeIdentityNode(n.allOutputFields); @@ -350,7 +351,7 @@ public class TridentTopology { } } // TODO: in the future, want a way to include this logic in the spout itself, - // or make it unecessary by having storm include metadata about which grouping a tuple + // or make it unnecessary by having storm include metadata about which grouping a tuple // came from for(Node n: forNewGroups) { @@ -366,8 +367,8 @@ public class TridentTopology { mergedGroups = grouper.getAllGroups(); - Map<Node, String> batchGroupMap = new HashMap(); - List<Set<Node>> connectedComponents = new ConnectivityInspector<Node, IndexedEdge>(graph).connectedSets(); + Map<Node, String> batchGroupMap = new HashMap<>(); + List<Set<Node>> connectedComponents = new ConnectivityInspector<>(graph).connectedSets(); for(int i=0; i<connectedComponents.size(); i++) { String groupId = "bg" + i; for(Node n: connectedComponents.get(i)) { @@ -413,14 +414,10 @@ public class TridentTopology { Collection<PartitionNode> inputs = uniquedSubscriptions(externalGroupInputs(g)); for(PartitionNode n: inputs) { Node parent = TridentUtils.getParent(graph, n); - String componentId; - if(parent instanceof SpoutNode) { - componentId = spoutIds.get(parent); - } else { - componentId = boltIds.get(grouper.nodeGroup(parent)); - } + String componentId = parent instanceof SpoutNode ? + spoutIds.get(parent) : boltIds.get(grouper.nodeGroup(parent)); d.grouping(new GlobalStreamId(componentId, n.streamId), n.thriftGrouping); - } + } } } @@ -428,7 +425,7 @@ public class TridentTopology { } private static void completeDRPC(DefaultDirectedGraph<Node, IndexedEdge> graph, Map<String, List<Node>> colocate, UniqueIdGen gen) { - List<Set<Node>> connectedComponents = new ConnectivityInspector<Node, IndexedEdge>(graph).connectedSets(); + List<Set<Node>> connectedComponents = new ConnectivityInspector<>(graph).connectedSets(); for(Set<Node> g: connectedComponents) { checkValidJoins(g); @@ -496,7 +493,7 @@ public class TridentTopology { } private static Collection<PartitionNode> uniquedSubscriptions(Set<PartitionNode> subscriptions) { - Map<String, PartitionNode> ret = new HashMap(); + Map<String, PartitionNode> ret = new HashMap<>(); for(PartitionNode n: subscriptions) { PartitionNode curr = ret.get(n.streamId); if(curr!=null && !curr.thriftGrouping.equals(n.thriftGrouping)) { @@ -508,7 +505,7 @@ public class TridentTopology { } private static Map<Node, String> genSpoutIds(Collection<SpoutNode> spoutNodes) { - Map<Node, String> ret = new HashMap(); + Map<Node, String> ret = new HashMap<>(); int ctr = 0; for(SpoutNode n: spoutNodes) { if (n.type == SpoutNode.SpoutType.BATCH) { // if Batch spout then id contains txId @@ -525,11 +522,11 @@ public class TridentTopology { } private static Map<Group, String> genBoltIds(Collection<Group> groups) { - Map<Group, String> ret = new HashMap(); + Map<Group, String> ret = new HashMap<>(); int ctr = 0; for(Group g: groups) { if(!isSpoutGroup(g)) { - List<String> name = new ArrayList(); + List<String> name = new ArrayList<>(); name.add("b"); name.add("" + ctr); String groupName = getGroupName(g); @@ -544,13 +541,13 @@ public class TridentTopology { } private static String getGroupName(Group g) { - TreeMap<Integer, String> sortedNames = new TreeMap(); + TreeMap<Integer, String> sortedNames = new TreeMap<>(); for(Node n: g.nodes) { if(n.name!=null) { sortedNames.put(n.creationIndex, n.name); } } - List<String> names = new ArrayList<String>(); + List<String> names = new ArrayList<>(); String prevName = null; for(String n: sortedNames.values()) { if(prevName==null || !n.equals(prevName)) { @@ -562,7 +559,7 @@ public class TridentTopology { } private static Map<String, String> getOutputStreamBatchGroups(Group g, Map<Node, String> batchGroupMap) { - Map<String, String> ret = new HashMap(); + Map<String, String> ret = new HashMap<>(); Set<PartitionNode> externalGroupOutputs = externalGroupOutputs(g); for(PartitionNode n: externalGroupOutputs) { ret.put(n.streamId, batchGroupMap.get(n)); @@ -571,7 +568,7 @@ public class TridentTopology { } private static Set<String> committerBatches(Group g, Map<Node, String> batchGroupMap) { - Set<String> ret = new HashSet(); + Set<String> ret = new HashSet<>(); for(Node n: g.nodes) { if(n instanceof ProcessorNode) { if(((ProcessorNode) n).committer) { @@ -583,7 +580,7 @@ public class TridentTopology { } private static Map<Group, Integer> getGroupParallelisms(DirectedGraph<Node, IndexedEdge> graph, GraphGrouper grouper, Collection<Group> groups) { - UndirectedGraph<Group, Object> equivs = new Pseudograph<Group, Object>(Object.class); + UndirectedGraph<Group, Object> equivs = new Pseudograph<>(Object.class); for(Group g: groups) { equivs.addVertex(g); } @@ -599,8 +596,8 @@ public class TridentTopology { } } - Map<Group, Integer> ret = new HashMap(); - List<Set<Group>> equivGroups = new ConnectivityInspector<Group, Object>(equivs).connectedSets(); + Map<Group, Integer> ret = new HashMap<>(); + List<Set<Group>> equivGroups = new ConnectivityInspector<>(equivs).connectedSets(); for(Set<Group> equivGroup: equivGroups) { Integer fixedP = getFixedParallelism(equivGroup); Integer maxP = getMaxParallelism(equivGroup); @@ -689,9 +686,9 @@ public class TridentTopology { } private static List<PartitionNode> extraPartitionInputs(Group g) { - List<PartitionNode> ret = new ArrayList(); + List<PartitionNode> ret = new ArrayList<>(); Set<PartitionNode> inputs = externalGroupInputs(g); - Map<String, List<PartitionNode>> grouped = new HashMap(); + Map<String, List<PartitionNode>> grouped = new HashMap<>(); for(PartitionNode n: inputs) { if(!grouped.containsKey(n.streamId)) { grouped.put(n.streamId, new ArrayList()); @@ -711,7 +708,7 @@ public class TridentTopology { } private static Set<PartitionNode> externalGroupInputs(Group g) { - Set<PartitionNode> ret = new HashSet(); + Set<PartitionNode> ret = new HashSet<>(); for(Node n: g.incomingNodes()) { if(n instanceof PartitionNode) { ret.add((PartitionNode) n); @@ -721,7 +718,7 @@ public class TridentTopology { } private static Set<PartitionNode> externalGroupOutputs(Group g) { - Set<PartitionNode> ret = new HashSet(); + Set<PartitionNode> ret = new HashSet<>(); for(Node n: g.outgoingNodes()) { if(n instanceof PartitionNode) { ret.add((PartitionNode) n); @@ -788,7 +785,7 @@ public class TridentTopology { } private static List<Fields> getAllOutputFields(List streams) { - List<Fields> ret = new ArrayList<Fields>(); + List<Fields> ret = new ArrayList<>(); for(Object o: streams) { ret.add(((IAggregatableStream) o).getOutputFields()); } @@ -797,7 +794,7 @@ public class TridentTopology { private static List<GroupedStream> groupedStreams(List<Stream> streams, List<Fields> joinFields) { - List<GroupedStream> ret = new ArrayList<GroupedStream>(); + List<GroupedStream> ret = new ArrayList<>(); for(int i=0; i<streams.size(); i++) { ret.add(streams.get(i).groupBy(joinFields.get(i))); } @@ -805,7 +802,7 @@ public class TridentTopology { } private static List<Fields> strippedInputFields(List<Stream> streams, List<Fields> joinFields) { - List<Fields> ret = new ArrayList<Fields>(); + List<Fields> ret = new ArrayList<>(); for(int i=0; i<streams.size(); i++) { ret.add(TridentUtils.fieldsSubtract(streams.get(i).getOutputFields(), joinFields.get(i))); } @@ -813,7 +810,7 @@ public class TridentTopology { } private static List<JoinType> repeat(int n, JoinType type) { - List<JoinType> ret = new ArrayList<JoinType>(); + List<JoinType> ret = new ArrayList<>(); for(int i=0; i<n; i++) { ret.add(type); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/drpc/ReturnResultsReducer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/drpc/ReturnResultsReducer.java b/storm-core/src/jvm/storm/trident/drpc/ReturnResultsReducer.java index 849fb10..515d3a2 100644 --- a/storm-core/src/jvm/storm/trident/drpc/ReturnResultsReducer.java +++ b/storm-core/src/jvm/storm/trident/drpc/ReturnResultsReducer.java @@ -40,7 +40,7 @@ import storm.trident.tuple.TridentTuple; public class ReturnResultsReducer implements MultiReducer<ReturnResultsState> { public static class ReturnResultsState { - List<TridentTuple> results = new ArrayList<TridentTuple>(); + List<TridentTuple> results = new ArrayList<>(); String returnInfo; @Override @@ -50,7 +50,7 @@ public class ReturnResultsReducer implements MultiReducer<ReturnResultsState> { } boolean local; Map conf; - Map<List, DRPCInvocationsClient> _clients = new HashMap<List, DRPCInvocationsClient>(); + Map<List, DRPCInvocationsClient> _clients = new HashMap<>(); @Override http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java b/storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java index 8040e8b..976f983 100644 --- a/storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java +++ b/storm-core/src/jvm/storm/trident/fluent/ChainedAggregatorDeclarer.java @@ -59,7 +59,7 @@ public class ChainedAggregatorDeclarer implements ChainedFullAggregatorDeclarer, } } - List<AggSpec> _aggs = new ArrayList<AggSpec>(); + List<AggSpec> _aggs = new ArrayList<>(); IAggregatableStream _stream; AggType _type = null; GlobalAggregationScheme _globalScheme; @@ -73,8 +73,8 @@ public class ChainedAggregatorDeclarer implements ChainedFullAggregatorDeclarer, Fields[] inputFields = new Fields[_aggs.size()]; Aggregator[] aggs = new Aggregator[_aggs.size()]; int[] outSizes = new int[_aggs.size()]; - List<String> allOutFields = new ArrayList<String>(); - Set<String> allInFields = new HashSet<String>(); + List<String> allOutFields = new ArrayList<>(); + Set<String> allInFields = new HashSet<>(); for(int i=0; i<_aggs.size(); i++) { AggSpec spec = _aggs.get(i); Fields infields = spec.inFields; @@ -92,7 +92,7 @@ public class ChainedAggregatorDeclarer implements ChainedFullAggregatorDeclarer, throw new IllegalArgumentException("Output fields for chained aggregators must be distinct: " + allOutFields.toString()); } - Fields inFields = new Fields(new ArrayList<String>(allInFields)); + Fields inFields = new Fields(new ArrayList<>(allInFields)); Fields outFields = new Fields(allOutFields); Aggregator combined = new ChainedAggregatorImpl(aggs, inputFields, new ComboList.Factory(outSizes)); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java b/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java index 94db077..6f34e8c 100644 --- a/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java +++ b/storm-core/src/jvm/storm/trident/graph/GraphGrouper.java @@ -28,14 +28,13 @@ import storm.trident.util.IndexedEdge; public class GraphGrouper { - - DirectedGraph<Node, IndexedEdge> graph; - Set<Group> currGroups; - Map<Node, Group> groupIndex = new HashMap(); + final DirectedGraph<Node, IndexedEdge> graph; + final Set<Group> currGroups; + final Map<Node, Group> groupIndex = new HashMap<>(); public GraphGrouper(DirectedGraph<Node, IndexedEdge> graph, Collection<Group> initialGroups) { this.graph = graph; - this.currGroups = new HashSet(initialGroups); + this.currGroups = new HashSet<>(initialGroups); reindex(); } @@ -95,7 +94,7 @@ public class GraphGrouper { } public Collection<Group> outgoingGroups(Group g) { - Set<Group> ret = new HashSet(); + Set<Group> ret = new HashSet<>(); for(Node n: g.outgoingNodes()) { Group other = nodeGroup(n); if(other==null || !other.equals(g)) { @@ -106,7 +105,7 @@ public class GraphGrouper { } public Collection<Group> incomingGroups(Group g) { - Set<Group> ret = new HashSet(); + Set<Group> ret = new HashSet<>(); for(Node n: g.incomingNodes()) { Group other = nodeGroup(n); if(other==null || !other.equals(g)) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/graph/Group.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/graph/Group.java b/storm-core/src/jvm/storm/trident/graph/Group.java index 8ed0023..64bdec6 100644 --- a/storm-core/src/jvm/storm/trident/graph/Group.java +++ b/storm-core/src/jvm/storm/trident/graph/Group.java @@ -29,14 +29,13 @@ import storm.trident.util.TridentUtils; public class Group { - public Set<Node> nodes = new HashSet<Node>(); - private DirectedGraph<Node, IndexedEdge> graph; - private String id; + public final Set<Node> nodes = new HashSet<>(); + private final DirectedGraph<Node, IndexedEdge> graph; + private final String id = UUID.randomUUID().toString(); public Group(DirectedGraph graph, List<Node> nodes) { - init(graph); - this.nodes.addAll(nodes); this.graph = graph; + this.nodes.addAll(nodes); } public Group(DirectedGraph graph, Node n) { @@ -44,18 +43,13 @@ public class Group { } public Group(Group g1, Group g2) { - init(g1.graph); + this.graph = g1.graph; nodes.addAll(g1.nodes); nodes.addAll(g2.nodes); } - private void init(DirectedGraph graph) { - this.graph = graph; - this.id = UUID.randomUUID().toString(); - } - public Set<Node> outgoingNodes() { - Set<Node> ret = new HashSet<Node>(); + Set<Node> ret = new HashSet<>(); for(Node n: nodes) { ret.addAll(TridentUtils.getChildren(graph, n)); } @@ -63,7 +57,7 @@ public class Group { } public Set<Node> incomingNodes() { - Set<Node> ret = new HashSet<Node>(); + Set<Node> ret = new HashSet<>(); for(Node n: nodes) { ret.addAll(TridentUtils.getParents(graph, n)); } @@ -77,6 +71,9 @@ public class Group { @Override public boolean equals(Object o) { + if (this == o) { + return true; + } return id.equals(((Group) o).id); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/operation/builtin/SnapshotGet.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/operation/builtin/SnapshotGet.java b/storm-core/src/jvm/storm/trident/operation/builtin/SnapshotGet.java index 6d24ae6..2a44eab 100644 --- a/storm-core/src/jvm/storm/trident/operation/builtin/SnapshotGet.java +++ b/storm-core/src/jvm/storm/trident/operation/builtin/SnapshotGet.java @@ -29,9 +29,9 @@ public class SnapshotGet extends BaseQueryFunction<ReadOnlySnapshottable, Object @Override public List<Object> batchRetrieve(ReadOnlySnapshottable state, List<TridentTuple> args) { - List<Object> ret = new ArrayList<Object>(args.size()); + List<Object> ret = new ArrayList<>(args.size()); Object snapshot = state.get(); - for(int i=0; i<args.size(); i++) { + for (TridentTuple arg : args) { ret.add(snapshot); } return ret; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/operation/builtin/TupleCollectionGet.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/operation/builtin/TupleCollectionGet.java b/storm-core/src/jvm/storm/trident/operation/builtin/TupleCollectionGet.java index 52dd633..426b56f 100644 --- a/storm-core/src/jvm/storm/trident/operation/builtin/TupleCollectionGet.java +++ b/storm-core/src/jvm/storm/trident/operation/builtin/TupleCollectionGet.java @@ -30,9 +30,9 @@ public class TupleCollectionGet extends BaseQueryFunction<State, Iterator<List<O @Override public List<Iterator<List<Object>>> batchRetrieve(State state, List<TridentTuple> args) { - List<Iterator<List<Object>>> ret = new ArrayList(args.size()); - for(int i=0; i<args.size(); i++) { - ret.add(((ITupleCollection)state).getTuples()); + List<Iterator<List<Object>>> ret = new ArrayList<>(args.size()); + for (TridentTuple arg : args) { + ret.add(((ITupleCollection) state).getTuples()); } return ret; } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/partition/GlobalGrouping.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/partition/GlobalGrouping.java b/storm-core/src/jvm/storm/trident/partition/GlobalGrouping.java index 54fa844..a535eb2 100644 --- a/storm-core/src/jvm/storm/trident/partition/GlobalGrouping.java +++ b/storm-core/src/jvm/storm/trident/partition/GlobalGrouping.java @@ -26,13 +26,11 @@ import java.util.Collections; import java.util.List; public class GlobalGrouping implements CustomStreamGrouping { - List<Integer> target; - @Override public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targets) { - List<Integer> sorted = new ArrayList<Integer>(targets); + List<Integer> sorted = new ArrayList<>(targets); Collections.sort(sorted); target = Arrays.asList(sorted.get(0)); } @@ -41,5 +39,4 @@ public class GlobalGrouping implements CustomStreamGrouping { public List<Integer> chooseTasks(int i, List<Object> list) { return target; } - } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java b/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java index 30f48ad..88c2922 100644 --- a/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java +++ b/storm-core/src/jvm/storm/trident/partition/IdentityGrouping.java @@ -29,18 +29,16 @@ import java.util.Map; public class IdentityGrouping implements CustomStreamGrouping { - - List<Integer> ret = new ArrayList<Integer>(); - Map<Integer, List<Integer>> _precomputed = new HashMap(); + final Map<Integer, List<Integer>> _precomputed = new HashMap<>(); @Override public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> tasks) { - List<Integer> sourceTasks = new ArrayList<Integer>(context.getComponentTasks(stream.get_componentId())); + List<Integer> sourceTasks = new ArrayList<>(context.getComponentTasks(stream.get_componentId())); Collections.sort(sourceTasks); if(sourceTasks.size()!=tasks.size()) { throw new RuntimeException("Can only do an identity grouping when source and target have same number of tasks"); } - tasks = new ArrayList<Integer>(tasks); + tasks = new ArrayList<>(tasks); Collections.sort(tasks); for(int i=0; i<sourceTasks.size(); i++) { int s = sourceTasks.get(i); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/planner/Node.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/planner/Node.java b/storm-core/src/jvm/storm/trident/planner/Node.java index 6284cb9..d35e7da 100644 --- a/storm-core/src/jvm/storm/trident/planner/Node.java +++ b/storm-core/src/jvm/storm/trident/planner/Node.java @@ -26,7 +26,7 @@ import org.apache.commons.lang.builder.ToStringStyle; public class Node implements Serializable { - private static AtomicInteger INDEX = new AtomicInteger(0); + private static final AtomicInteger INDEX = new AtomicInteger(0); private String nodeId; @@ -47,6 +47,9 @@ public class Node implements Serializable { @Override public boolean equals(Object o) { + if (this == o) { + return true; + } return nodeId.equals(((Node) o).nodeId); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/planner/PartitionNode.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/planner/PartitionNode.java b/storm-core/src/jvm/storm/trident/planner/PartitionNode.java index 4f10c25..4485765 100644 --- a/storm-core/src/jvm/storm/trident/planner/PartitionNode.java +++ b/storm-core/src/jvm/storm/trident/planner/PartitionNode.java @@ -22,8 +22,6 @@ import backtype.storm.tuple.Fields; import java.io.IOException; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; -import java.util.ArrayList; -import java.util.List; import storm.trident.util.TridentUtils; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java b/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java index cdceaf9..6320a21 100644 --- a/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java +++ b/storm-core/src/jvm/storm/trident/planner/SubtopologyBolt.java @@ -23,7 +23,6 @@ import backtype.storm.task.TopologyContext; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; -import backtype.storm.utils.Utils; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -49,9 +48,9 @@ import storm.trident.util.TridentUtils; public class SubtopologyBolt implements ITridentBatchBolt { DirectedGraph _graph; Set<Node> _nodes; - Map<String, InitialReceiver> _roots = new HashMap(); - Map<Node, Factory> _outputFactories = new HashMap(); - Map<String, List<TridentProcessor>> _myTopologicallyOrdered = new HashMap(); + Map<String, InitialReceiver> _roots = new HashMap<>(); + Map<Node, Factory> _outputFactories = new HashMap<>(); + Map<String, List<TridentProcessor>> _myTopologicallyOrdered = new HashMap<>(); Map<Node, String> _batchGroups; //given processornodes and static state nodes @@ -71,7 +70,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { } } DirectedSubgraph<Node, Object> subgraph = new DirectedSubgraph(_graph, _nodes, null); - TopologicalOrderIterator it = new TopologicalOrderIterator<Node, Object>(subgraph); + TopologicalOrderIterator it = new TopologicalOrderIterator<>(subgraph); int stateIndex = 0; while(it.hasNext()) { Node n = (Node) it.next(); @@ -82,8 +81,8 @@ public class SubtopologyBolt implements ITridentBatchBolt { _myTopologicallyOrdered.put(batchGroup, new ArrayList()); } _myTopologicallyOrdered.get(batchGroup).add(pn.processor); - List<String> parentStreams = new ArrayList(); - List<Factory> parentFactories = new ArrayList(); + List<String> parentStreams = new ArrayList<>(); + List<Factory> parentFactories = new ArrayList<>(); for(Node p: TridentUtils.getParents(_graph, n)) { parentStreams.add(p.streamId); if(_nodes.contains(p)) { @@ -96,7 +95,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { parentFactories.add(_roots.get(p.streamId).getOutputFactory()); } } - List<TupleReceiver> targets = new ArrayList(); + List<TupleReceiver> targets = new ArrayList<>(); boolean outgoingNode = false; for(Node cn: TridentUtils.getChildren(_graph, n)) { if(_nodes.contains(cn)) { @@ -185,7 +184,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { protected static class InitialReceiver { - List<TridentProcessor> _receivers = new ArrayList(); + List<TridentProcessor> _receivers = new ArrayList<>(); RootFactory _factory; ProjectionFactory _project; String _stream; @@ -195,7 +194,7 @@ public class SubtopologyBolt implements ITridentBatchBolt { // how to distinguish "batch" streams from non-batch streams? _stream = stream; _factory = new RootFactory(allFields); - List<String> projected = new ArrayList(allFields.toList()); + List<String> projected = new ArrayList<>(allFields.toList()); projected.remove(0); _project = new ProjectionFactory(_factory, new Fields(projected)); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/planner/processor/MultiReducerProcessor.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/planner/processor/MultiReducerProcessor.java b/storm-core/src/jvm/storm/trident/planner/processor/MultiReducerProcessor.java index 6777d2f..4086996 100644 --- a/storm-core/src/jvm/storm/trident/planner/processor/MultiReducerProcessor.java +++ b/storm-core/src/jvm/storm/trident/planner/processor/MultiReducerProcessor.java @@ -49,7 +49,7 @@ public class MultiReducerProcessor implements TridentProcessor { public void prepare(Map conf, TopologyContext context, TridentContext tridentContext) { List<Factory> parents = tridentContext.getParentTupleFactories(); _context = tridentContext; - _streamToIndex = new HashMap<String, Integer>(); + _streamToIndex = new HashMap<>(); List<String> parentStreams = tridentContext.getParentStreams(); for(int i=0; i<parentStreams.size(); i++) { _streamToIndex.put(parentStreams.get(i), i); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/spout/TridentSpoutExecutor.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/spout/TridentSpoutExecutor.java b/storm-core/src/jvm/storm/trident/spout/TridentSpoutExecutor.java index 63debdb..d43d4e4 100644 --- a/storm-core/src/jvm/storm/trident/spout/TridentSpoutExecutor.java +++ b/storm-core/src/jvm/storm/trident/spout/TridentSpoutExecutor.java @@ -39,7 +39,7 @@ import storm.trident.tuple.ConsList; public class TridentSpoutExecutor implements ITridentBatchBolt { public static final String ID_FIELD = "$tx"; - public static Logger LOG = LoggerFactory.getLogger(TridentSpoutExecutor.class); + public static final Logger LOG = LoggerFactory.getLogger(TridentSpoutExecutor.class); AddIdCollector _collector; ITridentSpout<Object> _spout; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/storm/trident/tuple/TridentTupleView.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/storm/trident/tuple/TridentTupleView.java b/storm-core/src/jvm/storm/trident/tuple/TridentTupleView.java index 04b554e..761b221 100644 --- a/storm-core/src/jvm/storm/trident/tuple/TridentTupleView.java +++ b/storm-core/src/jvm/storm/trident/tuple/TridentTupleView.java @@ -31,7 +31,9 @@ import java.util.Map; import java.util.Set; import java.util.Arrays; -//extends abstractlist so that it can be emitted directly as Storm tuples +/** + * Extends AbstractList so that it can be emitted directly as Storm tuples + */ public class TridentTupleView extends AbstractList<Object> implements TridentTuple { ValuePointer[] _index; Map<String, ValuePointer> _fieldIndex; @@ -114,7 +116,7 @@ public class TridentTupleView extends AbstractList<Object> implements TridentTup public OperationOutputFactory(Factory parent, Fields selfFields) { _parent = parent; - _fieldIndex = new HashMap(parent.getFieldIndex()); + _fieldIndex = new HashMap<>(parent.getFieldIndex()); int myIndex = parent.numDelegates(); for(int i=0; i<selfFields.size(); i++) { String field = selfFields.get(i); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/test/jvm/backtype/storm/utils/DisruptorQueueTest.java ---------------------------------------------------------------------- diff --git a/storm-core/test/jvm/backtype/storm/utils/DisruptorQueueTest.java b/storm-core/test/jvm/backtype/storm/utils/DisruptorQueueTest.java index 2ab0109..dfb78fc 100644 --- a/storm-core/test/jvm/backtype/storm/utils/DisruptorQueueTest.java +++ b/storm-core/test/jvm/backtype/storm/utils/DisruptorQueueTest.java @@ -22,7 +22,6 @@ import java.util.concurrent.atomic.AtomicReference; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.dsl.ProducerType; -import com.lmax.disruptor.InsufficientCapacityException; import org.junit.Assert; import org.junit.Test; import junit.framework.TestCase; @@ -41,7 +40,7 @@ public class DisruptorQueueTest extends TestCase { Runnable producer = new IncProducer(queue, i+100); - final AtomicReference<Object> result = new AtomicReference<Object>(); + final AtomicReference<Object> result = new AtomicReference<>(); Runnable consumer = new Consumer(queue, new EventHandler<Object>() { private boolean head = true; @@ -130,7 +129,7 @@ public class DisruptorQueueTest extends TestCase { queue.publish(i); } } - }; + } private static class Consumer implements Runnable { private EventHandler handler; @@ -151,7 +150,7 @@ public class DisruptorQueueTest extends TestCase { //break } } - }; + } private static DisruptorQueue createQueue(String name, int queueSize) { return new DisruptorQueue(name, ProducerType.MULTI, queueSize, 0L, 1, 1L);
