STORM-1164. Code cleanup for typos, warnings and conciseness
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f0e4c6f2 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f0e4c6f2 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f0e4c6f2 Branch: refs/heads/master Commit: f0e4c6f2099196760f0037e7812dda543d332c54 Parents: c12e28c Author: Suresh Srinivas <[email protected]> Authored: Tue Nov 3 16:03:16 2015 -0800 Committer: Suresh Srinivas <[email protected]> Committed: Mon Nov 9 11:56:11 2015 -0800 ---------------------------------------------------------------------- storm-core/src/jvm/backtype/storm/Config.java | 36 ++++---- .../src/jvm/backtype/storm/LogWriter.java | 2 +- .../src/jvm/backtype/storm/StormSubmitter.java | 4 - .../storm/drpc/DRPCInvocationsClient.java | 5 +- .../storm/generated/AlreadyAliveException.java | 7 +- .../storm/grouping/PartialKeyGrouping.java | 5 +- .../storm/messaging/ConnectionWithStatus.java | 4 +- .../jvm/backtype/storm/messaging/IContext.java | 2 +- .../storm/messaging/TransportFactory.java | 2 +- .../backtype/storm/messaging/netty/Client.java | 2 +- .../backtype/storm/messaging/netty/Context.java | 8 +- .../storm/messaging/netty/ControlMessage.java | 5 +- .../storm/messaging/netty/MessageBatch.java | 14 ++-- .../storm/messaging/netty/MessageDecoder.java | 7 +- .../storm/messaging/netty/SaslMessageToken.java | 10 +-- .../storm/messaging/netty/SaslNettyClient.java | 6 +- .../storm/messaging/netty/SaslNettyServer.java | 4 - .../messaging/netty/SaslStormClientHandler.java | 4 +- .../messaging/netty/SaslStormServerHandler.java | 11 +-- .../storm/messaging/netty/SaslUtils.java | 11 +-- .../backtype/storm/messaging/netty/Server.java | 26 +++--- .../backtype/storm/metric/EventLoggerBolt.java | 8 -- .../storm/metric/FileBasedEventLogger.java | 2 +- .../metric/HttpForwardingMetricsConsumer.java | 1 - .../metric/HttpForwardingMetricsServer.java | 1 - .../jvm/backtype/storm/metric/IEventLogger.java | 8 +- .../storm/metric/LoggingMetricsConsumer.java | 1 - .../storm/metric/MetricsConsumerBolt.java | 1 - .../jvm/backtype/storm/metric/SystemBolt.java | 5 -- .../backtype/storm/metric/api/CountMetric.java | 2 - .../backtype/storm/metric/api/MeanReducer.java | 4 +- .../storm/metric/api/MultiCountMetric.java | 2 +- .../storm/metric/api/MultiReducedMetric.java | 2 +- .../storm/metric/api/rpc/CountShellMetric.java | 3 +- .../AbstractDNSToSwitchMapping.java | 2 +- .../DefaultRackDNSToSwitchMapping.java | 4 +- .../backtype/storm/nimbus/ILeaderElector.java | 6 +- .../jvm/backtype/storm/nimbus/NimbusInfo.java | 4 +- .../jvm/backtype/storm/scheduler/Cluster.java | 5 +- .../scheduler/SchedulerAssignmentImpl.java | 15 ++-- .../storm/scheduler/SupervisorDetails.java | 6 +- .../backtype/storm/scheduler/Topologies.java | 6 +- .../storm/scheduler/TopologyDetails.java | 30 +++---- .../scheduler/multitenant/DefaultPool.java | 22 ++--- .../storm/scheduler/multitenant/FreePool.java | 6 +- .../scheduler/multitenant/IsolatedPool.java | 32 ++++--- .../multitenant/MultitenantScheduler.java | 6 +- .../storm/scheduler/multitenant/Node.java | 17 ++-- .../storm/scheduler/multitenant/NodePool.java | 16 ++-- .../strategies/ResourceAwareStrategy.java | 39 +++++---- .../backtype/storm/security/auth/AuthUtils.java | 27 +++--- .../auth/DefaultHttpCredentialsPlugin.java | 6 +- .../security/auth/DefaultPrincipalToLocal.java | 1 - .../storm/security/auth/IAuthorizer.java | 4 +- .../security/auth/ICredentialsRenewer.java | 3 +- .../security/auth/IHttpCredentialsPlugin.java | 2 - .../storm/security/auth/IPrincipalToLocal.java | 2 +- .../storm/security/auth/ITransportPlugin.java | 4 - .../security/auth/KerberosPrincipalToLocal.java | 2 +- .../storm/security/auth/ReqContext.java | 11 +-- .../security/auth/SaslTransportPlugin.java | 12 +-- .../security/auth/SimpleTransportPlugin.java | 6 +- .../security/auth/SingleUserPrincipal.java | 5 +- .../storm/security/auth/TBackoffConnect.java | 1 - .../storm/security/auth/ThriftClient.java | 10 +-- .../storm/security/auth/ThriftServer.java | 6 +- .../authorizer/DRPCSimpleACLAuthorizer.java | 11 ++- .../auth/authorizer/DenyAuthorizer.java | 11 +-- .../authorizer/ImpersonationAuthorizer.java | 10 +-- .../auth/authorizer/NoopAuthorizer.java | 7 +- .../auth/authorizer/SimpleACLAuthorizer.java | 18 ++-- .../authorizer/SimpleWhitelistAuthorizer.java | 4 - .../auth/digest/ClientCallbackHandler.java | 2 - .../auth/digest/DigestSaslTransportPlugin.java | 2 - .../auth/digest/ServerCallbackHandler.java | 5 +- .../GzipThriftSerializationDelegate.java | 1 - .../storm/serialization/ITupleDeserializer.java | 1 - .../serialization/KryoTupleDeserializer.java | 3 - .../serialization/KryoValuesDeserializer.java | 3 +- .../serialization/SerializationFactory.java | 17 ++-- .../storm/task/GeneralTopologyContext.java | 10 +-- .../backtype/storm/task/TopologyContext.java | 9 +- .../storm/topology/BasicBoltExecutor.java | 2 +- .../storm/topology/OutputFieldsGetter.java | 2 +- .../storm/topology/TopologyBuilder.java | 16 ++-- .../storm/topology/base/BaseBatchBolt.java | 1 - .../topology/base/BaseTransactionalSpout.java | 1 - .../TransactionalSpoutCoordinator.java | 2 +- ...uePartitionedTransactionalSpoutExecutor.java | 8 +- .../PartitionedTransactionalSpoutExecutor.java | 2 +- .../src/jvm/backtype/storm/tuple/Fields.java | 10 +-- .../src/jvm/backtype/storm/tuple/MessageId.java | 10 +-- .../src/jvm/backtype/storm/tuple/Tuple.java | 1 - .../src/jvm/backtype/storm/tuple/TupleImpl.java | 4 +- .../jvm/backtype/storm/utils/DRPCClient.java | 1 - .../backtype/storm/utils/DisruptorQueue.java | 5 +- .../backtype/storm/utils/InprocMessaging.java | 4 +- .../storm/utils/KeyedRoundRobinQueue.java | 6 +- .../jvm/backtype/storm/utils/ListDelegate.java | 6 +- .../src/jvm/backtype/storm/utils/Monitor.java | 3 +- .../jvm/backtype/storm/utils/NimbusClient.java | 10 +-- .../storm/utils/RegisteredGlobalState.java | 6 +- .../jvm/backtype/storm/utils/RotatingMap.java | 2 +- .../backtype/storm/utils/ServiceRegistry.java | 2 +- .../StormBoundedExponentialBackoffRetry.java | 3 +- .../jvm/backtype/storm/utils/VersionInfo.java | 2 +- .../src/jvm/storm/trident/TridentTopology.java | 87 ++++++++++---------- .../trident/drpc/ReturnResultsReducer.java | 4 +- .../fluent/ChainedAggregatorDeclarer.java | 8 +- .../jvm/storm/trident/graph/GraphGrouper.java | 13 ++- .../src/jvm/storm/trident/graph/Group.java | 23 +++--- .../trident/operation/builtin/SnapshotGet.java | 4 +- .../operation/builtin/TupleCollectionGet.java | 6 +- .../storm/trident/partition/GlobalGrouping.java | 5 +- .../trident/partition/IdentityGrouping.java | 8 +- .../src/jvm/storm/trident/planner/Node.java | 5 +- .../storm/trident/planner/PartitionNode.java | 2 - .../storm/trident/planner/SubtopologyBolt.java | 19 ++--- .../processor/MultiReducerProcessor.java | 2 +- .../trident/spout/TridentSpoutExecutor.java | 2 +- .../storm/trident/tuple/TridentTupleView.java | 6 +- .../storm/utils/DisruptorQueueTest.java | 7 +- 122 files changed, 378 insertions(+), 579 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/Config.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java index d80876e..1485361 100644 --- a/storm-core/src/jvm/backtype/storm/Config.java +++ b/storm-core/src/jvm/backtype/storm/Config.java @@ -166,7 +166,7 @@ public class Config extends HashMap<String, Object> { public static final String STORM_LOG4J2_CONF_DIR = "storm.log4j2.conf.dir"; /** - * A global task scheduler used to assign topologies's tasks to supervisors' wokers. + * A global task scheduler used to assign topologies's tasks to supervisors' workers. * * If this is not set, a default system scheduler will be used. */ @@ -192,7 +192,7 @@ public class Config extends HashMap<String, Object> { * The hostname the supervisors/workers should report to nimbus. If unset, Storm will * get the hostname to report by calling <code>InetAddress.getLocalHost().getCanonicalHostName()</code>. * - * You should set this config when you dont have a DNS which supervisors/workers + * You should set this config when you don't have a DNS which supervisors/workers * can utilize to find each other based on hostname got from calls to * <code>InetAddress.getLocalHost().getCanonicalHostName()</code>. */ @@ -445,7 +445,7 @@ public class Config extends HashMap<String, Object> { * How often nimbus should wake up to check heartbeats and do reassignments. Note * that if a machine ever goes down Nimbus will immediately wake up and take action. * This parameter is for checking for failures when there's no explicit event like that - * occuring. + * occurring. */ @isInteger @isPositiveNumber @@ -453,7 +453,7 @@ public class Config extends HashMap<String, Object> { /** * How often nimbus should wake the cleanup thread to clean the inbox. - * @see NIMBUS_INBOX_JAR_EXPIRATION_SECS + * @see #NIMBUS_INBOX_JAR_EXPIRATION_SECS */ @isInteger @isPositiveNumber @@ -466,7 +466,7 @@ public class Config extends HashMap<String, Object> { * Note that the time it takes to delete an inbox jar file is going to be somewhat more than * NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS (depending on how often NIMBUS_CLEANUP_FREQ_SECS * is set to). - * @see NIMBUS_CLEANUP_FREQ_SECS + * @see #NIMBUS_CLEANUP_INBOX_FREQ_SECS */ @isInteger public static final String NIMBUS_INBOX_JAR_EXPIRATION_SECS = "nimbus.inbox.jar.expiration.secs"; @@ -622,7 +622,7 @@ public class Config extends HashMap<String, Object> { public static final String LOGVIEWER_HTTPS_KEYSTORE_TYPE = "logviewer.https.keystore.type"; /** - * Password to the private key in the keystore for settting up HTTPS (SSL). + * Password to the private key in the keystore for setting up HTTPS (SSL). */ @isString public static final String LOGVIEWER_HTTPS_KEY_PASSWORD = "logviewer.https.key.password"; @@ -647,7 +647,7 @@ public class Config extends HashMap<String, Object> { public static final String LOGVIEWER_HTTPS_TRUSTSTORE_TYPE = "logviewer.https.truststore.type"; /** - * Password to the truststore used by Storm Logviewer settting up HTTPS (SSL). + * Password to the truststore used by Storm Logviewer setting up HTTPS (SSL). */ @isBoolean public static final String LOGVIEWER_HTTPS_WANT_CLIENT_AUTH = "logviewer.https.want.client.auth"; @@ -725,19 +725,19 @@ public class Config extends HashMap<String, Object> { public static final String UI_HTTPS_KEYSTORE_TYPE = "ui.https.keystore.type"; /** - * Password to the private key in the keystore for settting up HTTPS (SSL). + * Password to the private key in the keystore for setting up HTTPS (SSL). */ @isString public static final String UI_HTTPS_KEY_PASSWORD = "ui.https.key.password"; /** - * Path to the truststore used by Storm UI settting up HTTPS (SSL). + * Path to the truststore used by Storm UI setting up HTTPS (SSL). */ @isString public static final String UI_HTTPS_TRUSTSTORE_PATH = "ui.https.truststore.path"; /** - * Password to the truststore used by Storm UI settting up HTTPS (SSL). + * Password to the truststore used by Storm UI setting up HTTPS (SSL). */ @isString public static final String UI_HTTPS_TRUSTSTORE_PASSWORD = "ui.https.truststore.password"; @@ -750,7 +750,7 @@ public class Config extends HashMap<String, Object> { public static final String UI_HTTPS_TRUSTSTORE_TYPE = "ui.https.truststore.type"; /** - * Password to the truststore used by Storm DRPC settting up HTTPS (SSL). + * Password to the truststore used by Storm DRPC setting up HTTPS (SSL). */ @isBoolean public static final String UI_HTTPS_WANT_CLIENT_AUTH = "ui.https.want.client.auth"; @@ -796,19 +796,19 @@ public class Config extends HashMap<String, Object> { public static final String DRPC_HTTPS_KEYSTORE_TYPE = "drpc.https.keystore.type"; /** - * Password to the private key in the keystore for settting up HTTPS (SSL). + * Password to the private key in the keystore for setting up HTTPS (SSL). */ @isString public static final String DRPC_HTTPS_KEY_PASSWORD = "drpc.https.key.password"; /** - * Path to the truststore used by Storm DRPC settting up HTTPS (SSL). + * Path to the truststore used by Storm DRPC setting up HTTPS (SSL). */ @isString public static final String DRPC_HTTPS_TRUSTSTORE_PATH = "drpc.https.truststore.path"; /** - * Password to the truststore used by Storm DRPC settting up HTTPS (SSL). + * Password to the truststore used by Storm DRPC setting up HTTPS (SSL). */ @isString public static final String DRPC_HTTPS_TRUSTSTORE_PASSWORD = "drpc.https.truststore.password"; @@ -821,7 +821,7 @@ public class Config extends HashMap<String, Object> { public static final String DRPC_HTTPS_TRUSTSTORE_TYPE = "drpc.https.truststore.type"; /** - * Password to the truststore used by Storm DRPC settting up HTTPS (SSL). + * Password to the truststore used by Storm DRPC setting up HTTPS (SSL). */ @isBoolean public static final String DRPC_HTTPS_WANT_CLIENT_AUTH = "drpc.https.want.client.auth"; @@ -850,14 +850,14 @@ public class Config extends HashMap<String, Object> { /** * The Access Control List for the DRPC Authorizer. - * @see DRPCSimpleAclAuthorizer + * @see backtype.storm.security.auth.authorizer.DRPCSimpleACLAuthorizer */ @isType(type=Map.class) public static final String DRPC_AUTHORIZER_ACL = "drpc.authorizer.acl"; /** * File name of the DRPC Authorizer ACL. - * @see DRPCSimpleAclAuthorizer + * @see backtype.storm.security.auth.authorizer.DRPCSimpleACLAuthorizer */ @isString public static final String DRPC_AUTHORIZER_ACL_FILENAME = "drpc.authorizer.acl.filename"; @@ -869,7 +869,7 @@ public class Config extends HashMap<String, Object> { * permitted, which is appropriate for a development environment. When set * to true, explicit ACL entries are required for every DRPC function, and * any request for functions will be denied. - * @see DRPCSimpleAclAuthorizer + * @see backtype.storm.security.auth.authorizer.DRPCSimpleACLAuthorizer */ @isBoolean public static final String DRPC_AUTHORIZER_ACL_STRICT = "drpc.authorizer.acl.strict"; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/LogWriter.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/LogWriter.java b/storm-core/src/jvm/backtype/storm/LogWriter.java index b0857e8..849f5ca 100644 --- a/storm-core/src/jvm/backtype/storm/LogWriter.java +++ b/storm-core/src/jvm/backtype/storm/LogWriter.java @@ -41,7 +41,7 @@ public class LogWriter extends Thread { public void run() { Logger logger = this.logger; BufferedReader in = this.in; - String line = null; + String line; try { while ((line = in.readLine()) != null) { logger.info(line); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/StormSubmitter.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/StormSubmitter.java b/storm-core/src/jvm/backtype/storm/StormSubmitter.java index 27844f2..1d7a1f3 100644 --- a/storm-core/src/jvm/backtype/storm/StormSubmitter.java +++ b/storm-core/src/jvm/backtype/storm/StormSubmitter.java @@ -51,10 +51,6 @@ public class StormSubmitter { private static ILocalCluster localNimbus = null; - public static void setLocalNimbus(ILocalCluster localNimbusHandler) { - StormSubmitter.localNimbus = localNimbusHandler; - } - private static String generateZookeeperDigestSecretPayload() { return Utils.secureRandomLong() + ":" + Utils.secureRandomLong(); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/drpc/DRPCInvocationsClient.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/drpc/DRPCInvocationsClient.java b/storm-core/src/jvm/backtype/storm/drpc/DRPCInvocationsClient.java index 624db3e..78e8d9b 100644 --- a/storm-core/src/jvm/backtype/storm/drpc/DRPCInvocationsClient.java +++ b/storm-core/src/jvm/backtype/storm/drpc/DRPCInvocationsClient.java @@ -31,9 +31,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class DRPCInvocationsClient extends ThriftClient implements DistributedRPCInvocations.Iface { - public static Logger LOG = LoggerFactory.getLogger(DRPCInvocationsClient.class); - private final AtomicReference<DistributedRPCInvocations.Client> client = - new AtomicReference<DistributedRPCInvocations.Client>(); + public static final Logger LOG = LoggerFactory.getLogger(DRPCInvocationsClient.class); + private final AtomicReference<DistributedRPCInvocations.Client> client = new AtomicReference<>(); private String host; private int port; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/generated/AlreadyAliveException.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/generated/AlreadyAliveException.java b/storm-core/src/jvm/backtype/storm/generated/AlreadyAliveException.java index fb2eee3..adc71a2 100644 --- a/storm-core/src/jvm/backtype/storm/generated/AlreadyAliveException.java +++ b/storm-core/src/jvm/backtype/storm/generated/AlreadyAliveException.java @@ -221,11 +221,10 @@ public class AlreadyAliveException extends TException implements org.apache.thri @Override public boolean equals(Object that) { - if (that == null) + if (that == null) { return false; - if (that instanceof AlreadyAliveException) - return this.equals((AlreadyAliveException)that); - return false; + } + return (that instanceof AlreadyAliveException) ? this.equals((AlreadyAliveException)that) : false; } public boolean equals(AlreadyAliveException that) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java b/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java index 43ad5a0..e92156e 100644 --- a/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java +++ b/storm-core/src/jvm/backtype/storm/grouping/PartialKeyGrouping.java @@ -24,7 +24,6 @@ import java.util.Arrays; import java.util.List; import backtype.storm.generated.GlobalStreamId; -import backtype.storm.grouping.CustomStreamGrouping; import backtype.storm.task.WorkerTopologyContext; import backtype.storm.tuple.Fields; @@ -59,9 +58,9 @@ public class PartialKeyGrouping implements CustomStreamGrouping, Serializable { @Override public List<Integer> chooseTasks(int taskId, List<Object> values) { - List<Integer> boltIds = new ArrayList<Integer>(1); + List<Integer> boltIds = new ArrayList<>(1); if (values.size() > 0) { - byte[] raw = null; + byte[] raw; if (fields != null) { List<Object> selectedFields = outFields.select(fields, values); ByteBuffer out = ByteBuffer.allocate(selectedFields.size() * 4); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/ConnectionWithStatus.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/ConnectionWithStatus.java b/storm-core/src/jvm/backtype/storm/messaging/ConnectionWithStatus.java index 37981ca..b92b0d6 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/ConnectionWithStatus.java +++ b/storm-core/src/jvm/backtype/storm/messaging/ConnectionWithStatus.java @@ -39,9 +39,9 @@ public abstract class ConnectionWithStatus implements IConnection { * data sending or receiving. All data sending request will be dropped. */ Closed - }; + } - /** + /** * whether this connection is available to transfer data */ public abstract Status status(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/IContext.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/IContext.java b/storm-core/src/jvm/backtype/storm/messaging/IContext.java index 8645a6f..42f59bd 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/IContext.java +++ b/storm-core/src/jvm/backtype/storm/messaging/IContext.java @@ -56,4 +56,4 @@ public interface IContext { * @return client side connection */ public IConnection connect(String storm_id, String host, int port); -}; +} http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/TransportFactory.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/TransportFactory.java b/storm-core/src/jvm/backtype/storm/messaging/TransportFactory.java index 656b323..c551e0d 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/TransportFactory.java +++ b/storm-core/src/jvm/backtype/storm/messaging/TransportFactory.java @@ -32,7 +32,7 @@ public class TransportFactory { String transport_plugin_klassName = (String)storm_conf.get(Config.STORM_MESSAGING_TRANSPORT); LOG.info("Storm peer transport plugin:"+transport_plugin_klassName); - IContext transport = null; + IContext transport; try { //create a factory class Class klass = Class.forName(transport_plugin_klassName); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java index b9151f3..ccdef41 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Client.java @@ -84,7 +84,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa /** * The channel used for all write operations from this client to the remote destination. */ - private final AtomicReference<Channel> channelRef = new AtomicReference<Channel>(); + private final AtomicReference<Channel> channelRef = new AtomicReference<>(); /** * Total number of connection attempts. http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/Context.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Context.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Context.java index 5d27a16..bc978b6 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/Context.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Context.java @@ -19,12 +19,8 @@ package backtype.storm.messaging.netty; import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; import org.jboss.netty.util.HashedWheelTimer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; import java.util.HashMap; import java.util.Map; @@ -34,8 +30,6 @@ import backtype.storm.messaging.IContext; import backtype.storm.utils.Utils; public class Context implements IContext { - private static final Logger LOG = LoggerFactory.getLogger(Context.class); - @SuppressWarnings("rawtypes") private Map storm_conf; private Map<String, IConnection> connections; @@ -49,7 +43,7 @@ public class Context implements IContext { @SuppressWarnings("rawtypes") public void prepare(Map storm_conf) { this.storm_conf = storm_conf; - connections = new HashMap<String, IConnection>(); + connections = new HashMap<>(); //each context will have a single client channel factory int maxWorkers = Utils.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS)); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/ControlMessage.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/ControlMessage.java b/storm-core/src/jvm/backtype/storm/messaging/netty/ControlMessage.java index fb3efe6..e4507f5 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/ControlMessage.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/ControlMessage.java @@ -39,9 +39,8 @@ enum ControlMessage { } /** - * Return a control message per an encoded status code - * @param encoded - * @return + * @param encoded status code + * @return a control message per an encoded status code */ static ControlMessage mkMessage(short encoded) { for(ControlMessage cm: ControlMessage.values()) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/MessageBatch.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/MessageBatch.java b/storm-core/src/jvm/backtype/storm/messaging/netty/MessageBatch.java index ec0dc0f..fea33fd 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/MessageBatch.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/MessageBatch.java @@ -31,7 +31,7 @@ class MessageBatch { MessageBatch(int buffer_size) { this.buffer_size = buffer_size; - msgs = new ArrayList<TaskMessage>(); + msgs = new ArrayList<>(); encoded_length = ControlMessage.EOB_MESSAGE.encodeLength(); } @@ -54,24 +54,21 @@ class MessageBatch { } /** - * Has this batch used up allowed buffer size - * @return + * @return true if this batch used up allowed buffer size */ boolean isFull() { return encoded_length >= buffer_size; } /** - * true if this batch doesn't have any messages - * @return + * @return true if this batch doesn't have any messages */ boolean isEmpty() { return msgs.isEmpty(); } /** - * # of msgs in this batch - * @return + * @return number of msgs in this batch */ int size() { return msgs.size(); @@ -83,8 +80,9 @@ class MessageBatch { ChannelBuffer buffer() throws Exception { ChannelBufferOutputStream bout = new ChannelBufferOutputStream(ChannelBuffers.directBuffer(encoded_length)); - for (TaskMessage msg : msgs) + for (TaskMessage msg : msgs) { writeTaskMessage(bout, msg); + } //add a END_OF_BATCH indicator ControlMessage.EOB_MESSAGE.write(bout); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/MessageDecoder.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/MessageDecoder.java b/storm-core/src/jvm/backtype/storm/messaging/netty/MessageDecoder.java index 7d8bf54..06d2cde 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/MessageDecoder.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/MessageDecoder.java @@ -43,7 +43,7 @@ public class MessageDecoder extends FrameDecoder { return null; } - List<Object> ret = new ArrayList<Object>(); + List<Object> ret = new ArrayList<>(); // Use while loop, try to decode as more messages as possible in single call while (available >= 2) { @@ -100,7 +100,6 @@ public class MessageDecoder extends FrameDecoder { } // case 3: task Message - short task = code; // Make sure that we have received at least an integer (length) if (available < 4) { @@ -115,7 +114,7 @@ public class MessageDecoder extends FrameDecoder { available -= 4; if (length <= 0) { - ret.add(new TaskMessage(task, null)); + ret.add(new TaskMessage(code, null)); break; } @@ -133,7 +132,7 @@ public class MessageDecoder extends FrameDecoder { // Successfully decoded a frame. // Return a TaskMessage object - ret.add(new TaskMessage(task, payload.array())); + ret.add(new TaskMessage(code, payload.array())); } if (ret.size() == 0) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslMessageToken.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslMessageToken.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslMessageToken.java index d0d3ca1..d7a86d1 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslMessageToken.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslMessageToken.java @@ -20,17 +20,11 @@ package backtype.storm.messaging.netty; import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBufferOutputStream; import org.jboss.netty.buffer.ChannelBuffers; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Send and receive SASL tokens. */ public class SaslMessageToken { - /** Class logger */ - private static final Logger LOG = LoggerFactory - .getLogger(SaslMessageToken.class); - /** Used for client or server's token to send or receive from each other. */ private byte[] token; @@ -88,8 +82,8 @@ public class SaslMessageToken { if (token != null) payload_len = token.length; - bout.writeShort((short) identifier); - bout.writeInt((int) payload_len); + bout.writeShort(identifier); + bout.writeInt(payload_len); if (payload_len > 0) { bout.write(token); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyClient.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyClient.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyClient.java index 023e950..f6ea78f 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyClient.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyClient.java @@ -81,9 +81,7 @@ public class SaslNettyClient { */ public byte[] saslResponse(SaslMessageToken saslTokenMessage) { try { - byte[] retval = saslClient.evaluateChallenge(saslTokenMessage - .getSaslToken()); - return retval; + return saslClient.evaluateChallenge(saslTokenMessage.getSaslToken()); } catch (SaslException e) { LOG.error( "saslResponse: Failed to respond to SASL server's token:", @@ -104,8 +102,6 @@ public class SaslNettyClient { /** * Set private members using topology token. - * - * @param topologyToken */ public SaslClientCallbackHandler(String topologyToken, byte[] token) { this.userName = SaslUtils http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyServer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyServer.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyServer.java index 2cb47d9..f98e5bd 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyServer.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslNettyServer.java @@ -18,9 +18,6 @@ package backtype.storm.messaging.netty; import java.io.IOException; -import java.nio.charset.Charset; -import java.util.HashMap; -import java.util.Map; import javax.security.auth.callback.Callback; import javax.security.auth.callback.CallbackHandler; @@ -33,7 +30,6 @@ import javax.security.sasl.Sasl; import javax.security.sasl.SaslException; import javax.security.sasl.SaslServer; -import org.apache.commons.codec.binary.Base64; import org.slf4j.Logger; import org.slf4j.LoggerFactory; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java index 2a5ae99..980c0f6 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormClientHandler.java @@ -69,8 +69,6 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler { LOG.error("Failed to authenticate with server " + "due to error: ", e); } - return; - } @Override @@ -131,7 +129,7 @@ public class SaslStormClientHandler extends SimpleChannelUpstreamHandler { if (!saslNettyClient.isComplete()) { LOG.warn("Generated a null response, " + "but authentication is not complete."); - throw new Exception("Server reponse is null, but as far as " + throw new Exception("Server response is null, but as far as " + "we can tell, we are not authenticated yet."); } this.client.channelReady(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormServerHandler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormServerHandler.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormServerHandler.java index 02448e2..5ce90a3 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormServerHandler.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslStormServerHandler.java @@ -55,7 +55,7 @@ public class SaslStormServerHandler extends SimpleChannelUpstreamHandler { Channel channel = ctx.getChannel(); if (msg instanceof ControlMessage - && ((ControlMessage) e.getMessage()) == ControlMessage.SASL_TOKEN_MESSAGE_REQUEST) { + && e.getMessage() == ControlMessage.SASL_TOKEN_MESSAGE_REQUEST) { // initialize server-side SASL functionality, if we haven't yet // (in which case we are looking at the first SASL message from the // client). @@ -85,7 +85,7 @@ public class SaslStormServerHandler extends SimpleChannelUpstreamHandler { LOG.debug("processToken: With nettyServer: " + saslNettyServer + " and token length: " + token.length); - SaslMessageToken saslTokenMessageRequest = null; + SaslMessageToken saslTokenMessageRequest; saslTokenMessageRequest = new SaslMessageToken( saslNettyServer.response(new byte[0])); // Send response to client. @@ -102,10 +102,8 @@ public class SaslStormServerHandler extends SimpleChannelUpstreamHandler { SaslNettyServer saslNettyServer = SaslNettyServerState.getSaslNettyServer .get(channel); if (saslNettyServer == null) { - if (saslNettyServer == null) { - throw new Exception("saslNettyServer was unexpectedly " - + "null for channel: " + channel); - } + throw new Exception("saslNettyServer was unexpectedly " + + "null for channel: " + channel); } SaslMessageToken saslTokenMessageRequest = new SaslMessageToken( saslNettyServer.response(((SaslMessageToken) msg) @@ -124,7 +122,6 @@ public class SaslStormServerHandler extends SimpleChannelUpstreamHandler { + "authentication is complete."); ctx.getPipeline().remove(this); } - return; } else { // Client should not be sending other-than-SASL messages before // SaslServerHandler has removed itself from the pipeline. Such http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/SaslUtils.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslUtils.java b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslUtils.java index a2d0b26..cde4d9f 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/SaslUtils.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/SaslUtils.java @@ -17,7 +17,6 @@ */ package backtype.storm.messaging.netty; -import java.nio.charset.Charset; import java.util.HashMap; import java.util.Map; @@ -33,7 +32,7 @@ class SaslUtils { public static final String DEFAULT_REALM = "default"; static Map<String, String> getSaslProps() { - Map<String, String> props = new HashMap<String, String>(); + Map<String, String> props = new HashMap<>(); props.put(Sasl.POLICY_NOPLAINTEXT, "true"); return props; } @@ -62,13 +61,7 @@ class SaslUtils { } static String getSecretKey(Map conf) { - if (conf == null || conf.isEmpty()) - return null; - - String secretPayLoad = (String) conf - .get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD); - - return secretPayLoad; + return conf == null || conf.isEmpty() ? null : (String) conf.get(Config.STORM_ZOOKEEPER_TOPOLOGY_AUTH_PAYLOAD); } } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java index 7011b42..32c2bd7 100644 --- a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java +++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java @@ -56,7 +56,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe @SuppressWarnings("rawtypes") Map storm_conf; int port; - private final ConcurrentHashMap<String, AtomicInteger> messagesEnqueued = new ConcurrentHashMap<String, AtomicInteger>(); + private final ConcurrentHashMap<String, AtomicInteger> messagesEnqueued = new ConcurrentHashMap<>(); private final AtomicInteger messagesDequeued = new AtomicInteger(0); private final AtomicInteger[] pendingMessages; @@ -84,12 +84,12 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe queueCount = Utils.getInt(storm_conf.get(Config.WORKER_RECEIVER_THREAD_COUNT), 1); roundRobinQueueId = 0; - taskToQueueId = new HashMap<Integer, Integer>(); + taskToQueueId = new HashMap<>(); message_queue = new LinkedBlockingQueue[queueCount]; pendingMessages = new AtomicInteger[queueCount]; for (int i = 0; i < queueCount; i++) { - message_queue[i] = new LinkedBlockingQueue<ArrayList<TaskMessage>>(); + message_queue[i] = new LinkedBlockingQueue<>(); pendingMessages[i] = new AtomicInteger(0); } @@ -128,8 +128,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe private ArrayList<TaskMessage>[] groupMessages(List<TaskMessage> msgs) { ArrayList<TaskMessage> messageGroups[] = new ArrayList[queueCount]; - for (int i = 0; i < msgs.size(); i++) { - TaskMessage message = msgs.get(i); + for (TaskMessage message : msgs) { int task = message.task(); if (task == -1) { @@ -140,7 +139,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe Integer queueId = getMessageQueueId(task); if (null == messageGroups[queueId]) { - messageGroups[queueId] = new ArrayList<TaskMessage>(); + messageGroups[queueId] = new ArrayList<>(); } messageGroups[queueId].add(message); } @@ -158,7 +157,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe if (roundRobinQueueId == queueCount) { roundRobinQueueId = 0; } - HashMap<Integer, Integer> newRef = new HashMap<Integer, Integer>(taskToQueueId); + HashMap<Integer, Integer> newRef = new HashMap<>(taskToQueueId); newRef.put(task, queueId); taskToQueueId = newRef; } @@ -214,7 +213,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe return closeMessage.iterator(); } - ArrayList<TaskMessage> ret = null; + ArrayList<TaskMessage> ret; int queueId = receiverId % queueCount; if ((flags & 0x01) == 0x01) { //non-blocking @@ -240,15 +239,14 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe /** * register a newly created channel - * @param channel + * @param channel newly created channel */ protected void addChannel(Channel channel) { allChannels.add(channel); } /** - * close a channel - * @param channel + * @param channel channel to close */ public void closeChannel(Channel channel) { channel.close().awaitUninterruptibly(); @@ -326,14 +324,14 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe public Object getState() { LOG.debug("Getting metrics for server on port {}", port); - HashMap<String, Object> ret = new HashMap<String, Object>(); + HashMap<String, Object> ret = new HashMap<>(); ret.put("dequeuedMessages", messagesDequeued.getAndSet(0)); - ArrayList<Integer> pending = new ArrayList<Integer>(pendingMessages.length); + ArrayList<Integer> pending = new ArrayList<>(pendingMessages.length); for (AtomicInteger p: pendingMessages) { pending.add(p.get()); } ret.put("pending", pending); - HashMap<String, Integer> enqueued = new HashMap<String, Integer>(); + HashMap<String, Integer> enqueued = new HashMap<>(); Iterator<Map.Entry<String, AtomicInteger>> it = messagesEnqueued.entrySet().iterator(); while (it.hasNext()) { Map.Entry<String, AtomicInteger> ent = it.next(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/EventLoggerBolt.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/EventLoggerBolt.java b/storm-core/src/jvm/backtype/storm/metric/EventLoggerBolt.java index 6324237..6d43e65 100644 --- a/storm-core/src/jvm/backtype/storm/metric/EventLoggerBolt.java +++ b/storm-core/src/jvm/backtype/storm/metric/EventLoggerBolt.java @@ -20,18 +20,10 @@ package backtype.storm.metric; import backtype.storm.task.IBolt; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; -import backtype.storm.tuple.Fields; import backtype.storm.tuple.Tuple; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.BufferedWriter; -import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.nio.file.StandardOpenOption; import java.util.Map; import static backtype.storm.metric.IEventLogger.EventInfo; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/FileBasedEventLogger.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/FileBasedEventLogger.java b/storm-core/src/jvm/backtype/storm/metric/FileBasedEventLogger.java index 27aa9d9..3003427 100644 --- a/storm-core/src/jvm/backtype/storm/metric/FileBasedEventLogger.java +++ b/storm-core/src/jvm/backtype/storm/metric/FileBasedEventLogger.java @@ -35,7 +35,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; public class FileBasedEventLogger implements IEventLogger { - private static Logger LOG = LoggerFactory.getLogger(FileBasedEventLogger.class); + private static final Logger LOG = LoggerFactory.getLogger(FileBasedEventLogger.class); private static final int FLUSH_INTERVAL_MILLIS = 1000; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsConsumer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsConsumer.java b/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsConsumer.java index 77fe49c..4f4242a 100644 --- a/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsConsumer.java +++ b/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsConsumer.java @@ -21,7 +21,6 @@ import java.util.Arrays; import java.util.Collection; import java.util.Map; import java.net.URL; -import java.net.URLConnection; import java.net.HttpURLConnection; import com.esotericsoftware.kryo.io.Output; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsServer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsServer.java b/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsServer.java index 6441a39..7a8f676 100644 --- a/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsServer.java +++ b/storm-core/src/jvm/backtype/storm/metric/HttpForwardingMetricsServer.java @@ -18,7 +18,6 @@ package backtype.storm.metric; import java.io.IOException; -import java.util.Arrays; import java.util.Collection; import java.util.Map; import java.util.List; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/IEventLogger.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/IEventLogger.java b/storm-core/src/jvm/backtype/storm/metric/IEventLogger.java index 4f0b34e..5fd5a32 100644 --- a/storm-core/src/jvm/backtype/storm/metric/IEventLogger.java +++ b/storm-core/src/jvm/backtype/storm/metric/IEventLogger.java @@ -17,9 +17,7 @@ */ package backtype.storm.metric; -import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; -import backtype.storm.tuple.Tuple; import java.util.Date; import java.util.Map; @@ -54,11 +52,7 @@ public interface IEventLogger { */ @Override public String toString() { - return new StringBuilder(new Date(Long.parseLong(ts)).toString()).append(",") - .append(component).append(",") - .append(task).append(",") - .append(messageId).append(",") - .append(values).toString(); + return new Date(Long.parseLong(ts)).toString() + "," + component + "," + task + "," + messageId + "," + values; } } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/LoggingMetricsConsumer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/LoggingMetricsConsumer.java b/storm-core/src/jvm/backtype/storm/metric/LoggingMetricsConsumer.java index c1c7c0a..98fb527 100644 --- a/storm-core/src/jvm/backtype/storm/metric/LoggingMetricsConsumer.java +++ b/storm-core/src/jvm/backtype/storm/metric/LoggingMetricsConsumer.java @@ -26,7 +26,6 @@ import java.util.Map; import backtype.storm.metric.api.IMetricsConsumer; import backtype.storm.task.IErrorReporter; import backtype.storm.task.TopologyContext; -import backtype.storm.utils.Utils; /* * Listens for all metrics, dumps them to log http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/MetricsConsumerBolt.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/MetricsConsumerBolt.java b/storm-core/src/jvm/backtype/storm/metric/MetricsConsumerBolt.java index 5a40c27..e3ec069 100644 --- a/storm-core/src/jvm/backtype/storm/metric/MetricsConsumerBolt.java +++ b/storm-core/src/jvm/backtype/storm/metric/MetricsConsumerBolt.java @@ -20,7 +20,6 @@ package backtype.storm.metric; import backtype.storm.Config; import backtype.storm.metric.api.IMetricsConsumer; import backtype.storm.task.IBolt; -import backtype.storm.task.IErrorReporter; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.tuple.Tuple; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/SystemBolt.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/SystemBolt.java b/storm-core/src/jvm/backtype/storm/metric/SystemBolt.java index d0bbe44..5cba7d3 100644 --- a/storm-core/src/jvm/backtype/storm/metric/SystemBolt.java +++ b/storm-core/src/jvm/backtype/storm/metric/SystemBolt.java @@ -18,7 +18,6 @@ package backtype.storm.metric; import backtype.storm.Config; -import backtype.storm.metric.api.AssignableMetric; import backtype.storm.metric.api.IMetric; import backtype.storm.task.IBolt; import backtype.storm.task.OutputCollector; @@ -28,12 +27,9 @@ import backtype.storm.utils.Utils; import clojure.lang.AFn; import clojure.lang.IFn; import clojure.lang.RT; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.lang.management.*; import java.util.HashMap; -import java.util.List; import java.util.Map; @@ -41,7 +37,6 @@ import java.util.Map; // TaskID is always -1, therefore you can only send-unanchored tuples to co-located SystemBolt. // This bolt was conceived to export worker stats via metrics api. public class SystemBolt implements IBolt { - private static Logger LOG = LoggerFactory.getLogger(SystemBolt.class); private static boolean _prepareWasCalled = false; private static class MemoryUsageMetric implements IMetric { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/api/CountMetric.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/api/CountMetric.java b/storm-core/src/jvm/backtype/storm/metric/api/CountMetric.java index dd048b8..6b317da 100644 --- a/storm-core/src/jvm/backtype/storm/metric/api/CountMetric.java +++ b/storm-core/src/jvm/backtype/storm/metric/api/CountMetric.java @@ -17,8 +17,6 @@ */ package backtype.storm.metric.api; -import backtype.storm.metric.api.IMetric; - public class CountMetric implements IMetric { long _value = 0; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/api/MeanReducer.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/api/MeanReducer.java b/storm-core/src/jvm/backtype/storm/metric/api/MeanReducer.java index e25e26d..fde7053 100644 --- a/storm-core/src/jvm/backtype/storm/metric/api/MeanReducer.java +++ b/storm-core/src/jvm/backtype/storm/metric/api/MeanReducer.java @@ -17,8 +17,6 @@ */ package backtype.storm.metric.api; -import backtype.storm.metric.api.IReducer; - class MeanReducerState { public int count = 0; public double sum = 0.0; @@ -47,7 +45,7 @@ public class MeanReducer implements IReducer<MeanReducerState> { public Object extractResult(MeanReducerState acc) { if(acc.count > 0) { - return new Double(acc.sum / (double)acc.count); + return acc.sum / (double) acc.count; } else { return null; } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/api/MultiCountMetric.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/api/MultiCountMetric.java b/storm-core/src/jvm/backtype/storm/metric/api/MultiCountMetric.java index c420a16..a6077e6 100644 --- a/storm-core/src/jvm/backtype/storm/metric/api/MultiCountMetric.java +++ b/storm-core/src/jvm/backtype/storm/metric/api/MultiCountMetric.java @@ -22,7 +22,7 @@ import java.util.HashMap; import java.util.Map; public class MultiCountMetric implements IMetric { - Map<String, CountMetric> _value = new HashMap(); + Map<String, CountMetric> _value = new HashMap<>(); public MultiCountMetric() { } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/api/MultiReducedMetric.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/api/MultiReducedMetric.java b/storm-core/src/jvm/backtype/storm/metric/api/MultiReducedMetric.java index 530b168..d9d3a02 100644 --- a/storm-core/src/jvm/backtype/storm/metric/api/MultiReducedMetric.java +++ b/storm-core/src/jvm/backtype/storm/metric/api/MultiReducedMetric.java @@ -22,7 +22,7 @@ import java.util.HashMap; import java.util.Map; public class MultiReducedMetric implements IMetric { - Map<String, ReducedMetric> _value = new HashMap(); + Map<String, ReducedMetric> _value = new HashMap<>(); IReducer _reducer; public MultiReducedMetric(IReducer reducer) { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/metric/api/rpc/CountShellMetric.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/metric/api/rpc/CountShellMetric.java b/storm-core/src/jvm/backtype/storm/metric/api/rpc/CountShellMetric.java index def74c2..eae982b 100644 --- a/storm-core/src/jvm/backtype/storm/metric/api/rpc/CountShellMetric.java +++ b/storm-core/src/jvm/backtype/storm/metric/api/rpc/CountShellMetric.java @@ -21,8 +21,7 @@ import backtype.storm.metric.api.CountMetric; public class CountShellMetric extends CountMetric implements IShellMetric { /*** - * @param - * params should be null or long + * @param value should be null or long * if value is null, it will call incr() * if value is long, it will call incrBy((long)params) * */ http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java b/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java index e91e54a..b17ca40 100644 --- a/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java +++ b/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java @@ -75,7 +75,7 @@ public abstract class AbstractDNSToSwitchMapping builder.append("Mapping: ").append(toString()).append("\n"); if (rack != null) { builder.append("Map:\n"); - Set<String> switches = new HashSet<String>(); + Set<String> switches = new HashSet<>(); for (Map.Entry<String, String> entry : rack.entrySet()) { builder.append(" ") .append(entry.getKey()) http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java b/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java index 8e1e521..18eac60 100644 --- a/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java +++ b/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java @@ -28,12 +28,12 @@ import java.util.concurrent.ConcurrentHashMap; */ public final class DefaultRackDNSToSwitchMapping extends AbstractDNSToSwitchMapping { - private Map<String, String> mappingCache = new ConcurrentHashMap<String, String>(); + private Map<String, String> mappingCache = new ConcurrentHashMap<>(); @Override public Map<String,String> resolve(List<String> names) { - Map<String, String> m = new HashMap<String, String>(); + Map<String, String> m = new HashMap<>(); if (names.isEmpty()) { //name list is empty, return an empty map return m; http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/nimbus/ILeaderElector.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/nimbus/ILeaderElector.java b/storm-core/src/jvm/backtype/storm/nimbus/ILeaderElector.java index 1148be2..5b54d46 100644 --- a/storm-core/src/jvm/backtype/storm/nimbus/ILeaderElector.java +++ b/storm-core/src/jvm/backtype/storm/nimbus/ILeaderElector.java @@ -17,11 +17,7 @@ */ package backtype.storm.nimbus; -import org.apache.curator.framework.CuratorFramework; - import java.io.Closeable; -import java.net.InetSocketAddress; -import java.net.UnknownHostException; import java.util.List; import java.util.Map; @@ -32,7 +28,7 @@ public interface ILeaderElector extends Closeable { /** * Method guaranteed to be called as part of initialization of leader elector instance. - * @param conf + * @param conf configuration */ void prepare(Map conf); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/nimbus/NimbusInfo.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/nimbus/NimbusInfo.java b/storm-core/src/jvm/backtype/storm/nimbus/NimbusInfo.java index 77c0d7a..539df62 100644 --- a/storm-core/src/jvm/backtype/storm/nimbus/NimbusInfo.java +++ b/storm-core/src/jvm/backtype/storm/nimbus/NimbusInfo.java @@ -94,9 +94,7 @@ public class NimbusInfo implements Serializable { if (isLeader != that.isLeader) return false; if (port != that.port) return false; - if (!host.equals(that.host)) return false; - - return true; + return host.equals(that.host); } @Override http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java index b07576f..b3028e9 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java @@ -265,14 +265,13 @@ public class Cluster { * @return the number of workers assigned to this topology. */ public int getAssignedNumWorkers(TopologyDetails topology) { - SchedulerAssignment assignment = this.getAssignmentById(topology.getId()); - if (topology == null || assignment == null) { + SchedulerAssignment assignment = topology != null ? this.getAssignmentById(topology.getId()) : null; + if (assignment == null) { return 0; } Set<WorkerSlot> slots = new HashSet<WorkerSlot>(); slots.addAll(assignment.getExecutorToSlot().values()); - return slots.size(); } http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java b/storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java index 08af4b7..e06abf8 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/SchedulerAssignmentImpl.java @@ -38,7 +38,7 @@ public class SchedulerAssignmentImpl implements SchedulerAssignment { public SchedulerAssignmentImpl(String topologyId, Map<ExecutorDetails, WorkerSlot> executorToSlots) { this.topologyId = topologyId; - this.executorToSlot = new HashMap<ExecutorDetails, WorkerSlot>(0); + this.executorToSlot = new HashMap<>(0); if (executorToSlots != null) { this.executorToSlot.putAll(executorToSlots); } @@ -46,13 +46,11 @@ public class SchedulerAssignmentImpl implements SchedulerAssignment { @Override public Set<WorkerSlot> getSlots() { - return new HashSet(executorToSlot.values()); + return new HashSet<>(executorToSlot.values()); } /** * Assign the slot to executors. - * @param slot - * @param executors */ public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors) { for (ExecutorDetails executor : executors) { @@ -62,10 +60,9 @@ public class SchedulerAssignmentImpl implements SchedulerAssignment { /** * Release the slot occupied by this assignment. - * @param slot */ public void unassignBySlot(WorkerSlot slot) { - List<ExecutorDetails> executors = new ArrayList<ExecutorDetails>(); + List<ExecutorDetails> executors = new ArrayList<>(); for (ExecutorDetails executor : this.executorToSlot.keySet()) { WorkerSlot ws = this.executorToSlot.get(executor); if (ws.equals(slot)) { @@ -80,9 +77,8 @@ public class SchedulerAssignmentImpl implements SchedulerAssignment { } /** - * Does this slot occupied by this assignment? * @param slot - * @return + * @return true if slot is occupied by this assignment */ public boolean isSlotOccupied(WorkerSlot slot) { return this.executorToSlot.containsValue(slot); @@ -101,8 +97,7 @@ public class SchedulerAssignmentImpl implements SchedulerAssignment { } /** - * Return the executors covered by this assignments - * @return + * @return the executors covered by this assignments */ public Set<ExecutorDetails> getExecutors() { return this.executorToSlot.keySet(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java b/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java index 65e6e9b..a748e11 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java @@ -44,7 +44,7 @@ public class SupervisorDetails { */ Set<Integer> allPorts; /** - * Map containing a manifest of resources for the node the superivsor resides + * Map containing a manifest of resources for the node the supervisor resides */ private Map<String, Double> _total_resources; @@ -58,7 +58,7 @@ public class SupervisorDetails { if(allPorts!=null) { setAllPorts(allPorts); } else { - this.allPorts = new HashSet(); + this.allPorts = new HashSet<>(); } this._total_resources = total_resources; LOG.debug("Creating a new supervisor ({}-{}) with resources: {}", this.host, this.id, total_resources); @@ -86,7 +86,7 @@ public class SupervisorDetails { } private void setAllPorts(Collection<Number> allPorts) { - this.allPorts = new HashSet<Integer>(); + this.allPorts = new HashSet<>(); if(allPorts!=null) { for(Number n: allPorts) { this.allPorts.add(n.intValue()); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java index 443bf3f..0828a73 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java @@ -29,10 +29,10 @@ public class Topologies { Map<String, Map<String, Component>> _allComponents; public Topologies(Map<String, TopologyDetails> topologies) { - if(topologies==null) topologies = new HashMap(); - this.topologies = new HashMap<String, TopologyDetails>(topologies.size()); + if(topologies==null) topologies = new HashMap<>(); + this.topologies = new HashMap<>(topologies.size()); this.topologies.putAll(topologies); - this.nameToId = new HashMap<String, String>(topologies.size()); + this.nameToId = new HashMap<>(topologies.size()); for (Map.Entry<String, TopologyDetails> entry : topologies.entrySet()) { TopologyDetails topology = entry.getValue(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java b/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java index 01fa0df..95aa5c8 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java @@ -59,7 +59,7 @@ public class TopologyDetails { public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology, int numWorkers, Map<ExecutorDetails, String> executorToComponents) { this(topologyId, topologyConf, topology, numWorkers); - this.executorToComponent = new HashMap<ExecutorDetails, String>(0); + this.executorToComponent = new HashMap<>(0); if (executorToComponents != null) { this.executorToComponent.putAll(executorToComponents); } @@ -88,7 +88,7 @@ public class TopologyDetails { } public Map<ExecutorDetails, String> selectExecutorToComponent(Collection<ExecutorDetails> executors) { - Map<ExecutorDetails, String> ret = new HashMap<ExecutorDetails, String>(executors.size()); + Map<ExecutorDetails, String> ret = new HashMap<>(executors.size()); for (ExecutorDetails executor : executors) { String compId = this.executorToComponent.get(executor); if (compId != null) { @@ -104,7 +104,7 @@ public class TopologyDetails { } private void initResourceList() { - _resourceList = new HashMap<ExecutorDetails, Map<String, Double>>(); + _resourceList = new HashMap<>(); // Extract bolt memory info if (this.topology.get_bolts() != null) { for (Map.Entry<String, Bolt> bolt : this.topology.get_bolts().entrySet()) { @@ -136,7 +136,7 @@ public class TopologyDetails { } //schedule tasks that are not part of components returned from topology.get_spout or topology.getbolt (AKA sys tasks most specifically __acker tasks) for(ExecutorDetails exec : this.getExecutors()) { - if (_resourceList.containsKey(exec) == false) { + if (!_resourceList.containsKey(exec)) { LOG.debug( "Scheduling {} {} with memory requirement as 'on heap' - {} and 'off heap' - {} and CPU requirement as {}", this.getExecutorToComponent().get(exec), @@ -166,7 +166,7 @@ public class TopologyDetails { * @return a map of components */ public Map<String, Component> getComponents() { - Map<String, Component> all_comp = new HashMap<String, Component>(); + Map<String, Component> all_comp = new HashMap<>(); StormTopology storm_topo = this.topology; // spouts @@ -174,7 +174,7 @@ public class TopologyDetails { for (Map.Entry<String, SpoutSpec> spoutEntry : storm_topo .get_spouts().entrySet()) { if (!Utils.isSystemId(spoutEntry.getKey())) { - Component newComp = null; + Component newComp; if (all_comp.containsKey(spoutEntry.getKey())) { newComp = all_comp.get(spoutEntry.getKey()); newComp.execs = componentToExecs(newComp.id); @@ -209,7 +209,7 @@ public class TopologyDetails { for (Map.Entry<String, Bolt> boltEntry : storm_topo.get_bolts() .entrySet()) { if (!Utils.isSystemId(boltEntry.getKey())) { - Component newComp = null; + Component newComp; if (all_comp.containsKey(boltEntry.getKey())) { newComp = all_comp.get(boltEntry.getKey()); newComp.execs = componentToExecs(newComp.id); @@ -297,7 +297,7 @@ public class TopologyDetails { * for all tasks in topology topoId. */ public Map<ExecutorDetails, Double> getTotalMemoryResourceList() { - Map<ExecutorDetails, Double> ret = new HashMap<ExecutorDetails, Double>(); + Map<ExecutorDetails, Double> ret = new HashMap<>(); for (ExecutorDetails exec : _resourceList.keySet()) { ret.put(exec, getTotalMemReqTask(exec)); } @@ -306,7 +306,6 @@ public class TopologyDetails { /** * Get the total CPU requirement for executor - * @param exec * @return Double the total about of cpu requirement for executor */ public Double getTotalCpuReqTask(ExecutorDetails exec) { @@ -386,17 +385,11 @@ public class TopologyDetails { * @return Boolean whether or not a certain ExecutorDetail is included in the _resourceList. */ public boolean hasExecInTopo(ExecutorDetails exec) { - if (_resourceList != null) { // null is possible if the first constructor of TopologyDetails is used - return _resourceList.containsKey(exec); - } else { - return false; - } + return _resourceList != null && _resourceList.containsKey(exec); } /** * add resource requirements for a executor - * @param exec - * @param resourceList */ public void addResourcesForExec(ExecutorDetails exec, Map<String, Double> resourceList) { if (hasExecInTopo(exec)) { @@ -408,10 +401,9 @@ public class TopologyDetails { /** * Add default resource requirements for a executor - * @param exec */ public void addDefaultResforExec(ExecutorDetails exec) { - Map<String, Double> defaultResourceList = new HashMap<String, Double>(); + Map<String, Double> defaultResourceList = new HashMap<>(); defaultResourceList.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, Utils.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null)); defaultResourceList.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, @@ -427,7 +419,7 @@ public class TopologyDetails { } /** - * initalizes the scheduler member variable by extracting what scheduler + * initializes the scheduler member variable by extracting what scheduler * this topology is going to use from topologyConf */ private void initConfigs() { http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/DefaultPool.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/DefaultPool.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/DefaultPool.java index 3053b5b..2e418e9 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/DefaultPool.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/DefaultPool.java @@ -37,8 +37,8 @@ import backtype.storm.scheduler.WorkerSlot; */ public class DefaultPool extends NodePool { private static final Logger LOG = LoggerFactory.getLogger(DefaultPool.class); - private Set<Node> _nodes = new HashSet<Node>(); - private HashMap<String, TopologyDetails> _tds = new HashMap<String, TopologyDetails>(); + private Set<Node> _nodes = new HashSet<>(); + private HashMap<String, TopologyDetails> _tds = new HashMap<>(); @Override public void addTopology(TopologyDetails td) { @@ -61,8 +61,8 @@ public class DefaultPool extends NodePool { @Override public Collection<Node> takeNodes(int nodesNeeded) { - HashSet<Node> ret = new HashSet<Node>(); - LinkedList<Node> sortedNodes = new LinkedList<Node>(_nodes); + HashSet<Node> ret = new HashSet<>(); + LinkedList<Node> sortedNodes = new LinkedList<>(_nodes); Collections.sort(sortedNodes, Node.FREE_NODE_COMPARATOR_DEC); for (Node n: sortedNodes) { if (nodesNeeded <= ret.size()) { @@ -95,7 +95,7 @@ public class DefaultPool extends NodePool { public NodeAndSlotCounts getNodeAndSlotCountIfSlotsWereTaken(int slotsNeeded) { int nodesFound = 0; int slotsFound = 0; - LinkedList<Node> sortedNodes = new LinkedList<Node>(_nodes); + LinkedList<Node> sortedNodes = new LinkedList<>(_nodes); Collections.sort(sortedNodes, Node.FREE_NODE_COMPARATOR_DEC); for (Node n: sortedNodes) { if (slotsNeeded <= 0) { @@ -113,8 +113,8 @@ public class DefaultPool extends NodePool { @Override public Collection<Node> takeNodesBySlots(int slotsNeeded) { - HashSet<Node> ret = new HashSet<Node>(); - LinkedList<Node> sortedNodes = new LinkedList<Node>(_nodes); + HashSet<Node> ret = new HashSet<>(); + LinkedList<Node> sortedNodes = new LinkedList<>(_nodes); Collections.sort(sortedNodes, Node.FREE_NODE_COMPARATOR_DEC); for (Node n: sortedNodes) { if (slotsNeeded <= 0) { @@ -148,8 +148,8 @@ public class DefaultPool extends NodePool { } int slotsToUse = Math.min(slotsRequested - slotsUsed, slotsFree + slotsAvailable); int executorsNotRunning = _cluster.getUnassignedExecutors(td).size(); - LOG.debug("Slots... requested {} used {} free {} available {} to be used {}, executors not running {}", - new Object[] {slotsRequested, slotsUsed, slotsFree, slotsAvailable, slotsToUse, executorsNotRunning}); + LOG.debug("Slots... requested {} used {} free {} available {} to be used {}, executors not running {}", + slotsRequested, slotsUsed, slotsFree, slotsAvailable, slotsToUse, executorsNotRunning); if (slotsToUse <= 0) { if (executorsNotRunning > 0) { _cluster.setStatus(topId,"Not fully scheduled (No free slots in default pool) "+executorsNotRunning+" executors not scheduled"); @@ -180,9 +180,9 @@ public class DefaultPool extends NodePool { RoundRobinSlotScheduler slotSched = new RoundRobinSlotScheduler(td, slotsToUse, _cluster); - LinkedList<Node> nodes = new LinkedList<Node>(_nodes); + LinkedList<Node> nodes = new LinkedList<>(_nodes); while (true) { - Node n = null; + Node n; do { if (nodes.isEmpty()) { throw new IllegalStateException("This should not happen, we" + http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/FreePool.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/FreePool.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/FreePool.java index c625895..456900b 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/FreePool.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/FreePool.java @@ -35,7 +35,7 @@ import backtype.storm.scheduler.TopologyDetails; */ public class FreePool extends NodePool { private static final Logger LOG = LoggerFactory.getLogger(FreePool.class); - private Set<Node> _nodes = new HashSet<Node>(); + private Set<Node> _nodes = new HashSet<>(); private int _totalSlots = 0; @Override @@ -63,7 +63,7 @@ public class FreePool extends NodePool { @Override public Collection<Node> takeNodes(int nodesNeeded) { - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); Iterator<Node> it = _nodes.iterator(); while (it.hasNext() && nodesNeeded > ret.size()) { Node n = it.next(); @@ -86,7 +86,7 @@ public class FreePool extends NodePool { @Override public Collection<Node> takeNodesBySlots(int slotsNeeded) { - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); Iterator<Node> it = _nodes.iterator(); while (it.hasNext() && slotsNeeded > 0) { Node n = it.next(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/IsolatedPool.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/IsolatedPool.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/IsolatedPool.java index dc7eded..25a6f25 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/IsolatedPool.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/IsolatedPool.java @@ -41,9 +41,9 @@ import backtype.storm.scheduler.WorkerSlot; */ public class IsolatedPool extends NodePool { private static final Logger LOG = LoggerFactory.getLogger(IsolatedPool.class); - private Map<String, Set<Node>> _topologyIdToNodes = new HashMap<String, Set<Node>>(); - private HashMap<String, TopologyDetails> _tds = new HashMap<String, TopologyDetails>(); - private HashSet<String> _isolated = new HashSet<String>(); + private Map<String, Set<Node>> _topologyIdToNodes = new HashMap<>(); + private HashMap<String, TopologyDetails> _tds = new HashMap<>(); + private HashSet<String> _isolated = new HashSet<>(); private int _maxNodes; private int _usedNodes; @@ -57,7 +57,7 @@ public class IsolatedPool extends NodePool { String topId = td.getId(); LOG.debug("Adding in Topology {}", topId); SchedulerAssignment assignment = _cluster.getAssignmentById(topId); - Set<Node> assignedNodes = new HashSet<Node>(); + Set<Node> assignedNodes = new HashSet<>(); if (assignment != null) { for (WorkerSlot ws: assignment.getSlots()) { Node n = _nodeIdToNode.get(ws.getNodeId()); @@ -96,7 +96,7 @@ public class IsolatedPool extends NodePool { LOG.debug("Scheduling topology {}",topId); Set<Node> allNodes = _topologyIdToNodes.get(topId); Number nodesRequested = (Number) td.getConf().get(Config.TOPOLOGY_ISOLATED_MACHINES); - int slotsToUse = 0; + int slotsToUse; if (nodesRequested == null) { slotsToUse = getNodesForNotIsolatedTop(td, allNodes, lesserPools); } else { @@ -111,7 +111,7 @@ public class IsolatedPool extends NodePool { RoundRobinSlotScheduler slotSched = new RoundRobinSlotScheduler(td, slotsToUse, _cluster); - LinkedList<Node> sortedNodes = new LinkedList<Node>(allNodes); + LinkedList<Node> sortedNodes = new LinkedList<>(allNodes); Collections.sort(sortedNodes, Node.FREE_NODE_COMPARATOR_DEC); LOG.debug("Nodes sorted by free space {}", sortedNodes); @@ -157,9 +157,9 @@ public class IsolatedPool extends NodePool { int nodesUsed = _topologyIdToNodes.get(topId).size(); int nodesNeeded = nodesRequested - nodesUsed; LOG.debug("Nodes... requested {} used {} available from us {} " + - "avail from other {} needed {}", new Object[] {nodesRequested, - nodesUsed, nodesFromUsAvailable, nodesFromOthersAvailable, - nodesNeeded}); + "avail from other {} needed {}", nodesRequested, + nodesUsed, nodesFromUsAvailable, nodesFromOthersAvailable, + nodesNeeded); if ((nodesNeeded - nodesFromUsAvailable) > (_maxNodes - _usedNodes)) { _cluster.setStatus(topId,"Max Nodes("+_maxNodes+") for this user would be exceeded. " + ((nodesNeeded - nodesFromUsAvailable) - (_maxNodes - _usedNodes)) @@ -224,8 +224,8 @@ public class IsolatedPool extends NodePool { slotsAvailable = NodePool.slotsAvailable(lesserPools); } int slotsToUse = Math.min(slotsRequested - slotsUsed, slotsFree + slotsAvailable); - LOG.debug("Slots... requested {} used {} free {} available {} to be used {}", - new Object[] {slotsRequested, slotsUsed, slotsFree, slotsAvailable, slotsToUse}); + LOG.debug("Slots... requested {} used {} free {} available {} to be used {}", + slotsRequested, slotsUsed, slotsFree, slotsAvailable, slotsToUse); if (slotsToUse <= 0) { _cluster.setStatus(topId, "Not Enough Slots Available to Schedule Topology"); return 0; @@ -233,7 +233,7 @@ public class IsolatedPool extends NodePool { int slotsNeeded = slotsToUse - slotsFree; int numNewNodes = NodePool.getNodeCountIfSlotsWereTaken(slotsNeeded, lesserPools); LOG.debug("Nodes... new {} used {} max {}", - new Object[]{numNewNodes, _usedNodes, _maxNodes}); + numNewNodes, _usedNodes, _maxNodes); if ((numNewNodes + _usedNodes) > _maxNodes) { _cluster.setStatus(topId,"Max Nodes("+_maxNodes+") for this user would be exceeded. " + (numNewNodes - (_maxNodes - _usedNodes)) + " more nodes needed to run topology."); @@ -249,7 +249,7 @@ public class IsolatedPool extends NodePool { @Override public Collection<Node> takeNodes(int nodesNeeded) { LOG.debug("Taking {} from {}", nodesNeeded, this); - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); for (Entry<String, Set<Node>> entry: _topologyIdToNodes.entrySet()) { if (!_isolated.contains(entry.getKey())) { Iterator<Node> it = entry.getValue().iterator(); @@ -293,7 +293,7 @@ public class IsolatedPool extends NodePool { @Override public Collection<Node> takeNodesBySlots(int slotsNeeded) { - HashSet<Node> ret = new HashSet<Node>(); + HashSet<Node> ret = new HashSet<>(); for (Entry<String, Set<Node>> entry: _topologyIdToNodes.entrySet()) { if (!_isolated.contains(entry.getKey())) { Iterator<Node> it = entry.getValue().iterator(); @@ -321,9 +321,7 @@ public class IsolatedPool extends NodePool { int slotsFound = 0; for (Entry<String, Set<Node>> entry: _topologyIdToNodes.entrySet()) { if (!_isolated.contains(entry.getKey())) { - Iterator<Node> it = entry.getValue().iterator(); - while (it.hasNext()) { - Node n = it.next(); + for (Node n : entry.getValue()) { if (n.isAlive()) { nodesFound++; int totalSlotsFree = n.totalSlots(); http://git-wip-us.apache.org/repos/asf/storm/blob/f0e4c6f2/storm-core/src/jvm/backtype/storm/scheduler/multitenant/MultitenantScheduler.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/MultitenantScheduler.java b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/MultitenantScheduler.java index 320b388..6b77c63 100644 --- a/storm-core/src/jvm/backtype/storm/scheduler/multitenant/MultitenantScheduler.java +++ b/storm-core/src/jvm/backtype/storm/scheduler/multitenant/MultitenantScheduler.java @@ -44,9 +44,9 @@ public class MultitenantScheduler implements IScheduler { private Map<String, Number> getUserConf() { Map<String, Number> ret = (Map<String, Number>)_conf.get(Config.MULTITENANT_SCHEDULER_USER_POOLS); if (ret == null) { - ret = new HashMap<String, Number>(); + ret = new HashMap<>(); } else { - ret = new HashMap<String, Number>(ret); + ret = new HashMap<>(ret); } Map fromFile = Utils.findAndReadConfigFile("multitenant-scheduler.yaml", false); @@ -65,7 +65,7 @@ public class MultitenantScheduler implements IScheduler { Map<String, Number> userConf = getUserConf(); - Map<String, IsolatedPool> userPools = new HashMap<String, IsolatedPool>(); + Map<String, IsolatedPool> userPools = new HashMap<>(); for (Map.Entry<String, Number> entry : userConf.entrySet()) { userPools.put(entry.getKey(), new IsolatedPool(entry.getValue().intValue())); }
