http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/Node.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/Node.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/Node.java index 883c65f..6c2f06b 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/Node.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/Node.java @@ -40,8 +40,8 @@ import backtype.storm.scheduler.WorkerSlot; */ public class Node { private static final Logger LOG = LoggerFactory.getLogger(Node.class); - private Map<String, Set<WorkerSlot>> _topIdToUsedSlots = new HashMap<String,Set<WorkerSlot>>(); - private Set<WorkerSlot> _freeSlots = new HashSet<WorkerSlot>(); + private Map<String, Set<WorkerSlot>> _topIdToUsedSlots = new HashMap<>(); + private Set<WorkerSlot> _freeSlots = new HashSet<>(); private final String _nodeId; private boolean _isAlive; @@ -143,7 +143,7 @@ public class Node { } Set<WorkerSlot> usedSlots = _topIdToUsedSlots.get(topId); if (usedSlots == null) { - usedSlots = new HashSet<WorkerSlot>(); + usedSlots = new HashSet<>(); _topIdToUsedSlots.put(topId, usedSlots); } usedSlots.add(ws); @@ -164,7 +164,7 @@ public class Node { _freeSlots.addAll(entry.getValue()); } } - _topIdToUsedSlots = new HashMap<String,Set<WorkerSlot>>(); + _topIdToUsedSlots = new HashMap<>(); } /** @@ -242,10 +242,7 @@ public class Node { @Override public boolean equals(Object other) { - if (other instanceof Node) { - return _nodeId.equals(((Node)other)._nodeId); - } - return false; + return other instanceof Node && _nodeId.equals(((Node) other)._nodeId); } @Override @@ -295,13 +292,13 @@ public class Node { } public static Map<String, Node> getAllNodesFrom(Cluster cluster) { - Map<String, Node> nodeIdToNode = new HashMap<String, Node>(); + Map<String, Node> nodeIdToNode = new HashMap<>(); for (SupervisorDetails sup : cluster.getSupervisors().values()) { //Node ID and supervisor ID are the same. String id = sup.getId(); boolean isAlive = !cluster.isBlackListed(id); LOG.debug("Found a {} Node {} {}", - new Object[] {isAlive? "living":"dead", id, sup.getAllPorts()}); + isAlive? "living":"dead", id, sup.getAllPorts()); nodeIdToNode.put(id, new Node(id, sup.getAllPorts(), isAlive)); }
http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/NodePool.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/NodePool.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/NodePool.java index 21d1577..5a46df5 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/NodePool.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/NodePool.java @@ -81,7 +81,7 @@ public abstract class NodePool { Map<ExecutorDetails, String> execToComp = td.getExecutorToComponent(); SchedulerAssignment assignment = _cluster.getAssignmentById(_topId); - _nodeToComps = new HashMap<String, Set<String>>(); + _nodeToComps = new HashMap<>(); if (assignment != null) { Map<ExecutorDetails, WorkerSlot> execToSlot = assignment.getExecutorToSlot(); @@ -90,14 +90,14 @@ public abstract class NodePool { String nodeId = entry.getValue().getNodeId(); Set<String> comps = _nodeToComps.get(nodeId); if (comps == null) { - comps = new HashSet<String>(); + comps = new HashSet<>(); _nodeToComps.put(nodeId, comps); } comps.add(execToComp.get(entry.getKey())); } } - _spreadToSchedule = new HashMap<String, List<ExecutorDetails>>(); + _spreadToSchedule = new HashMap<>(); List<String> spreadComps = (List<String>)td.getConf().get(Config.TOPOLOGY_SPREAD_COMPONENTS); if (spreadComps != null) { for (String comp: spreadComps) { @@ -105,7 +105,7 @@ public abstract class NodePool { } } - _slots = new LinkedList<Set<ExecutorDetails>>(); + _slots = new LinkedList<>(); for (int i = 0; i < slotsToUse; i++) { _slots.add(new HashSet<ExecutorDetails>()); } @@ -118,7 +118,7 @@ public abstract class NodePool { _spreadToSchedule.get(entry.getKey()).addAll(entry.getValue()); } else { for (ExecutorDetails ed: entry.getValue()) { - LOG.debug("Assigning {} {} to slot {}", new Object[]{entry.getKey(), ed, at}); + LOG.debug("Assigning {} {} to slot {}", entry.getKey(), ed, at); _slots.get(at).add(ed); at++; if (at >= _slots.size()) { @@ -151,7 +151,7 @@ public abstract class NodePool { String nodeId = n.getId(); Set<String> nodeComps = _nodeToComps.get(nodeId); if (nodeComps == null) { - nodeComps = new HashSet<String>(); + nodeComps = new HashSet<>(); _nodeToComps.put(nodeId, nodeComps); } for (Entry<String, List<ExecutorDetails>> entry: _spreadToSchedule.entrySet()) { @@ -251,7 +251,7 @@ public abstract class NodePool { public static Collection<Node> takeNodesBySlot(int slotsNeeded,NodePool[] pools) { LOG.debug("Trying to grab {} free slots from {}",slotsNeeded, pools); - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); for (NodePool pool: pools) { Collection<Node> got = pool.takeNodesBySlots(slotsNeeded); ret.addAll(got); @@ -266,7 +266,7 @@ public abstract class NodePool { public static Collection<Node> takeNodes(int nodesNeeded,NodePool[] pools) { LOG.debug("Trying to grab {} free nodes from {}",nodesNeeded, pools); - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); for (NodePool pool: pools) { Collection<Node> got = pool.takeNodes(nodesNeeded); ret.addAll(got); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/resource/strategies/ResourceAwareStrategy.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/resource/strategies/ResourceAwareStrategy.java b/storm-core/src/jvm/backtype/storm/scheduler/resource/strategies/ResourceAwareStrategy.java index f98d9ed..b8a96a2 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/resource/strategies/ResourceAwareStrategy.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/resource/strategies/ResourceAwareStrategy.java @@ -42,7 +42,7 @@ import backtype.storm.scheduler.resource.Component; import backtype.storm.scheduler.resource.RAS_Node; public class ResourceAwareStrategy implements IStrategy { - private Logger LOG = null; + private static final Logger LOG = LoggerFactory.getLogger(ResourceAwareStrategy.class); private Topologies _topologies; private Cluster _cluster; //Map key is the supervisor id and the value is the corresponding RAS_Node Object @@ -63,7 +63,6 @@ public class ResourceAwareStrategy implements IStrategy { _cluster = cluster; _nodes = RAS_Node.getAllNodesFrom(cluster, _topologies); _availNodes = this.getAvailNodes(); - this.LOG = LoggerFactory.getLogger(this.getClass()); _clusterInfo = cluster.getNetworkTopography(); LOG.debug(this.getClusterInfo()); } @@ -71,7 +70,7 @@ public class ResourceAwareStrategy implements IStrategy { //the returned TreeMap keeps the Components sorted private TreeMap<Integer, List<ExecutorDetails>> getPriorityToExecutorDetailsListMap( Queue<Component> ordered__Component_list, Collection<ExecutorDetails> unassignedExecutors) { - TreeMap<Integer, List<ExecutorDetails>> retMap = new TreeMap<Integer, List<ExecutorDetails>>(); + TreeMap<Integer, List<ExecutorDetails>> retMap = new TreeMap<>(); Integer rank = 0; for (Component ras_comp : ordered__Component_list) { retMap.put(rank, new ArrayList<ExecutorDetails>()); @@ -91,9 +90,9 @@ public class ResourceAwareStrategy implements IStrategy { return null; } Collection<ExecutorDetails> unassignedExecutors = _cluster.getUnassignedExecutors(td); - Map<WorkerSlot, Collection<ExecutorDetails>> schedulerAssignmentMap = new HashMap<WorkerSlot, Collection<ExecutorDetails>>(); + Map<WorkerSlot, Collection<ExecutorDetails>> schedulerAssignmentMap = new HashMap<>(); LOG.debug("ExecutorsNeedScheduling: {}", unassignedExecutors); - Collection<ExecutorDetails> scheduledTasks = new ArrayList<ExecutorDetails>(); + Collection<ExecutorDetails> scheduledTasks = new ArrayList<>(); List<Component> spouts = this.getSpouts(_topologies, td); if (spouts.size() == 0) { @@ -104,7 +103,7 @@ public class ResourceAwareStrategy implements IStrategy { Queue<Component> ordered__Component_list = bfs(_topologies, td, spouts); Map<Integer, List<ExecutorDetails>> priorityToExecutorMap = getPriorityToExecutorDetailsListMap(ordered__Component_list, unassignedExecutors); - Collection<ExecutorDetails> executorsNotScheduled = new HashSet<ExecutorDetails>(unassignedExecutors); + Collection<ExecutorDetails> executorsNotScheduled = new HashSet<>(unassignedExecutors); Integer longestPriorityListSize = this.getLongestPriorityListSize(priorityToExecutorMap); //Pick the first executor with priority one, then the 1st exec with priority 2, so on an so forth. //Once we reach the last priority, we go back to priority 1 and schedule the second task with priority 1. @@ -145,7 +144,7 @@ public class ResourceAwareStrategy implements IStrategy { WorkerSlot targetSlot = this.findWorkerForExec(exec, td, schedulerAssignmentMap); if (targetSlot != null) { RAS_Node targetNode = this.idToNode(targetSlot.getNodeId()); - if(schedulerAssignmentMap.containsKey(targetSlot) == false) { + if(!schedulerAssignmentMap.containsKey(targetSlot)) { schedulerAssignmentMap.put(targetSlot, new LinkedList<ExecutorDetails>()); } @@ -175,7 +174,7 @@ public class ResourceAwareStrategy implements IStrategy { } private WorkerSlot findWorkerForExec(ExecutorDetails exec, TopologyDetails td, Map<WorkerSlot, Collection<ExecutorDetails>> scheduleAssignmentMap) { - WorkerSlot ws = null; + WorkerSlot ws; // first scheduling if (this.refNode == null) { String clus = this.getBestClustering(); @@ -205,7 +204,7 @@ public class ResourceAwareStrategy implements IStrategy { nodes = this.getAvailableNodes(); } //First sort nodes by distance - TreeMap<Double, RAS_Node> nodeRankMap = new TreeMap<Double, RAS_Node>(); + TreeMap<Double, RAS_Node> nodeRankMap = new TreeMap<>(); for (RAS_Node n : nodes) { if(n.getFreeSlots().size()>0) { if (n.getAvailableMemoryResources() >= taskMem @@ -262,9 +261,9 @@ public class ResourceAwareStrategy implements IStrategy { } private Double distToNode(RAS_Node src, RAS_Node dest) { - if (src.getId().equals(dest.getId()) == true) { + if (src.getId().equals(dest.getId())) { return 0.0; - }else if (this.NodeToCluster(src) == this.NodeToCluster(dest)) { + } else if (this.NodeToCluster(src) == this.NodeToCluster(dest)) { return 0.5; } else { return 1.0; @@ -283,7 +282,7 @@ public class ResourceAwareStrategy implements IStrategy { } private List<RAS_Node> getAvailableNodes() { - LinkedList<RAS_Node> nodes = new LinkedList<RAS_Node>(); + LinkedList<RAS_Node> nodes = new LinkedList<>(); for (String clusterId : _clusterInfo.keySet()) { nodes.addAll(this.getAvailableNodesFromCluster(clusterId)); } @@ -291,7 +290,7 @@ public class ResourceAwareStrategy implements IStrategy { } private List<RAS_Node> getAvailableNodesFromCluster(String clus) { - List<RAS_Node> retList = new ArrayList<RAS_Node>(); + List<RAS_Node> retList = new ArrayList<>(); for (String node_id : _clusterInfo.get(clus)) { retList.add(_availNodes.get(this .NodeHostnameToId(node_id))); @@ -301,7 +300,7 @@ public class ResourceAwareStrategy implements IStrategy { private List<WorkerSlot> getAvailableWorkersFromCluster(String clusterId) { List<RAS_Node> nodes = this.getAvailableNodesFromCluster(clusterId); - List<WorkerSlot> workers = new LinkedList<WorkerSlot>(); + List<WorkerSlot> workers = new LinkedList<>(); for(RAS_Node node : nodes) { workers.addAll(node.getFreeSlots()); } @@ -309,7 +308,7 @@ public class ResourceAwareStrategy implements IStrategy { } private List<WorkerSlot> getAvailableWorker() { - List<WorkerSlot> workers = new LinkedList<WorkerSlot>(); + List<WorkerSlot> workers = new LinkedList<>(); for (String clusterId : _clusterInfo.keySet()) { workers.addAll(this.getAvailableWorkersFromCluster(clusterId)); } @@ -333,18 +332,18 @@ public class ResourceAwareStrategy implements IStrategy { private Queue<Component> bfs(Topologies topologies, TopologyDetails td, List<Component> spouts) { // Since queue is a interface Queue<Component> ordered__Component_list = new LinkedList<Component>(); - HashMap<String, Component> visited = new HashMap<String, Component>(); + HashMap<String, Component> visited = new HashMap<>(); /* start from each spout that is not visited, each does a breadth-first traverse */ for (Component spout : spouts) { if (!visited.containsKey(spout.id)) { - Queue<Component> queue = new LinkedList<Component>(); + Queue<Component> queue = new LinkedList<>(); queue.offer(spout); while (!queue.isEmpty()) { Component comp = queue.poll(); visited.put(comp.id, comp); ordered__Component_list.add(comp); - List<String> neighbors = new ArrayList<String>(); + List<String> neighbors = new ArrayList<>(); neighbors.addAll(comp.children); neighbors.addAll(comp.parents); for (String nbID : neighbors) { @@ -360,7 +359,7 @@ public class ResourceAwareStrategy implements IStrategy { } private List<Component> getSpouts(Topologies topologies, TopologyDetails td) { - List<Component> spouts = new ArrayList<Component>(); + List<Component> spouts = new ArrayList<>(); for (Component c : topologies.getAllComponents().get(td.getId()) .values()) { if (c.type == Component.ComponentType.SPOUT) { @@ -413,7 +412,7 @@ public class ResourceAwareStrategy implements IStrategy { /** * Checks whether we can schedule an Executor exec on the worker slot ws - * Only considers memory currenlty. May include CPU in the future + * Only considers memory currently. May include CPU in the future * @param exec * @param ws * @param td http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java b/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java index ac3fb53..8062b4e 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/AuthUtils.java @@ -21,21 +21,18 @@ import backtype.storm.Config; import javax.security.auth.login.Configuration; import javax.security.auth.login.AppConfigurationEntry; import javax.security.auth.Subject; -import java.security.NoSuchAlgorithmException; import java.security.URIParameter; import backtype.storm.security.INimbusCredentialPlugin; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; -import java.io.FileNotFoundException; import java.io.IOException; import java.net.URI; import java.util.Collection; import java.util.Set; import java.util.HashSet; import java.util.Map; -import java.util.concurrent.ExecutorService; public class AuthUtils { private static final Logger LOG = LoggerFactory.getLogger(AuthUtils.class); @@ -72,11 +69,11 @@ public class AuthUtils { /** * Construct a principal to local plugin - * @param conf storm configuration + * @param storm_conf storm configuration * @return the plugin */ public static IPrincipalToLocal GetPrincipalToLocalPlugin(Map storm_conf) { - IPrincipalToLocal ptol = null; + IPrincipalToLocal ptol; try { String ptol_klassName = (String) storm_conf.get(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN); Class klass = Class.forName(ptol_klassName); @@ -90,11 +87,11 @@ public class AuthUtils { /** * Construct a group mapping service provider plugin - * @param conf storm configuration + * @param storm_conf storm configuration * @return the plugin */ public static IGroupMappingServiceProvider GetGroupMappingServiceProviderPlugin(Map storm_conf) { - IGroupMappingServiceProvider gmsp = null; + IGroupMappingServiceProvider gmsp; try { String gmsp_klassName = (String) storm_conf.get(Config.STORM_GROUP_MAPPING_SERVICE_PROVIDER_PLUGIN); Class klass = Class.forName(gmsp_klassName); @@ -107,13 +104,13 @@ public class AuthUtils { } /** - * Get all of the configured Credential Renwer Plugins. - * @param storm_conf the storm configuration to use. + * Get all of the configured Credential Renewer Plugins. + * @param conf the storm configuration to use. * @return the configured credential renewers. */ public static Collection<ICredentialsRenewer> GetCredentialRenewers(Map conf) { try { - Set<ICredentialsRenewer> ret = new HashSet<ICredentialsRenewer>(); + Set<ICredentialsRenewer> ret = new HashSet<>(); Collection<String> clazzes = (Collection<String>)conf.get(Config.NIMBUS_CREDENTIAL_RENEWERS); if (clazzes != null) { for (String clazz : clazzes) { @@ -135,7 +132,7 @@ public class AuthUtils { */ public static Collection<INimbusCredentialPlugin> getNimbusAutoCredPlugins(Map conf) { try { - Set<INimbusCredentialPlugin> ret = new HashSet<INimbusCredentialPlugin>(); + Set<INimbusCredentialPlugin> ret = new HashSet<>(); Collection<String> clazzes = (Collection<String>)conf.get(Config.NIMBUS_AUTO_CRED_PLUGINS); if (clazzes != null) { for (String clazz : clazzes) { @@ -157,7 +154,7 @@ public class AuthUtils { */ public static Collection<IAutoCredentials> GetAutoCredentials(Map storm_conf) { try { - Set<IAutoCredentials> autos = new HashSet<IAutoCredentials>(); + Set<IAutoCredentials> autos = new HashSet<>(); Collection<String> clazzes = (Collection<String>)storm_conf.get(Config.TOPOLOGY_AUTO_CREDENTIALS); if (clazzes != null) { for (String clazz : clazzes) { @@ -216,11 +213,9 @@ public class AuthUtils { /** * Construct a transport plugin per storm configuration - * @param conf storm configuration - * @return */ public static ITransportPlugin GetTransportPlugin(ThriftConnectionType type, Map storm_conf, Configuration login_conf) { - ITransportPlugin transportPlugin = null; + ITransportPlugin transportPlugin; try { String transport_plugin_klassName = type.getTransportPlugin(storm_conf); Class klass = Class.forName(transport_plugin_klassName); @@ -234,7 +229,7 @@ public class AuthUtils { private static IHttpCredentialsPlugin GetHttpCredentialsPlugin(Map conf, String klassName) { - IHttpCredentialsPlugin plugin = null; + IHttpCredentialsPlugin plugin; try { Class klass = Class.forName(klassName); plugin = (IHttpCredentialsPlugin)klass.newInstance(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/DefaultHttpCredentialsPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/DefaultHttpCredentialsPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/DefaultHttpCredentialsPlugin.java index e2469e5..9c81cdf 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/DefaultHttpCredentialsPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/DefaultHttpCredentialsPlugin.java @@ -28,8 +28,6 @@ import javax.servlet.http.HttpServletRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import backtype.storm.security.auth.ReqContext; - public class DefaultHttpCredentialsPlugin implements IHttpCredentialsPlugin { private static final Logger LOG = LoggerFactory.getLogger(DefaultHttpCredentialsPlugin.class); @@ -50,7 +48,7 @@ public class DefaultHttpCredentialsPlugin implements IHttpCredentialsPlugin { */ @Override public String getUserName(HttpServletRequest req) { - Principal princ = null; + Principal princ; if (req != null && (princ = req.getUserPrincipal()) != null) { String userName = princ.getName(); if (userName != null && !userName.isEmpty()) { @@ -83,7 +81,7 @@ public class DefaultHttpCredentialsPlugin implements IHttpCredentialsPlugin { userName = doAsUser; } - Set<Principal> principals = new HashSet<Principal>(); + Set<Principal> principals = new HashSet<>(); if(userName != null) { Principal p = new SingleUserPrincipal(userName); principals.add(p); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/DefaultPrincipalToLocal.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/DefaultPrincipalToLocal.java b/storm-core/src/jvm/backtype/storm/security/auth/DefaultPrincipalToLocal.java index 729d744..9f95101 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/DefaultPrincipalToLocal.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/DefaultPrincipalToLocal.java @@ -28,7 +28,6 @@ import java.security.Principal; public class DefaultPrincipalToLocal implements IPrincipalToLocal { /** * Invoked once immediately after construction - * @param conf Storm configuration */ public void prepare(Map storm_conf) {} http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/IAuthorizer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/IAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/IAuthorizer.java index d592bb7..ff1e2ba 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/IAuthorizer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/IAuthorizer.java @@ -32,7 +32,7 @@ import java.util.Map; public interface IAuthorizer { /** * Invoked once immediately after construction - * @param conf Storm configuration + * @param storm_conf Storm configuration */ void prepare(Map storm_conf); @@ -40,7 +40,7 @@ public interface IAuthorizer { * permit() method is invoked for each incoming Thrift request. * @param context request context includes info about * @param operation operation name - * @param topology_storm configuration of targeted topology + * @param topology_conf configuration of targeted topology * @return true if the request is authorized, false if reject */ public boolean permit(ReqContext context, String operation, Map topology_conf); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java b/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java index 3eaf6c4..9a6f02e 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/ICredentialsRenewer.java @@ -18,11 +18,10 @@ package backtype.storm.security.auth; -import java.util.Collection; import java.util.Map; /** - * Provides a way to renew credentials on behelf of a user. + * Provides a way to renew credentials on behalf of a user. */ public interface ICredentialsRenewer { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/IHttpCredentialsPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/IHttpCredentialsPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/IHttpCredentialsPlugin.java index a012ce4..0b57eca 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/IHttpCredentialsPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/IHttpCredentialsPlugin.java @@ -21,8 +21,6 @@ package backtype.storm.security.auth; import java.util.Map; import javax.servlet.http.HttpServletRequest; -import backtype.storm.security.auth.ReqContext; - /** * Interface for handling credentials in an HttpServletRequest */ http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/IPrincipalToLocal.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/IPrincipalToLocal.java b/storm-core/src/jvm/backtype/storm/security/auth/IPrincipalToLocal.java index fca3d37..e938d39 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/IPrincipalToLocal.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/IPrincipalToLocal.java @@ -28,7 +28,7 @@ import java.security.Principal; public interface IPrincipalToLocal { /** * Invoked once immediately after construction - * @param conf Storm configuration + * @param storm_conf Storm configuration */ void prepare(Map storm_conf); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/ITransportPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ITransportPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/ITransportPlugin.java index 5ba2557..ba09fad 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/ITransportPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/ITransportPlugin.java @@ -18,9 +18,7 @@ package backtype.storm.security.auth; import java.io.IOException; -import java.security.Principal; import java.util.Map; -import java.util.concurrent.ExecutorService; import javax.security.auth.login.Configuration; @@ -29,8 +27,6 @@ import org.apache.thrift.server.TServer; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; -import backtype.storm.security.auth.ThriftConnectionType; - /** * Interface for Thrift Transport plugin */ http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/KerberosPrincipalToLocal.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/KerberosPrincipalToLocal.java b/storm-core/src/jvm/backtype/storm/security/auth/KerberosPrincipalToLocal.java index 35c7788..1f67c14 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/KerberosPrincipalToLocal.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/KerberosPrincipalToLocal.java @@ -28,7 +28,7 @@ public class KerberosPrincipalToLocal implements IPrincipalToLocal { /** * Invoked once immediately after construction - * @param conf Storm configuration + * @param storm_conf Storm configuration */ public void prepare(Map storm_conf) {} http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/ReqContext.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ReqContext.java b/storm-core/src/jvm/backtype/storm/security/auth/ReqContext.java index 2f18982..31aeef9 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/ReqContext.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/ReqContext.java @@ -22,8 +22,6 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.net.InetAddress; import com.google.common.annotations.VisibleForTesting; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.security.AccessControlContext; import java.security.AccessController; @@ -44,12 +42,8 @@ public class ReqContext { private Map _storm_conf; private Principal realPrincipal; - private static final Logger LOG = LoggerFactory.getLogger(ReqContext.class); - - /** - * Get a request context associated with current thread - * @return + * @return a request context associated with current thread */ public static ReqContext context() { return ctxt.get(); @@ -132,8 +126,7 @@ public class ReqContext { } /** - * Returns true if this request is an impersonation request. - * @return + * @return true if this request is an impersonation request. */ public boolean isImpersonating() { return this.realPrincipal != null; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/SaslTransportPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/SaslTransportPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/SaslTransportPlugin.java index 7013cd4..92004fa 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/SaslTransportPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/SaslTransportPlugin.java @@ -45,10 +45,6 @@ import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; import org.apache.thrift.transport.TTransportFactory; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import backtype.storm.security.auth.ThriftConnectionType; /** * Base class for SASL authentication plugin. @@ -57,7 +53,6 @@ public abstract class SaslTransportPlugin implements ITransportPlugin { protected ThriftConnectionType type; protected Map storm_conf; protected Configuration login_conf; - private static final Logger LOG = LoggerFactory.getLogger(SaslTransportPlugin.class); @Override public void prepare(ThriftConnectionType type, Map storm_conf, Configuration login_conf) { @@ -95,7 +90,7 @@ public abstract class SaslTransportPlugin implements ITransportPlugin { /** * All subclass must implement this method - * @return + * @return server transport factory * @throws IOException */ protected abstract TTransportFactory getServerTransportFactory() throws IOException; @@ -162,11 +157,8 @@ public abstract class SaslTransportPlugin implements ITransportPlugin { public boolean equals(Object o) { if (this == o) { return true; - } else if (o == null || getClass() != o.getClass()) { - return false; - } else { - return (name.equals(((User) o).name)); } + return !(o == null || getClass() != o.getClass()) && (name.equals(((User) o).name)); } @Override http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/SimpleTransportPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/SimpleTransportPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/SimpleTransportPlugin.java index 2abcdae..7a0a6f2 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/SimpleTransportPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/SimpleTransportPlugin.java @@ -45,8 +45,6 @@ import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import backtype.storm.security.auth.ThriftConnectionType; - /** * Simple transport for Thrift plugin. * @@ -146,12 +144,12 @@ public class SimpleTransportPlugin implements ITransportPlugin { if (s == null) { final String user = (String)storm_conf.get("debug.simple.transport.user"); if (user != null) { - HashSet<Principal> principals = new HashSet<Principal>(); + HashSet<Principal> principals = new HashSet<>(); principals.add(new Principal() { public String getName() { return user; } public String toString() { return user; } }); - s = new Subject(true, principals, new HashSet<Object>(), new HashSet<Object>()); + s = new Subject(true, principals, new HashSet<>(), new HashSet<>()); } } req_context.setSubject(s); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/SingleUserPrincipal.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/SingleUserPrincipal.java b/storm-core/src/jvm/backtype/storm/security/auth/SingleUserPrincipal.java index 6af17fa..0cadba6 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/SingleUserPrincipal.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/SingleUserPrincipal.java @@ -33,10 +33,7 @@ public class SingleUserPrincipal implements Principal { @Override public boolean equals(Object another) { - if (another instanceof SingleUserPrincipal) { - return _userName.equals(((SingleUserPrincipal)another)._userName); - } - return false; + return another instanceof SingleUserPrincipal && _userName.equals(((SingleUserPrincipal) another)._userName); } @Override http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/TBackoffConnect.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/TBackoffConnect.java b/storm-core/src/jvm/backtype/storm/security/auth/TBackoffConnect.java index f547868..9729671 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/TBackoffConnect.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/TBackoffConnect.java @@ -19,7 +19,6 @@ package backtype.storm.security.auth; import java.io.IOException; -import java.util.Random; import org.apache.thrift.transport.TTransport; import org.apache.thrift.transport.TTransportException; import org.slf4j.Logger; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/ThriftClient.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ThriftClient.java b/storm-core/src/jvm/backtype/storm/security/auth/ThriftClient.java index 9f77ab9..8b3d4c5 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/ThriftClient.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/ThriftClient.java @@ -24,14 +24,10 @@ import org.apache.thrift.protocol.TBinaryProtocol; import org.apache.thrift.protocol.TProtocol; import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import backtype.storm.utils.Utils; import backtype.storm.Config; -import backtype.storm.security.auth.TBackoffConnect; -public class ThriftClient { - private static final Logger LOG = LoggerFactory.getLogger(ThriftClient.class); +public class ThriftClient { private TTransport _transport; protected TProtocol _protocol; private String _host; @@ -90,8 +86,6 @@ public class ThriftClient { //construct a transport plugin ITransportPlugin transportPlugin = AuthUtils.GetTransportPlugin(_type, _conf, login_conf); - final TTransport underlyingTransport = socket; - //TODO get this from type instead of hardcoding to Nimbus. //establish client-server transport via plugin //do retries if the connect fails @@ -100,7 +94,7 @@ public class ThriftClient { Utils.getInt(_conf.get(Config.STORM_NIMBUS_RETRY_TIMES)), Utils.getInt(_conf.get(Config.STORM_NIMBUS_RETRY_INTERVAL)), Utils.getInt(_conf.get(Config.STORM_NIMBUS_RETRY_INTERVAL_CEILING))); - _transport = connectionRetry.doConnectWithRetry(transportPlugin, underlyingTransport, _host, _asUser); + _transport = connectionRetry.doConnectWithRetry(transportPlugin, socket, _host, _asUser); } catch (IOException ex) { throw new RuntimeException(ex); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/ThriftServer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/ThriftServer.java b/storm-core/src/jvm/backtype/storm/security/auth/ThriftServer.java index 64243ce..fdbdc7c 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/ThriftServer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/ThriftServer.java @@ -53,12 +53,10 @@ public class ThriftServer { } /** - * Is ThriftServer listening to requests? - * @return + * @return true if ThriftServer is listening to requests? */ public boolean isServing() { - if (_server == null) return false; - return _server.isServing(); + return _server != null && _server.isServing(); } public void serve() { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/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 5e83b9f..4a9b379 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 @@ -28,7 +28,6 @@ import java.util.Set; import backtype.storm.Config; import backtype.storm.security.auth.ReqContext; -import backtype.storm.security.auth.authorizer.DRPCAuthorizerBase; import backtype.storm.security.auth.AuthUtils; import backtype.storm.security.auth.IPrincipalToLocal; import backtype.storm.utils.Utils; @@ -54,7 +53,7 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase { public AclFunctionEntry(Collection<String> clientUsers, String invocationUser) { this.clientUsers = (clientUsers != null) ? - new HashSet<String>(clientUsers) : new HashSet<String>(); + new HashSet<>(clientUsers) : new HashSet<String>(); this.invocationUser = invocationUser; } } @@ -68,7 +67,7 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase { //change is atomic long now = System.currentTimeMillis(); if ((now - 5000) > _lastUpdate || _acl == null) { - Map<String,AclFunctionEntry> acl = new HashMap<String,AclFunctionEntry>(); + Map<String,AclFunctionEntry> acl = new HashMap<>(); Map conf = Utils.findAndReadConfigFile(_aclFileName); if (conf.containsKey(Config.DRPC_AUTHORIZER_ACL)) { Map<String,Map<String,?>> confAcl = @@ -88,7 +87,7 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase { } } else if (!_permitWhenMissingFunctionEntry) { LOG.warn("Requiring explicit ACL entries, but none given. " + - "Therefore, all operiations will be denied."); + "Therefore, all operations will be denied."); } _acl = acl; _lastUpdate = System.currentTimeMillis(); @@ -100,8 +99,8 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase { public void prepare(Map conf) { Boolean isStrict = (Boolean) conf.get(Config.DRPC_AUTHORIZER_ACL_STRICT); - _permitWhenMissingFunctionEntry = - (isStrict != null && !isStrict) ? true : false; + _permitWhenMissingFunctionEntry = + (isStrict != null && !isStrict); _aclFileName = (String) conf.get(Config.DRPC_AUTHORIZER_ACL_FILENAME); _ptol = AuthUtils.GetPrincipalToLocalPlugin(conf); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DenyAuthorizer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DenyAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DenyAuthorizer.java index 8d61492..d1d6f87 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DenyAuthorizer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/DenyAuthorizer.java @@ -19,19 +19,14 @@ package backtype.storm.security.auth.authorizer; import java.util.Map; -import backtype.storm.Config; import backtype.storm.security.auth.IAuthorizer; import backtype.storm.security.auth.ReqContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** * An authorization implementation that denies everything, for testing purposes */ public class DenyAuthorizer implements IAuthorizer { - private static final Logger LOG = LoggerFactory.getLogger(DenyAuthorizer.class); - + /** * Invoked once immediately after construction * @param conf Storm configuration @@ -41,9 +36,9 @@ public class DenyAuthorizer implements IAuthorizer { /** * permit() method is invoked for each incoming Thrift request - * @param contrext request context + * @param context request context * @param operation operation name - * @param topology_storm configuration of targeted topology + * @param topology_conf configuration of targeted topology * @return true if the request is authorized, false if reject */ public boolean permit(ReqContext context, String operation, Map topology_conf) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/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 0cfd488..07e6447 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 @@ -38,7 +38,7 @@ public class ImpersonationAuthorizer implements IAuthorizer { @Override public void prepare(Map conf) { - userImpersonationACL = new HashMap<String, ImpersonationACL>(); + userImpersonationACL = new HashMap<>(); Map<String, Map<String, List<String>>> userToHostAndGroup = (Map<String, Map<String, List<String>>>) conf.get(Config.NIMBUS_IMPERSONATION_ACL); @@ -67,7 +67,7 @@ public class ImpersonationAuthorizer implements IAuthorizer { String userBeingImpersonated = _ptol.toLocal(context.principal()); InetAddress remoteAddress = context.remoteAddress(); - LOG.info("user = {}, principal = {} is attmepting to impersonate user = {} for operation = {} from host = {}", + LOG.info("user = {}, principal = {} is attempting to impersonate user = {} for operation = {} from host = {}", impersonatingUser, impersonatingPrincipal, userBeingImpersonated, operation, remoteAddress); /** @@ -83,8 +83,8 @@ public class ImpersonationAuthorizer implements IAuthorizer { ImpersonationACL principalACL = userImpersonationACL.get(impersonatingPrincipal); ImpersonationACL userACL = userImpersonationACL.get(impersonatingUser); - Set<String> authorizedHosts = new HashSet<String>(); - Set<String> authorizedGroups = new HashSet<String>(); + Set<String> authorizedHosts = new HashSet<>(); + Set<String> authorizedGroups = new HashSet<>(); if (principalACL != null) { authorizedHosts.addAll(principalACL.authorizedHosts); @@ -127,7 +127,7 @@ public class ImpersonationAuthorizer implements IAuthorizer { return true; } - Set<String> groups = null; + Set<String> groups; try { groups = _groupMappingProvider.getGroups(userBeingImpersonated); } catch (IOException e) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/authorizer/NoopAuthorizer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/NoopAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/NoopAuthorizer.java index c8008f1..ab5bd4b 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/NoopAuthorizer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/NoopAuthorizer.java @@ -19,18 +19,13 @@ package backtype.storm.security.auth.authorizer; import java.util.Map; -import backtype.storm.Config; import backtype.storm.security.auth.IAuthorizer; import backtype.storm.security.auth.ReqContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** * A no-op authorization implementation that illustrate info available for authorization decisions. */ public class NoopAuthorizer implements IAuthorizer { - private static final Logger LOG = LoggerFactory.getLogger(NoopAuthorizer.class); /** * Invoked once immediately after construction @@ -43,7 +38,7 @@ public class NoopAuthorizer implements IAuthorizer { * permit() method is invoked for each incoming Thrift request * @param context request context includes info about * @param operation operation name - * @param topology_storm configuration of targeted topology + * @param topology_conf configuration of targeted topology * @return true if the request is authorized, false if reject */ public boolean permit(ReqContext context, String operation, Map topology_conf) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleACLAuthorizer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleACLAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleACLAuthorizer.java index a2549a5..0063f92 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleACLAuthorizer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleACLAuthorizer.java @@ -42,9 +42,9 @@ import org.slf4j.LoggerFactory; public class SimpleACLAuthorizer implements IAuthorizer { private static final Logger LOG = LoggerFactory.getLogger(SimpleACLAuthorizer.class); - protected Set<String> _userCommands = new HashSet<String>(Arrays.asList("submitTopology", "fileUpload", "getNimbusConf", "getClusterInfo")); - protected Set<String> _supervisorCommands = new HashSet<String>(Arrays.asList("fileDownload")); - protected Set<String> _topoCommands = new HashSet<String>(Arrays.asList( + protected Set<String> _userCommands = new HashSet<>(Arrays.asList("submitTopology", "fileUpload", "getNimbusConf", "getClusterInfo")); + protected Set<String> _supervisorCommands = new HashSet<>(Arrays.asList("fileDownload")); + protected Set<String> _topoCommands = new HashSet<>(Arrays.asList( "killTopology", "rebalance", "activate", @@ -79,10 +79,10 @@ public class SimpleACLAuthorizer implements IAuthorizer { */ @Override public void prepare(Map conf) { - _admins = new HashSet<String>(); - _supervisors = new HashSet<String>(); - _nimbusUsers = new HashSet<String>(); - _nimbusGroups = new HashSet<String>(); + _admins = new HashSet<>(); + _supervisors = new HashSet<>(); + _nimbusUsers = new HashSet<>(); + _nimbusGroups = new HashSet<>(); if (conf.containsKey(Config.NIMBUS_ADMINS)) { _admins.addAll((Collection<String>)conf.get(Config.NIMBUS_ADMINS)); @@ -113,7 +113,7 @@ public class SimpleACLAuthorizer implements IAuthorizer { public boolean permit(ReqContext context, String operation, Map topology_conf) { String principal = context.principal().getName(); String user = _ptol.toLocal(context.principal()); - Set<String> userGroups = new HashSet<String>(); + Set<String> userGroups = new HashSet<>(); if (_groupMappingProvider != null) { try { @@ -145,7 +145,7 @@ public class SimpleACLAuthorizer implements IAuthorizer { return true; } - Set<String> topoGroups = new HashSet<String>(); + Set<String> topoGroups = new HashSet<>(); if (topology_conf.containsKey(Config.TOPOLOGY_GROUPS) && topology_conf.get(Config.TOPOLOGY_GROUPS) != null) { topoGroups.addAll((Collection<String>)topology_conf.get(Config.TOPOLOGY_GROUPS)); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java index 21ee9a6..5731f06 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java @@ -26,15 +26,11 @@ import java.util.Collection; import backtype.storm.security.auth.IAuthorizer; import backtype.storm.security.auth.ReqContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** * An authorization implementation that simply checks a whitelist of users that * are allowed to use the cluster. */ public class SimpleWhitelistAuthorizer implements IAuthorizer { - private static final Logger LOG = LoggerFactory.getLogger(SimpleWhitelistAuthorizer.class); public static final String WHITELIST_USERS_CONF = "storm.auth.simple-white-list.users"; protected Set<String> users; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/digest/ClientCallbackHandler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/digest/ClientCallbackHandler.java b/storm-core/src/jvm/backtype/storm/security/auth/digest/ClientCallbackHandler.java index 3caacaa..420326c 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/digest/ClientCallbackHandler.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/digest/ClientCallbackHandler.java @@ -46,8 +46,6 @@ public class ClientCallbackHandler implements CallbackHandler { * Constructor based on a JAAS configuration * * For digest, you should have a pair of user name and password defined. - * - * @param configuration * @throws IOException */ public ClientCallbackHandler(Configuration configuration) throws IOException { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/digest/DigestSaslTransportPlugin.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/digest/DigestSaslTransportPlugin.java b/storm-core/src/jvm/backtype/storm/security/auth/digest/DigestSaslTransportPlugin.java index ad642d8..09d6f78 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/digest/DigestSaslTransportPlugin.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/digest/DigestSaslTransportPlugin.java @@ -18,10 +18,8 @@ package backtype.storm.security.auth.digest; import java.io.IOException; -import java.util.Map; import javax.security.auth.callback.CallbackHandler; -import javax.security.auth.login.Configuration; import org.apache.thrift.transport.TSaslClientTransport; import org.apache.thrift.transport.TSaslServerTransport; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/security/auth/digest/ServerCallbackHandler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/security/auth/digest/ServerCallbackHandler.java b/storm-core/src/jvm/backtype/storm/security/auth/digest/ServerCallbackHandler.java index 1788dab..e80072c 100644 --- a/storm-core/src/jvm/backtype/storm/security/auth/digest/ServerCallbackHandler.java +++ b/storm-core/src/jvm/backtype/storm/security/auth/digest/ServerCallbackHandler.java @@ -26,7 +26,6 @@ import backtype.storm.security.auth.SaslTransportPlugin; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.security.auth.Subject; import javax.security.auth.callback.Callback; import javax.security.auth.callback.CallbackHandler; import javax.security.auth.callback.NameCallback; @@ -40,7 +39,7 @@ import javax.security.sasl.RealmCallback; import backtype.storm.security.auth.AuthUtils; /** - * SASL server side collback handler + * SASL server side callback handler */ public class ServerCallbackHandler implements CallbackHandler { private static final String USER_PREFIX = "user_"; @@ -48,7 +47,7 @@ public class ServerCallbackHandler implements CallbackHandler { private static final String SYSPROP_SUPER_PASSWORD = "storm.SASLAuthenticationProvider.superPassword"; private String userName; - private final Map<String,String> credentials = new HashMap<String,String>(); + private final Map<String,String> credentials = new HashMap<>(); public ServerCallbackHandler(Configuration configuration) throws IOException { if (configuration==null) return; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/serialization/GzipThriftSerializationDelegate.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/serialization/GzipThriftSerializationDelegate.java b/storm-core/src/jvm/backtype/storm/serialization/GzipThriftSerializationDelegate.java index 933a125..54193da 100644 --- a/storm-core/src/jvm/backtype/storm/serialization/GzipThriftSerializationDelegate.java +++ b/storm-core/src/jvm/backtype/storm/serialization/GzipThriftSerializationDelegate.java @@ -17,7 +17,6 @@ */ package backtype.storm.serialization; -import java.io.IOException; import java.util.Map; import backtype.storm.utils.Utils; import org.apache.thrift.TBase; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/serialization/ITupleDeserializer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/serialization/ITupleDeserializer.java b/storm-core/src/jvm/backtype/storm/serialization/ITupleDeserializer.java index 4e68658..09cc422 100644 --- a/storm-core/src/jvm/backtype/storm/serialization/ITupleDeserializer.java +++ b/storm-core/src/jvm/backtype/storm/serialization/ITupleDeserializer.java @@ -18,7 +18,6 @@ package backtype.storm.serialization; import backtype.storm.tuple.Tuple; -import java.io.IOException; public interface ITupleDeserializer { Tuple deserialize(byte[] ser); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/serialization/KryoTupleDeserializer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/serialization/KryoTupleDeserializer.java b/storm-core/src/jvm/backtype/storm/serialization/KryoTupleDeserializer.java index 5a5e3a4..07b9cd5 100644 --- a/storm-core/src/jvm/backtype/storm/serialization/KryoTupleDeserializer.java +++ b/storm-core/src/jvm/backtype/storm/serialization/KryoTupleDeserializer.java @@ -21,10 +21,7 @@ import backtype.storm.task.GeneralTopologyContext; import backtype.storm.tuple.MessageId; import backtype.storm.tuple.Tuple; import backtype.storm.tuple.TupleImpl; -import backtype.storm.utils.WritableUtils; import com.esotericsoftware.kryo.io.Input; -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; import java.io.IOException; import java.util.List; import java.util.Map; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/serialization/KryoValuesDeserializer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/serialization/KryoValuesDeserializer.java b/storm-core/src/jvm/backtype/storm/serialization/KryoValuesDeserializer.java index 209ae53..418068e 100644 --- a/storm-core/src/jvm/backtype/storm/serialization/KryoValuesDeserializer.java +++ b/storm-core/src/jvm/backtype/storm/serialization/KryoValuesDeserializer.java @@ -21,7 +21,6 @@ import backtype.storm.utils.ListDelegate; import com.esotericsoftware.kryo.Kryo; import com.esotericsoftware.kryo.io.Input; import java.io.IOException; -import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -35,7 +34,7 @@ public class KryoValuesDeserializer { } public List<Object> deserializeFrom(Input input) { - ListDelegate delegate = (ListDelegate) _kryo.readObject(input, ListDelegate.class); + ListDelegate delegate = _kryo.readObject(input, ListDelegate.class); return delegate.getDelegate(); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/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 2a829d1..417f102 100644 --- a/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java +++ b/storm-core/src/jvm/backtype/storm/serialization/SerializationFactory.java @@ -21,7 +21,6 @@ import backtype.storm.Config; import backtype.storm.generated.ComponentCommon; import backtype.storm.generated.StormTopology; import backtype.storm.serialization.types.ArrayListSerializer; -import backtype.storm.serialization.types.ListDelegateSerializer; import backtype.storm.serialization.types.HashMapSerializer; import backtype.storm.serialization.types.HashSetSerializer; import backtype.storm.transactional.TransactionAttempt; @@ -130,17 +129,17 @@ public class SerializationFactory { } public static class IdDictionary { - Map<String, Map<String, Integer>> streamNametoId = new HashMap<String, Map<String, Integer>>(); - Map<String, Map<Integer, String>> streamIdToName = new HashMap<String, Map<Integer, String>>(); + Map<String, Map<String, Integer>> streamNametoId = new HashMap<>(); + Map<String, Map<Integer, String>> streamIdToName = new HashMap<>(); public IdDictionary(StormTopology topology) { - List<String> componentNames = new ArrayList<String>(topology.get_spouts().keySet()); + List<String> componentNames = new ArrayList<>(topology.get_spouts().keySet()); componentNames.addAll(topology.get_bolts().keySet()); componentNames.addAll(topology.get_state_spouts().keySet()); for(String name: componentNames) { ComponentCommon common = Utils.getComponentCommon(topology, name); - List<String> streams = new ArrayList<String>(common.get_streams().keySet()); + List<String> streams = new ArrayList<>(common.get_streams().keySet()); streamNametoId.put(name, idify(streams)); streamIdToName.put(name, Utils.reverseMap(streamNametoId.get(name))); } @@ -156,7 +155,7 @@ public class SerializationFactory { private static Map<String, Integer> idify(List<String> names) { Collections.sort(names); - Map<String, Integer> ret = new HashMap<String, Integer>(); + Map<String, Integer> ret = new HashMap<>(); int i = 1; for(String name: names) { ret.put(name, i); @@ -204,8 +203,8 @@ public class SerializationFactory { private static Map<String, String> normalizeKryoRegister(Map conf) { // TODO: de-duplicate this logic with the code in nimbus Object res = conf.get(Config.TOPOLOGY_KRYO_REGISTER); - if(res==null) return new TreeMap<String, String>(); - Map<String, String> ret = new HashMap<String, String>(); + if(res==null) return new TreeMap<>(); + Map<String, String> ret = new HashMap<>(); if(res instanceof Map) { ret = (Map<String, String>) res; } else { @@ -219,6 +218,6 @@ public class SerializationFactory { } //ensure always same order for registrations with TreeMap - return new TreeMap<String, String>(ret); + return new TreeMap<>(ret); } } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/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 54de390..11bdc19 100644 --- a/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java +++ b/storm-core/src/jvm/backtype/storm/task/GeneralTopologyContext.java @@ -100,8 +100,8 @@ public class GeneralTopologyContext implements JSONAware { */ public List<Integer> getComponentTasks(String componentId) { List<Integer> ret = _componentToTasks.get(componentId); - if(ret==null) return new ArrayList<Integer>(); - else return new ArrayList<Integer>(ret); + if(ret==null) return new ArrayList<>(); + else return new ArrayList<>(ret); } /** @@ -138,14 +138,14 @@ public class GeneralTopologyContext implements JSONAware { * @return Map from stream id to component id to the Grouping used. */ public Map<String, Map<String, Grouping>> getTargets(String componentId) { - Map<String, Map<String, Grouping>> ret = new HashMap<String, Map<String, Grouping>>(); + Map<String, Map<String, Grouping>> ret = new HashMap<>(); for(String otherComponentId: getComponentIds()) { Map<GlobalStreamId, Grouping> inputs = getComponentCommon(otherComponentId).get_inputs(); 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>(); + if(curr==null) curr = new HashMap<>(); curr.put(otherComponentId, entry.getValue()); ret.put(id.get_streamId(), curr); } @@ -196,4 +196,4 @@ public class GeneralTopologyContext implements JSONAware { } return max; } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/task/TopologyContext.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java index cefa207..e9cdb5b 100644 --- a/storm-core/src/jvm/backtype/storm/task/TopologyContext.java +++ b/storm-core/src/jvm/backtype/storm/task/TopologyContext.java @@ -51,8 +51,8 @@ import org.json.simple.JSONValue; */ public class TopologyContext extends WorkerTopologyContext implements IMetricsContext { private Integer _taskId; - private Map<String, Object> _taskData = new HashMap<String, Object>(); - private List<ITaskHook> _hooks = new ArrayList<ITaskHook>(); + private Map<String, Object> _taskData = new HashMap<>(); + private List<ITaskHook> _hooks = new ArrayList<>(); private Map<String, Object> _executorData; private Map<Integer,Map<Integer, Map<String, IMetric>>> _registeredMetrics; private clojure.lang.Atom _openOrPrepareWasCalled; @@ -139,9 +139,8 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo } /** - * Gets the component id for this task. The component id maps + * @return the component id for this task. The component id maps * to a component id specified for a Spout or Bolt in the topology definition. - * @return */ public String getThisComponentId() { return getComponentId(_taskId); @@ -308,7 +307,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo * @return The IMetric argument unchanged. */ public <T extends IMetric> T registerMetric(String name, T metric, int timeBucketSizeInSecs) { - if((Boolean)_openOrPrepareWasCalled.deref() == true) { + if((Boolean) _openOrPrepareWasCalled.deref()) { throw new RuntimeException("TopologyContext.registerMetric can only be called from within overridden " + "IBolt::prepare() or ISpout::open() method."); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/topology/BasicBoltExecutor.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/topology/BasicBoltExecutor.java b/storm-core/src/jvm/backtype/storm/topology/BasicBoltExecutor.java index 6c9cdc1..cf828fe 100644 --- a/storm-core/src/jvm/backtype/storm/topology/BasicBoltExecutor.java +++ b/storm-core/src/jvm/backtype/storm/topology/BasicBoltExecutor.java @@ -25,7 +25,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class BasicBoltExecutor implements IRichBolt { - public static Logger LOG = LoggerFactory.getLogger(BasicBoltExecutor.class); + public static final Logger LOG = LoggerFactory.getLogger(BasicBoltExecutor.class); private IBasicBolt _bolt; private transient BasicOutputCollector _collector; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/topology/OutputFieldsGetter.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/topology/OutputFieldsGetter.java b/storm-core/src/jvm/backtype/storm/topology/OutputFieldsGetter.java index 0e7fd59..26a791e 100644 --- a/storm-core/src/jvm/backtype/storm/topology/OutputFieldsGetter.java +++ b/storm-core/src/jvm/backtype/storm/topology/OutputFieldsGetter.java @@ -24,7 +24,7 @@ import java.util.HashMap; import java.util.Map; public class OutputFieldsGetter implements OutputFieldsDeclarer { - private Map<String, StreamInfo> _fields = new HashMap<String, StreamInfo>(); + private Map<String, StreamInfo> _fields = new HashMap<>(); public void declare(Fields fields) { declare(false, fields); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java b/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java index 38b30d7..bccb1bf 100644 --- a/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java +++ b/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java @@ -89,18 +89,18 @@ import org.json.simple.JSONValue; * the inputs for that component.</p> */ public class TopologyBuilder { - private Map<String, IRichBolt> _bolts = new HashMap<String, IRichBolt>(); - private Map<String, IRichSpout> _spouts = new HashMap<String, IRichSpout>(); - private Map<String, ComponentCommon> _commons = new HashMap<String, ComponentCommon>(); + private Map<String, IRichBolt> _bolts = new HashMap<>(); + private Map<String, IRichSpout> _spouts = new HashMap<>(); + private Map<String, ComponentCommon> _commons = new HashMap<>(); // private Map<String, Map<GlobalStreamId, Grouping>> _inputs = new HashMap<String, Map<GlobalStreamId, Grouping>>(); - private Map<String, StateSpoutSpec> _stateSpouts = new HashMap<String, StateSpoutSpec>(); + private Map<String, StateSpoutSpec> _stateSpouts = new HashMap<>(); public StormTopology createTopology() { - Map<String, Bolt> boltSpecs = new HashMap<String, Bolt>(); - Map<String, SpoutSpec> spoutSpecs = new HashMap<String, SpoutSpec>(); + Map<String, Bolt> boltSpecs = new HashMap<>(); + Map<String, SpoutSpec> spoutSpecs = new HashMap<>(); for(String boltId: _bolts.keySet()) { IRichBolt bolt = _bolts.get(boltId); ComponentCommon common = getComponentCommon(boltId, bolt); @@ -168,7 +168,7 @@ public class TopologyBuilder { * * @param id the id of this component. This id is referenced by other components that want to consume this bolt's outputs. * @param bolt the basic bolt - * @param parallelism_hint the number of tasks that should be assigned to execute this bolt. Each task will run on a thread in a process somwehere around the cluster. + * @param parallelism_hint the number of tasks that should be assigned to execute this bolt. Each task will run on a thread in a process somewhere around the cluster. * @return use the returned object to declare the inputs to this component * @throws IllegalArgumentException if {@code parallelism_hint} is not positive */ @@ -193,7 +193,7 @@ public class TopologyBuilder { * will be allocated to this component. * * @param id the id of this component. This id is referenced by other components that want to consume this spout's outputs. - * @param parallelism_hint the number of tasks that should be assigned to execute this spout. Each task will run on a thread in a process somwehere around the cluster. + * @param parallelism_hint the number of tasks that should be assigned to execute this spout. Each task will run on a thread in a process somewhere around the cluster. * @param spout the spout * @throws IllegalArgumentException if {@code parallelism_hint} is not positive */ http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/topology/base/BaseBatchBolt.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/topology/base/BaseBatchBolt.java b/storm-core/src/jvm/backtype/storm/topology/base/BaseBatchBolt.java index 3206941..3f35149 100644 --- a/storm-core/src/jvm/backtype/storm/topology/base/BaseBatchBolt.java +++ b/storm-core/src/jvm/backtype/storm/topology/base/BaseBatchBolt.java @@ -18,7 +18,6 @@ package backtype.storm.topology.base; import backtype.storm.coordination.IBatchBolt; -import java.util.Map; public abstract class BaseBatchBolt<T> extends BaseComponent implements IBatchBolt<T> { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/topology/base/BaseTransactionalSpout.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/topology/base/BaseTransactionalSpout.java b/storm-core/src/jvm/backtype/storm/topology/base/BaseTransactionalSpout.java index 704a95b..e2b3a93 100644 --- a/storm-core/src/jvm/backtype/storm/topology/base/BaseTransactionalSpout.java +++ b/storm-core/src/jvm/backtype/storm/topology/base/BaseTransactionalSpout.java @@ -18,7 +18,6 @@ package backtype.storm.topology.base; import backtype.storm.transactional.ITransactionalSpout; -import java.util.Map; public abstract class BaseTransactionalSpout<T> extends BaseComponent implements ITransactionalSpout<T> { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java b/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java index f7ce534..f8f73f6 100644 --- a/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java +++ b/storm-core/src/jvm/backtype/storm/transactional/TransactionalSpoutCoordinator.java @@ -52,7 +52,7 @@ public class TransactionalSpoutCoordinator extends BaseRichSpout { private TransactionalState _state; private RotatingTransactionalState _coordinatorState; - TreeMap<BigInteger, TransactionStatus> _activeTx = new TreeMap<BigInteger, TransactionStatus>(); + TreeMap<BigInteger, TransactionStatus> _activeTx = new TreeMap<>(); private SpoutOutputCollector _collector; private Random _rand; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/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 3be5d37..02185f4 100644 --- a/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java +++ b/storm-core/src/jvm/backtype/storm/transactional/partitioned/OpaquePartitionedTransactionalSpoutExecutor.java @@ -63,8 +63,8 @@ public class OpaquePartitionedTransactionalSpoutExecutor implements ICommitterTr public class Emitter implements ICommitterTransactionalSpout.Emitter { IOpaquePartitionedTransactionalSpout.Emitter _emitter; TransactionalState _state; - TreeMap<BigInteger, Map<Integer, Object>> _cachedMetas = new TreeMap<BigInteger, Map<Integer, Object>>(); - Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<Integer, RotatingTransactionalState>(); + TreeMap<BigInteger, Map<Integer, Object>> _cachedMetas = new TreeMap<>(); + Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<>(); int _index; int _numTasks; @@ -84,7 +84,7 @@ public class OpaquePartitionedTransactionalSpoutExecutor implements ICommitterTr @Override public void emitBatch(TransactionAttempt tx, Object coordinatorMeta, BatchOutputCollector collector) { - Map<Integer, Object> metas = new HashMap<Integer, Object>(); + Map<Integer, Object> metas = new HashMap<>(); _cachedMetas.put(tx.getTransactionId(), metas); int partitions = _emitter.numPartitions(); Entry<BigInteger, Map<Integer, Object>> entry = _cachedMetas.lowerEntry(tx.getTransactionId()); @@ -92,7 +92,7 @@ public class OpaquePartitionedTransactionalSpoutExecutor implements ICommitterTr if(entry!=null) { prevCached = entry.getValue(); } else { - prevCached = new HashMap<Integer, Object>(); + prevCached = new HashMap<>(); } for(int i=_index; i < partitions; i+=_numTasks) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java b/storm-core/src/jvm/backtype/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java index 479dda4..76859cf 100644 --- a/storm-core/src/jvm/backtype/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java +++ b/storm-core/src/jvm/backtype/storm/transactional/partitioned/PartitionedTransactionalSpoutExecutor.java @@ -67,7 +67,7 @@ public class PartitionedTransactionalSpoutExecutor implements ITransactionalSpou class Emitter implements ITransactionalSpout.Emitter<Integer> { private IPartitionedTransactionalSpout.Emitter _emitter; private TransactionalState _state; - private Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<Integer, RotatingTransactionalState>(); + private Map<Integer, RotatingTransactionalState> _partitionStates = new HashMap<>(); private int _index; private int _numTasks; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/tuple/Fields.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/tuple/Fields.java b/storm-core/src/jvm/backtype/storm/tuple/Fields.java index 3eed409..b52b798 100644 --- a/storm-core/src/jvm/backtype/storm/tuple/Fields.java +++ b/storm-core/src/jvm/backtype/storm/tuple/Fields.java @@ -30,14 +30,14 @@ import java.io.Serializable; */ public class Fields implements Iterable<String>, Serializable { private List<String> _fields; - private Map<String, Integer> _index = new HashMap<String, Integer>(); + private Map<String, Integer> _index = new HashMap<>(); public Fields(String... fields) { this(Arrays.asList(fields)); } public Fields(List<String> fields) { - _fields = new ArrayList<String>(fields.size()); + _fields = new ArrayList<>(fields.size()); for (String field : fields) { if (_fields.contains(field)) throw new IllegalArgumentException( @@ -49,7 +49,7 @@ public class Fields implements Iterable<String>, Serializable { } public List<Object> select(Fields selector, List<Object> tuple) { - List<Object> ret = new ArrayList<Object>(selector.size()); + List<Object> ret = new ArrayList<>(selector.size()); for(String s: selector) { ret.add(tuple.get(_index.get(s))); } @@ -57,7 +57,7 @@ public class Fields implements Iterable<String>, Serializable { } public List<String> toList() { - return new ArrayList<String>(_fields); + return new ArrayList<>(_fields); } /** @@ -98,7 +98,7 @@ public class Fields implements Iterable<String>, Serializable { } /** - * @returns true if this contains the specified name of the field. + * @return true if this contains the specified name of the field. */ public boolean contains(String field) { return _index.containsKey(field); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/tuple/MessageId.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/tuple/MessageId.java b/storm-core/src/jvm/backtype/storm/tuple/MessageId.java index 680af38..554bab6 100644 --- a/storm-core/src/jvm/backtype/storm/tuple/MessageId.java +++ b/storm-core/src/jvm/backtype/storm/tuple/MessageId.java @@ -43,7 +43,7 @@ public class MessageId { } public static MessageId makeRootId(long id, long val) { - Map<Long, Long> anchorsToIds = new HashMap<Long, Long>(); + Map<Long, Long> anchorsToIds = new HashMap<>(); anchorsToIds.put(id, val); return new MessageId(anchorsToIds); } @@ -67,11 +67,7 @@ public class MessageId { @Override public boolean equals(Object other) { - if(other instanceof MessageId) { - return _anchorsToIds.equals(((MessageId) other)._anchorsToIds); - } else { - return false; - } + return other instanceof MessageId && _anchorsToIds.equals(((MessageId) other)._anchorsToIds); } @Override @@ -89,7 +85,7 @@ public class MessageId { public static MessageId deserialize(Input in) throws IOException { int numAnchors = in.readInt(true); - Map<Long, Long> anchorsToIds = new HashMap<Long, Long>(); + Map<Long, Long> anchorsToIds = new HashMap<>(); for(int i=0; i<numAnchors; i++) { anchorsToIds.put(in.readLong(), in.readLong()); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/tuple/Tuple.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/tuple/Tuple.java b/storm-core/src/jvm/backtype/storm/tuple/Tuple.java index c4ea7c8..19d0cb4 100644 --- a/storm-core/src/jvm/backtype/storm/tuple/Tuple.java +++ b/storm-core/src/jvm/backtype/storm/tuple/Tuple.java @@ -18,7 +18,6 @@ package backtype.storm.tuple; import backtype.storm.generated.GlobalStreamId; -import java.util.List; /** * The tuple is the main data structure in Storm. A tuple is a named list of values, http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/tuple/TupleImpl.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/tuple/TupleImpl.java b/storm-core/src/jvm/backtype/storm/tuple/TupleImpl.java index 53b63ba..dd31c96 100644 --- a/storm-core/src/jvm/backtype/storm/tuple/TupleImpl.java +++ b/storm-core/src/jvm/backtype/storm/tuple/TupleImpl.java @@ -237,7 +237,7 @@ public class TupleImpl extends IndifferentAccessMap implements Seqable, Indexed, return System.identityHashCode(this); } - private final Keyword makeKeyword(String name) { + private Keyword makeKeyword(String name) { return Keyword.intern(Symbol.create(name)); } @@ -250,7 +250,7 @@ public class TupleImpl extends IndifferentAccessMap implements Seqable, Indexed, } else if(o instanceof String) { return getValueByField((String) o); } - } catch(IllegalArgumentException e) { + } catch(IllegalArgumentException ignored) { } return null; } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/DRPCClient.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/DRPCClient.java b/storm-core/src/jvm/backtype/storm/utils/DRPCClient.java index b2a2a7d..23deb28 100644 --- a/storm-core/src/jvm/backtype/storm/utils/DRPCClient.java +++ b/storm-core/src/jvm/backtype/storm/utils/DRPCClient.java @@ -17,7 +17,6 @@ */ package backtype.storm.utils; -import backtype.storm.Config; import backtype.storm.generated.DRPCExecutionException; import backtype.storm.generated.DistributedRPC; import backtype.storm.generated.AuthorizationException; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java b/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java index 246f9a7..7ad6ae0 100644 --- a/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java +++ b/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java @@ -34,14 +34,11 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.Map; -import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.concurrent.locks.ReentrantLock; import org.slf4j.Logger; @@ -255,7 +252,7 @@ public class DisruptorQueue implements IStatefulObject { public DisruptorQueue(String queueName, ProducerType type, int size, long readTimeout, int inputBatchSize, long flushInterval) { this._queueName = PREFIX + queueName; - WaitStrategy wait = null; + WaitStrategy wait; if (readTimeout <= 0) { wait = new LiteBlockingWaitStrategy(); } else { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/utils/InprocMessaging.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/utils/InprocMessaging.java b/storm-core/src/jvm/backtype/storm/utils/InprocMessaging.java index b20c775..4abbc3e 100644 --- a/storm-core/src/jvm/backtype/storm/utils/InprocMessaging.java +++ b/storm-core/src/jvm/backtype/storm/utils/InprocMessaging.java @@ -22,7 +22,7 @@ import java.util.Map; import java.util.concurrent.LinkedBlockingQueue; public class InprocMessaging { - private static Map<Integer, LinkedBlockingQueue<Object>> _queues = new HashMap<Integer, LinkedBlockingQueue<Object>>(); + private static Map<Integer, LinkedBlockingQueue<Object>> _queues = new HashMap<>(); private static final Object _lock = new Object(); private static int port = 1; @@ -50,7 +50,7 @@ public class InprocMessaging { private static LinkedBlockingQueue<Object> getQueue(int port) { synchronized(_lock) { if(!_queues.containsKey(port)) { - _queues.put(port, new LinkedBlockingQueue<Object>()); + _queues.put(port, new LinkedBlockingQueue<>()); } return _queues.get(port); }
