You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/05/18 14:08:34 UTC
[07/13] storm git commit: STORM-2477: Result of transforming the code
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/DefaultHttpCredentialsPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/DefaultHttpCredentialsPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/DefaultHttpCredentialsPlugin.java
index c0c9951..fb873d5 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/DefaultHttpCredentialsPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/DefaultHttpCredentialsPlugin.java
@@ -34,10 +34,10 @@ public class DefaultHttpCredentialsPlugin implements IHttpCredentialsPlugin {
/**
* No-op
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
@Override
- public void prepare(Map storm_conf) {
+ public void prepare(Map<String, Object> topoConf) {
// Do nothing.
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/DefaultPrincipalToLocal.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/DefaultPrincipalToLocal.java b/storm-client/src/jvm/org/apache/storm/security/auth/DefaultPrincipalToLocal.java
index e780771..af41441 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/DefaultPrincipalToLocal.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/DefaultPrincipalToLocal.java
@@ -29,7 +29,7 @@ public class DefaultPrincipalToLocal implements IPrincipalToLocal {
/**
* Invoked once immediately after construction
*/
- public void prepare(Map storm_conf) {}
+ public void prepare(Map<String, Object> topoConf) {}
/**
* Convert a Principal to a local user name.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/IAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/IAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/IAuthorizer.java
index bc60af9..70f9e32 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/IAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/IAuthorizer.java
@@ -38,16 +38,16 @@ import java.util.Map;
public interface IAuthorizer {
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
- void prepare(Map storm_conf);
+ void prepare(Map<String, Object> topoConf);
/**
* permit() method is invoked for each incoming Thrift request.
* @param context request context includes info about
* @param operation operation name
- * @param topology_conf configuration of targeted topology
+ * @param topoConf configuration of targeted topology
* @return true if the request is authorized, false if reject
*/
- public boolean permit(ReqContext context, String operation, Map topology_conf);
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/IAutoCredentials.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/IAutoCredentials.java b/storm-client/src/jvm/org/apache/storm/security/auth/IAutoCredentials.java
index 00d59df..ebaad69 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/IAutoCredentials.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/IAutoCredentials.java
@@ -28,7 +28,7 @@ import javax.security.auth.Subject;
*/
public interface IAutoCredentials {
- public void prepare(Map conf);
+ public void prepare(Map<String, Object> conf);
/**
* Called to populate the credentials on the client side.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ICredentialsRenewer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ICredentialsRenewer.java b/storm-client/src/jvm/org/apache/storm/security/auth/ICredentialsRenewer.java
index 1f947aa..0f529e6 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ICredentialsRenewer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ICredentialsRenewer.java
@@ -29,12 +29,12 @@ public interface ICredentialsRenewer {
* Called when initializing the service.
* @param conf the storm cluster configuration.
*/
- public void prepare(Map conf);
+ public void prepare(Map<String, Object> conf);
/**
* Renew any credentials that need to be renewed. (Update the credentials if needed)
* @param credentials the credentials that may have something to renew.
* @param topologyConf topology configuration.
*/
- public void renew(Map<String, String> credentials, Map topologyConf);
+ public void renew(Map<String, String> credentials, Map<String, Object> topologyConf);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/IGroupMappingServiceProvider.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/IGroupMappingServiceProvider.java b/storm-client/src/jvm/org/apache/storm/security/auth/IGroupMappingServiceProvider.java
index 2075939..d1a17d4 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/IGroupMappingServiceProvider.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/IGroupMappingServiceProvider.java
@@ -26,9 +26,9 @@ public interface IGroupMappingServiceProvider {
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
- void prepare(Map storm_conf);
+ void prepare(Map<String, Object> topoConf);
/**
* Get all various group memberships of a given user.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/IHttpCredentialsPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/IHttpCredentialsPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/IHttpCredentialsPlugin.java
index 62c8271..54c16ac 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/IHttpCredentialsPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/IHttpCredentialsPlugin.java
@@ -27,9 +27,9 @@ import javax.servlet.http.HttpServletRequest;
public interface IHttpCredentialsPlugin {
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
- void prepare(Map storm_conf);
+ void prepare(Map<String, Object> topoConf);
/**
* Gets the user name from the request.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/IPrincipalToLocal.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/IPrincipalToLocal.java b/storm-client/src/jvm/org/apache/storm/security/auth/IPrincipalToLocal.java
index 996166d..35c96d8 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/IPrincipalToLocal.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/IPrincipalToLocal.java
@@ -28,9 +28,9 @@ import java.security.Principal;
public interface IPrincipalToLocal {
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
- void prepare(Map storm_conf);
+ void prepare(Map<String, Object> topoConf);
/**
* Convert a Principal to a local user name.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
index c60b2f2..f51b1e2 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
@@ -34,10 +34,10 @@ public interface ITransportPlugin {
/**
* Invoked once immediately after construction
* @param type the type of connection this will process.
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
* @param login_conf login configuration
*/
- void prepare(ThriftConnectionType type, Map storm_conf, Configuration login_conf);
+ void prepare(ThriftConnectionType type, Map<String, Object> topoConf, Configuration login_conf);
/**
* Create a server associated with a given port, service handler, and purpose
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/KerberosPrincipalToLocal.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/KerberosPrincipalToLocal.java b/storm-client/src/jvm/org/apache/storm/security/auth/KerberosPrincipalToLocal.java
index 9f5985e..52c4263 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/KerberosPrincipalToLocal.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/KerberosPrincipalToLocal.java
@@ -28,9 +28,9 @@ public class KerberosPrincipalToLocal implements IPrincipalToLocal {
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
- public void prepare(Map storm_conf) {}
+ public void prepare(Map<String, Object> topoConf) {}
/**
* Convert a Principal to a local user name.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/SaslTransportPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/SaslTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/SaslTransportPlugin.java
index d93573b..5ad3a5a 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/SaslTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/SaslTransportPlugin.java
@@ -51,21 +51,21 @@ import org.apache.thrift.transport.TTransportFactory;
*/
public abstract class SaslTransportPlugin implements ITransportPlugin {
protected ThriftConnectionType type;
- protected Map storm_conf;
+ protected Map<String, Object> topoConf;
protected Configuration login_conf;
private int port;
@Override
- public void prepare(ThriftConnectionType type, Map storm_conf, Configuration login_conf) {
+ public void prepare(ThriftConnectionType type, Map<String, Object> topoConf, Configuration login_conf) {
this.type = type;
- this.storm_conf = storm_conf;
+ this.topoConf = topoConf;
this.login_conf = login_conf;
}
@Override
public TServer getServer(TProcessor processor) throws IOException, TTransportException {
- int configuredPort = type.getPort(storm_conf);
- Integer socketTimeout = type.getSocketTimeOut(storm_conf);
+ int configuredPort = type.getPort(topoConf);
+ Integer socketTimeout = type.getSocketTimeOut(topoConf);
TTransportFactory serverTransportFactory = getServerTransportFactory();
TServerSocket serverTransport = null;
if (socketTimeout != null) {
@@ -74,8 +74,8 @@ public abstract class SaslTransportPlugin implements ITransportPlugin {
serverTransport = new TServerSocket(configuredPort);
}
this.port = serverTransport.getServerSocket().getLocalPort();
- int numWorkerThreads = type.getNumThreads(storm_conf);
- Integer queueSize = type.getQueueSize(storm_conf);
+ int numWorkerThreads = type.getNumThreads(topoConf);
+ Integer queueSize = type.getQueueSize(topoConf);
TThreadPoolServer.Args server_args = new TThreadPoolServer.Args(serverTransport).
processor(new TUGIWrapProcessor(processor)).
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ShellBasedGroupsMapping.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ShellBasedGroupsMapping.java b/storm-client/src/jvm/org/apache/storm/security/auth/ShellBasedGroupsMapping.java
index c9f14cd..3ee5754 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ShellBasedGroupsMapping.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ShellBasedGroupsMapping.java
@@ -40,11 +40,11 @@ public class ShellBasedGroupsMapping implements
/**
* Invoked once immediately after construction
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
*/
@Override
- public void prepare(Map storm_conf) {
- int timeout = ObjectReader.getInt(storm_conf.get(Config.STORM_GROUP_MAPPING_SERVICE_CACHE_DURATION_SECS));
+ public void prepare(Map<String, Object> topoConf) {
+ int timeout = ObjectReader.getInt(topoConf.get(Config.STORM_GROUP_MAPPING_SERVICE_CACHE_DURATION_SECS));
cachedGroups = new TimeCacheMap<>(timeout);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
index b41af75..698f797 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
@@ -52,26 +52,26 @@ import org.slf4j.LoggerFactory;
*/
public class SimpleTransportPlugin implements ITransportPlugin {
protected ThriftConnectionType type;
- protected Map storm_conf;
+ protected Map<String, Object> topoConf;
protected Configuration login_conf;
private static final Logger LOG = LoggerFactory.getLogger(SimpleTransportPlugin.class);
private int port;
@Override
- public void prepare(ThriftConnectionType type, Map storm_conf, Configuration login_conf) {
+ public void prepare(ThriftConnectionType type, Map<String, Object> topoConf, Configuration login_conf) {
this.type = type;
- this.storm_conf = storm_conf;
+ this.topoConf = topoConf;
this.login_conf = login_conf;
}
@Override
public TServer getServer(TProcessor processor) throws IOException, TTransportException {
- int configuredPort = type.getPort(storm_conf);
+ int configuredPort = type.getPort(topoConf);
TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(configuredPort);
this.port = serverTransport.getPort();
- int numWorkerThreads = type.getNumThreads(storm_conf);
- int maxBufferSize = type.getMaxBufferSize(storm_conf);
- Integer queueSize = type.getQueueSize(storm_conf);
+ int numWorkerThreads = type.getNumThreads(topoConf);
+ int maxBufferSize = type.getMaxBufferSize(topoConf);
+ Integer queueSize = type.getQueueSize(topoConf);
THsHaServer.Args server_args = new THsHaServer.Args(serverTransport).
processor(new SimpleWrapProcessor(processor)).
@@ -97,7 +97,7 @@ public class SimpleTransportPlugin implements ITransportPlugin {
*/
@Override
public TTransport connect(TTransport transport, String serverHost, String asUser) throws TTransportException {
- int maxBufferSize = type.getMaxBufferSize(storm_conf);
+ int maxBufferSize = type.getMaxBufferSize(topoConf);
//create a framed transport
TTransport conn = new TFramedTransport(transport, maxBufferSize);
@@ -151,7 +151,7 @@ public class SimpleTransportPlugin implements ITransportPlugin {
//anonymous user
Subject s = getDefaultSubject();
if (s == null) {
- final String user = (String)storm_conf.get("debug.simple.transport.user");
+ final String user = (String)topoConf.get("debug.simple.transport.user");
if (user != null) {
HashSet<Principal> principals = new HashSet<>();
principals.add(new Principal() {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
index d84c5fa..febf5e2 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
@@ -38,22 +38,22 @@ public class ThriftClient implements AutoCloseable {
private String _asUser;
protected boolean _retryForever = false;
- public ThriftClient(Map storm_conf, ThriftConnectionType type, String host) {
- this(storm_conf, type, host, null, null, null);
+ public ThriftClient(Map<String, Object> topoConf, ThriftConnectionType type, String host) {
+ this(topoConf, type, host, null, null, null);
}
- public ThriftClient(Map storm_conf, ThriftConnectionType type, String host, Integer port, Integer timeout){
- this(storm_conf, type, host, port, timeout, null);
+ public ThriftClient(Map<String, Object> topoConf, ThriftConnectionType type, String host, Integer port, Integer timeout){
+ this(topoConf, type, host, port, timeout, null);
}
- public ThriftClient(Map storm_conf, ThriftConnectionType type, String host, Integer port, Integer timeout, String asUser) {
+ public ThriftClient(Map<String, Object> topoConf, ThriftConnectionType type, String host, Integer port, Integer timeout, String asUser) {
//create a socket with server
if (host==null) {
throw new IllegalArgumentException("host is not set");
}
if (port == null) {
- port = type.getPort(storm_conf);
+ port = type.getPort(topoConf);
}
if (port<=0 && !type.isFake()) {
@@ -63,7 +63,7 @@ public class ThriftClient implements AutoCloseable {
_host = host;
_port = port;
_timeout = timeout;
- _conf = storm_conf;
+ _conf = topoConf;
_type = type;
_asUser = asUser;
if (!type.isFake()) {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
index 19d8dd6..2d86e47 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftConnectionType.java
@@ -66,7 +66,7 @@ public enum ThriftConnectionType {
return _isFake;
}
- public String getTransportPlugin(Map conf) {
+ public String getTransportPlugin(Map<String, Object> conf) {
String ret = (String)conf.get(_transConf);
if (ret == null) {
ret = (String)conf.get(Config.STORM_THRIFT_TRANSPORT_PLUGIN);
@@ -74,35 +74,35 @@ public enum ThriftConnectionType {
return ret;
}
- public int getPort(Map conf) {
+ public int getPort(Map<String, Object> conf) {
if (_isFake) {
return -1;
}
return ObjectReader.getInt(conf.get(_portConf));
}
- public Integer getQueueSize(Map conf) {
+ public Integer getQueueSize(Map<String, Object> conf) {
if (_qConf == null) {
return null;
}
return (Integer)conf.get(_qConf);
}
- public int getNumThreads(Map conf) {
+ public int getNumThreads(Map<String, Object> conf) {
if (_isFake) {
return 1;
}
return ObjectReader.getInt(conf.get(_threadsConf));
}
- public int getMaxBufferSize(Map conf) {
+ public int getMaxBufferSize(Map<String, Object> conf) {
if (_isFake) {
return 1;
}
return ObjectReader.getInt(conf.get(_buffConf));
}
- public Integer getSocketTimeOut(Map conf) {
+ public Integer getSocketTimeOut(Map<String, Object> conf) {
if (_socketTimeoutConf == null) {
return null;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
index 059b0d6..8fa8a7e 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
@@ -30,27 +30,27 @@ import org.slf4j.LoggerFactory;
public class ThriftServer {
private static final Logger LOG = LoggerFactory.getLogger(ThriftServer.class);
- private Map _storm_conf; //storm configuration
+ private Map _topoConf; //storm configuration
protected TProcessor _processor = null;
private final ThriftConnectionType _type;
private TServer _server;
private Configuration _login_conf;
private int _port;
- public ThriftServer(Map storm_conf, TProcessor processor, ThriftConnectionType type) {
- _storm_conf = storm_conf;
+ public ThriftServer(Map<String, Object> topoConf, TProcessor processor, ThriftConnectionType type) {
+ _topoConf = topoConf;
_processor = processor;
_type = type;
try {
//retrieve authentication configuration
- _login_conf = AuthUtils.GetConfiguration(_storm_conf);
+ _login_conf = AuthUtils.GetConfiguration(_topoConf);
} catch (Exception x) {
LOG.error(x.getMessage(), x);
}
try {
//locate our thrift transport plugin
- ITransportPlugin transportPlugin = AuthUtils.GetTransportPlugin(_type, _storm_conf, _login_conf);
+ ITransportPlugin transportPlugin = AuthUtils.GetTransportPlugin(_type, _topoConf, _login_conf);
//server
_server = transportPlugin.getServer(_processor);
_port = transportPlugin.getPort();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCAuthorizerBase.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCAuthorizerBase.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCAuthorizerBase.java
index 3447377..fdcd31e 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCAuthorizerBase.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCAuthorizerBase.java
@@ -34,7 +34,7 @@ public abstract class DRPCAuthorizerBase implements IAuthorizer {
public static final String FUNCTION_NAME = "function.name";
@Override
- public abstract void prepare(Map conf);
+ public abstract void prepare(Map<String, Object> conf);
abstract protected boolean permitClientRequest(ReqContext context, String operation, Map params);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
index cd6cab2..dc4d911 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DRPCSimpleACLAuthorizer.java
@@ -71,7 +71,7 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase {
long now = System.currentTimeMillis();
if ((now - 5000) > _lastUpdate || _acl == null) {
Map<String,AclFunctionEntry> acl = new HashMap<>();
- Map conf = Utils.findAndReadConfigFile(_aclFileName);
+ Map<String, Object> conf = Utils.findAndReadConfigFile(_aclFileName);
if (conf.containsKey(Config.DRPC_AUTHORIZER_ACL)) {
Map<String,Map<String,?>> confAcl =
(Map<String,Map<String,?>>)
@@ -99,7 +99,7 @@ public class DRPCSimpleACLAuthorizer extends DRPCAuthorizerBase {
}
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
Boolean isStrict =
(Boolean) conf.get(Config.DRPC_AUTHORIZER_ACL_STRICT);
_permitWhenMissingFunctionEntry =
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DenyAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DenyAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DenyAuthorizer.java
index 85f3ae4..a1c18ea 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DenyAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/DenyAuthorizer.java
@@ -31,17 +31,17 @@ public class DenyAuthorizer implements IAuthorizer {
* Invoked once immediately after construction
* @param conf Storm configuration
*/
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
}
/**
* permit() method is invoked for each incoming Thrift request
* @param context request context
* @param operation operation name
- * @param topology_conf configuration of targeted topology
+ * @param topoConf configuration of targeted topology
* @return true if the request is authorized, false if reject
*/
- public boolean permit(ReqContext context, String operation, Map topology_conf) {
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf) {
return false;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/ImpersonationAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/ImpersonationAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/ImpersonationAuthorizer.java
index 6f7c4de..4cb3145 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/ImpersonationAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/ImpersonationAuthorizer.java
@@ -37,7 +37,7 @@ public class ImpersonationAuthorizer implements IAuthorizer {
protected IGroupMappingServiceProvider _groupMappingProvider;
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
userImpersonationACL = new HashMap<>();
Map<String, Map<String, List<String>>> userToHostAndGroup = (Map<String, Map<String, List<String>>>) conf.get(Config.NIMBUS_IMPERSONATION_ACL);
@@ -56,7 +56,7 @@ public class ImpersonationAuthorizer implements IAuthorizer {
}
@Override
- public boolean permit(ReqContext context, String operation, Map topology_conf) {
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf) {
if (!context.isImpersonating()) {
LOG.debug("Not an impersonation attempt.");
return true;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/NoopAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/NoopAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/NoopAuthorizer.java
index fc11c18..68d117d 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/NoopAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/NoopAuthorizer.java
@@ -31,17 +31,17 @@ public class NoopAuthorizer implements IAuthorizer {
* Invoked once immediately after construction
* @param conf Storm configuration
*/
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
}
/**
* permit() method is invoked for each incoming Thrift request
* @param context request context includes info about
* @param operation operation name
- * @param topology_conf configuration of targeted topology
+ * @param topoConf configuration of targeted topology
* @return true if the request is authorized, false if reject
*/
- public boolean permit(ReqContext context, String operation, Map topology_conf) {
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf) {
return true;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
index 71705ce..783e28c 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleACLAuthorizer.java
@@ -84,7 +84,7 @@ public class SimpleACLAuthorizer implements IAuthorizer {
* @param conf Storm configuration
*/
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
_admins = new HashSet<>();
_supervisors = new HashSet<>();
_nimbusUsers = new HashSet<>();
@@ -112,11 +112,11 @@ public class SimpleACLAuthorizer implements IAuthorizer {
* permit() method is invoked for each incoming Thrift request
* @param context request context includes info about
* @param operation operation name
- * @param topology_conf configuration of targeted topology
+ * @param topoConf configuration of targeted topology
* @return true if the request is authorized, false if reject
*/
@Override
- public boolean permit(ReqContext context, String operation, Map topology_conf) {
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf) {
String principal = context.principal().getName();
String user = _ptol.toLocal(context.principal());
Set<String> userGroups = new HashSet<>();
@@ -143,8 +143,8 @@ public class SimpleACLAuthorizer implements IAuthorizer {
if (_topoCommands.contains(operation)) {
Set topoUsers = new HashSet<String>();
- if (topology_conf.containsKey(Config.TOPOLOGY_USERS)) {
- topoUsers.addAll((Collection<String>)topology_conf.get(Config.TOPOLOGY_USERS));
+ if (topoConf.containsKey(Config.TOPOLOGY_USERS)) {
+ topoUsers.addAll((Collection<String>)topoConf.get(Config.TOPOLOGY_USERS));
}
if (topoUsers.contains(principal) || topoUsers.contains(user)) {
@@ -152,8 +152,8 @@ public class SimpleACLAuthorizer implements IAuthorizer {
}
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));
+ if (topoConf.containsKey(Config.TOPOLOGY_GROUPS) && topoConf.get(Config.TOPOLOGY_GROUPS) != null) {
+ topoGroups.addAll((Collection<String>)topoConf.get(Config.TOPOLOGY_GROUPS));
}
if (checkUserGroupAllowed(userGroups, topoGroups)) return true;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java
index 7c89a6d..762500c 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/authorizer/SimpleWhitelistAuthorizer.java
@@ -39,7 +39,7 @@ public class SimpleWhitelistAuthorizer implements IAuthorizer {
* @param conf Storm configuration
*/
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
users = new HashSet<>();
if (conf.containsKey(WHITELIST_USERS_CONF)) {
users.addAll((Collection<String>)conf.get(WHITELIST_USERS_CONF));
@@ -50,11 +50,11 @@ public class SimpleWhitelistAuthorizer implements IAuthorizer {
* `permit()` method is invoked for each incoming Thrift request
* @param context request context includes info about
* @param operation operation name
- * @param topology_conf configuration of targeted topology
+ * @param topoConf configuration of targeted topology
* @return true if the request is authorized, false if reject
*/
@Override
- public boolean permit(ReqContext context, String operation, Map topology_conf) {
+ public boolean permit(ReqContext context, String operation, Map<String, Object> topoConf) {
return context.principal() != null ? users.contains(context.principal().getName()) : false;
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
index c3f8560..6d8057b 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
@@ -49,9 +49,9 @@ public class AutoTGT implements IAutoCredentials, ICredentialsRenewer {
private static final Logger LOG = LoggerFactory.getLogger(AutoTGT.class);
private static final float TICKET_RENEW_WINDOW = 0.80f;
protected static final AtomicReference<KerberosTicket> kerbTicket = new AtomicReference<>();
- private Map conf;
+ private Map<String, Object> conf;
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
this.conf = conf;
}
@@ -238,7 +238,7 @@ public class AutoTGT implements IAutoCredentials, ICredentialsRenewer {
}
@Override
- public void renew(Map<String,String> credentials, Map topologyConf) {
+ public void renew(Map<String,String> credentials, Map<String, Object> topologyConf) {
KerberosTicket tgt = getTGT(credentials);
if (tgt != null) {
long refreshTime = getRefreshTime(tgt);
@@ -257,7 +257,7 @@ public class AutoTGT implements IAutoCredentials, ICredentialsRenewer {
public static void main(String[] args) throws Exception {
AutoTGT at = new AutoTGT();
- Map conf = new java.util.HashMap();
+ Map<String, Object> conf = new java.util.HashMap();
conf.put("java.security.auth.login.config", args[0]);
at.prepare(conf);
Map<String,String> creds = new java.util.HashMap<String,String>();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
index d2df9dd..642be76 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
@@ -93,7 +93,7 @@ public class KerberosSaslTransportPlugin extends SaslTransportPlugin {
public TTransportFactory getServerTransportFactory() throws IOException {
//create an authentication callback handler
- CallbackHandler server_callback_handler = new ServerCallbackHandler(login_conf, storm_conf);
+ CallbackHandler server_callback_handler = new ServerCallbackHandler(login_conf, topoConf);
//login our principal
Subject subject = null;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
index ba2f4af..59eb80d 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
@@ -40,7 +40,7 @@ public class ServerCallbackHandler implements CallbackHandler {
private String userName;
- public ServerCallbackHandler(Configuration configuration, Map stormConf) throws IOException {
+ public ServerCallbackHandler(Configuration configuration, Map<String, Object> topoConf) throws IOException {
if (configuration==null) return;
AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_SERVER);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/serialization/BlowfishTupleSerializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/serialization/BlowfishTupleSerializer.java b/storm-client/src/jvm/org/apache/storm/security/serialization/BlowfishTupleSerializer.java
index 8cc2f23..00cca44 100644
--- a/storm-client/src/jvm/org/apache/storm/security/serialization/BlowfishTupleSerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/serialization/BlowfishTupleSerializer.java
@@ -49,10 +49,10 @@ public class BlowfishTupleSerializer extends Serializer<ListDelegate> {
private static final Logger LOG = LoggerFactory.getLogger(BlowfishTupleSerializer.class);
private BlowfishSerializer _serializer;
- public BlowfishTupleSerializer(Kryo kryo, Map storm_conf) {
+ public BlowfishTupleSerializer(Kryo kryo, Map<String, Object> topoConf) {
String encryption_key;
try {
- encryption_key = (String)storm_conf.get(SECRET_KEY);
+ encryption_key = (String)topoConf.get(SECRET_KEY);
LOG.debug("Blowfish serializer being constructed ...");
if (encryption_key == null) {
throw new RuntimeException("Blowfish encryption key not specified");
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java
index 89436f7..7a7cf00 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java
@@ -43,7 +43,7 @@ public class DefaultKryoFactory implements IKryoFactory {
}
@Override
- public Kryo getKryo(Map conf) {
+ public Kryo getKryo(Map<String, Object> conf) {
KryoSerializableDefault k = new KryoSerializableDefault();
k.setRegistrationRequired(!((Boolean) conf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION)));
k.setReferences(false);
@@ -51,14 +51,14 @@ public class DefaultKryoFactory implements IKryoFactory {
}
@Override
- public void preRegister(Kryo k, Map conf) {
+ public void preRegister(Kryo k, Map<String, Object> conf) {
}
- public void postRegister(Kryo k, Map conf) {
+ public void postRegister(Kryo k, Map<String, Object> conf) {
((KryoSerializableDefault)k).overrideDefault(true);
}
@Override
- public void postDecorate(Kryo k, Map conf) {
+ public void postDecorate(Kryo k, Map<String, Object> conf) {
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/DefaultSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/DefaultSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/DefaultSerializationDelegate.java
index f104481..60b3982 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/DefaultSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/DefaultSerializationDelegate.java
@@ -24,7 +24,7 @@ import java.util.Map;
public class DefaultSerializationDelegate implements SerializationDelegate {
@Override
- public void prepare(Map stormConf) {
+ public void prepare(Map<String, Object> topoConf) {
// No-op
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeSerializationDelegate.java
index 93b725b..8b4b207 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeSerializationDelegate.java
@@ -32,9 +32,9 @@ public class GzipBridgeSerializationDelegate implements SerializationDelegate {
private GzipSerializationDelegate gzipDelegate = new GzipSerializationDelegate();
@Override
- public void prepare(Map stormConf) {
- defaultDelegate.prepare(stormConf);
- gzipDelegate.prepare(stormConf);
+ public void prepare(Map<String, Object> topoConf) {
+ defaultDelegate.prepare(topoConf);
+ gzipDelegate.prepare(topoConf);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java
index 4b072e4..0ab494b 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java
@@ -31,9 +31,9 @@ public class GzipBridgeThriftSerializationDelegate implements SerializationDeleg
private GzipThriftSerializationDelegate gzipDelegate = new GzipThriftSerializationDelegate();
@Override
- public void prepare(Map stormConf) {
- defaultDelegate.prepare(stormConf);
- gzipDelegate.prepare(stormConf);
+ public void prepare(Map<String, Object> topoConf) {
+ defaultDelegate.prepare(topoConf);
+ gzipDelegate.prepare(topoConf);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java
index 7d64628..18e50ba 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java
@@ -28,7 +28,7 @@ import java.util.zip.GZIPOutputStream;
public class GzipSerializationDelegate implements SerializationDelegate {
@Override
- public void prepare(Map stormConf) {
+ public void prepare(Map<String, Object> topoConf) {
// No-op
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java
index 867ac7d..bfec065 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java
@@ -31,7 +31,7 @@ import org.apache.thrift.TSerializer;
public class GzipThriftSerializationDelegate implements SerializationDelegate {
@Override
- public void prepare(Map stormConf) {
+ public void prepare(Map<String, Object> topoConf) {
// No-op
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/IKryoFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/IKryoFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/IKryoFactory.java
index 409e7dd..ef6fec5 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/IKryoFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/IKryoFactory.java
@@ -33,8 +33,8 @@ import java.util.Map;
* 7. Storm calls postDecorate hook
*/
public interface IKryoFactory {
- Kryo getKryo(Map conf);
- void preRegister(Kryo k, Map conf);
- void postRegister(Kryo k, Map conf);
- void postDecorate(Kryo k, Map conf);
+ Kryo getKryo(Map<String, Object> conf);
+ void preRegister(Kryo k, Map<String, Object> conf);
+ void postRegister(Kryo k, Map<String, Object> conf);
+ void postDecorate(Kryo k, Map<String, Object> conf);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
index 4e877a8..f2d8bf1 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
@@ -32,7 +32,7 @@ public class KryoTupleDeserializer implements ITupleDeserializer {
SerializationFactory.IdDictionary _ids;
Input _kryoInput;
- public KryoTupleDeserializer(final Map conf, final GeneralTopologyContext context) {
+ public KryoTupleDeserializer(final Map<String, Object> conf, final GeneralTopologyContext context) {
_kryo = new KryoValuesDeserializer(conf);
_context = context;
_ids = new SerializationFactory.IdDictionary(context.getRawTopology());
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
index 3151c33..7302278 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
@@ -28,7 +28,7 @@ public class KryoTupleSerializer implements ITupleSerializer {
SerializationFactory.IdDictionary _ids;
Output _kryoOut;
- public KryoTupleSerializer(final Map conf, final GeneralTopologyContext context) {
+ public KryoTupleSerializer(final Map<String, Object> conf, final GeneralTopologyContext context) {
_kryo = new KryoValuesSerializer(conf);
_kryoOut = new Output(2000, 2000000000);
_ids = new SerializationFactory.IdDictionary(context.getRawTopology());
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesDeserializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesDeserializer.java b/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesDeserializer.java
index 0e9edea..2ef326a 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesDeserializer.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesDeserializer.java
@@ -28,7 +28,7 @@ public class KryoValuesDeserializer {
Kryo _kryo;
Input _kryoInput;
- public KryoValuesDeserializer(Map conf) {
+ public KryoValuesDeserializer(Map<String, Object> conf) {
_kryo = SerializationFactory.getKryo(conf);
_kryoInput = new Input(1);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java b/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java
index 2d936a9..af41e15 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java
@@ -29,7 +29,7 @@ public class KryoValuesSerializer {
ListDelegate _delegate;
Output _kryoOut;
- public KryoValuesSerializer(Map conf) {
+ public KryoValuesSerializer(Map<String, Object> conf) {
_kryo = SerializationFactory.getKryo(conf);
_delegate = new ListDelegate();
_kryoOut = new Output(2000, 2000000000);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/SerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/SerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/SerializationDelegate.java
index 7087397..2c8c27b 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/SerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/SerializationDelegate.java
@@ -27,7 +27,7 @@ public interface SerializationDelegate {
/**
* Lifecycle step that will be called after instantiating with nullary constructor.
*/
- void prepare(Map stormConf);
+ void prepare(Map<String, Object> topoConf);
byte[] serialize(Object object);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
index fbe4cfa..1faa926 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
@@ -48,7 +48,7 @@ public class SerializationFactory {
public static final Logger LOG = LoggerFactory.getLogger(SerializationFactory.class);
public static final ServiceLoader<SerializationRegister> loader = ServiceLoader.load(SerializationRegister.class);
- public static Kryo getKryo(Map conf) {
+ public static Kryo getKryo(Map<String, Object> conf) {
IKryoFactory kryoFactory = (IKryoFactory) ReflectionUtils.newInstance((String) conf.get(Config.TOPOLOGY_KRYO_FACTORY));
Kryo k = kryoFactory.getKryo(conf);
k.register(byte[].class);
@@ -188,7 +188,7 @@ public class SerializationFactory {
}
}
- private static Serializer resolveSerializerInstance(Kryo k, Class superClass, Class<? extends Serializer> serializerClass, Map conf) {
+ private static Serializer resolveSerializerInstance(Kryo k, Class superClass, Class<? extends Serializer> serializerClass, Map<String, Object> conf) {
try {
try {
return serializerClass.getConstructor(Kryo.class, Class.class, Map.class).newInstance(k, superClass, conf);
@@ -223,7 +223,7 @@ public class SerializationFactory {
}
}
- private static Map<String, String> normalizeKryoRegister(Map conf) {
+ private static Map<String, String> normalizeKryoRegister(Map<String, Object> 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<>();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/serialization/ThriftSerializationDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/serialization/ThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/ThriftSerializationDelegate.java
index ed3cdf6..a2dfecf 100644
--- a/storm-client/src/jvm/org/apache/storm/serialization/ThriftSerializationDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/serialization/ThriftSerializationDelegate.java
@@ -27,7 +27,7 @@ import java.util.Map;
public class ThriftSerializationDelegate implements SerializationDelegate {
@Override
- public void prepare(Map stormConf) {
+ public void prepare(Map<String, Object> topoConf) {
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/CheckpointSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/CheckpointSpout.java b/storm-client/src/jvm/org/apache/storm/spout/CheckpointSpout.java
index 14d6572..798ebff 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/CheckpointSpout.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/CheckpointSpout.java
@@ -63,7 +63,7 @@ public class CheckpointSpout extends BaseRichSpout {
private CheckPointState curTxState;
@Override
- public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
open(context, collector, loadCheckpointInterval(conf), loadCheckpointState(conf, context));
}
@@ -132,7 +132,7 @@ public class CheckpointSpout extends BaseRichSpout {
/**
* Loads the last saved checkpoint state the from persistent storage.
*/
- private KeyValueState<String, CheckPointState> loadCheckpointState(Map conf, TopologyContext ctx) {
+ private KeyValueState<String, CheckPointState> loadCheckpointState(Map<String, Object> conf, TopologyContext ctx) {
String namespace = ctx.getThisComponentId() + "-" + ctx.getThisTaskId();
KeyValueState<String, CheckPointState> state =
(KeyValueState<String, CheckPointState>) StateFactory.getState(namespace, conf, ctx);
@@ -147,10 +147,10 @@ public class CheckpointSpout extends BaseRichSpout {
return state;
}
- private int loadCheckpointInterval(Map stormConf) {
+ private int loadCheckpointInterval(Map<String, Object> topoConf) {
int interval = 0;
- if (stormConf.containsKey(Config.TOPOLOGY_STATE_CHECKPOINT_INTERVAL)) {
- interval = ((Number) stormConf.get(Config.TOPOLOGY_STATE_CHECKPOINT_INTERVAL)).intValue();
+ if (topoConf.containsKey(Config.TOPOLOGY_STATE_CHECKPOINT_INTERVAL)) {
+ interval = ((Number) topoConf.get(Config.TOPOLOGY_STATE_CHECKPOINT_INTERVAL)).intValue();
}
// ensure checkpoint interval is not less than a sane low value.
interval = Math.max(100, interval);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/ISpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/ISpout.java b/storm-client/src/jvm/org/apache/storm/spout/ISpout.java
index 48986c4..f889fc3 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/ISpout.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/ISpout.java
@@ -54,7 +54,7 @@ public interface ISpout extends Serializable {
* @param context This object can be used to get information about this task's place within the topology, including the task id and component id of this task, input and output information, etc.
* @param collector The collector is used to emit tuples from this spout. Tuples can be emitted at any time, including the open and close methods. The collector is thread-safe and should be saved as an instance variable of this spout object.
*/
- void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
+ void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector);
/**
* Called when an ISpout is going to be shutdown. There is no guarentee that close
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/ISpoutWaitStrategy.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/ISpoutWaitStrategy.java b/storm-client/src/jvm/org/apache/storm/spout/ISpoutWaitStrategy.java
index 3f1fcf6..3726f12 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/ISpoutWaitStrategy.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/ISpoutWaitStrategy.java
@@ -29,6 +29,6 @@ import java.util.Map;
* The default strategy sleeps for one millisecond.
*/
public interface ISpoutWaitStrategy {
- void prepare(Map conf);
+ void prepare(Map<String, Object> conf);
void emptyEmit(long streak);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/NothingEmptyEmitStrategy.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/NothingEmptyEmitStrategy.java b/storm-client/src/jvm/org/apache/storm/spout/NothingEmptyEmitStrategy.java
index a334dc3..cc1f1eb 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/NothingEmptyEmitStrategy.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/NothingEmptyEmitStrategy.java
@@ -25,7 +25,7 @@ public class NothingEmptyEmitStrategy implements ISpoutWaitStrategy {
}
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
throw new UnsupportedOperationException("Not supported yet.");
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/ShellSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/ShellSpout.java b/storm-client/src/jvm/org/apache/storm/spout/ShellSpout.java
index 72b7f17..eb65c9a 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/ShellSpout.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/ShellSpout.java
@@ -92,15 +92,15 @@ public class ShellSpout implements ISpout {
this.changeDirectory = changeDirectory;
}
- public void open(Map stormConf, TopologyContext context,
+ public void open(Map<String, Object> topoConf, TopologyContext context,
SpoutOutputCollector collector) {
_collector = collector;
_context = context;
- if (stormConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) {
- workerTimeoutMills = 1000 * ObjectReader.getInt(stormConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS));
+ if (topoConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) {
+ workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS));
} else {
- workerTimeoutMills = 1000 * ObjectReader.getInt(stormConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
+ workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
}
_process = new ShellProcess(_command);
@@ -108,7 +108,7 @@ public class ShellSpout implements ISpout {
_process.setEnv(env);
}
- Number subpid = _process.launch(stormConf, context, changeDirectory);
+ Number subpid = _process.launch(topoConf, context, changeDirectory);
LOG.info("Launched subprocess with pid " + subpid);
heartBeatExecutorService = MoreExecutors.getExitingScheduledExecutorService(new ScheduledThreadPoolExecutor(1));
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/spout/SleepSpoutWaitStrategy.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/spout/SleepSpoutWaitStrategy.java b/storm-client/src/jvm/org/apache/storm/spout/SleepSpoutWaitStrategy.java
index b67253a..c7826f9 100644
--- a/storm-client/src/jvm/org/apache/storm/spout/SleepSpoutWaitStrategy.java
+++ b/storm-client/src/jvm/org/apache/storm/spout/SleepSpoutWaitStrategy.java
@@ -26,7 +26,7 @@ public class SleepSpoutWaitStrategy implements ISpoutWaitStrategy {
long sleepMillis;
@Override
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
sleepMillis = ((Number) conf.get(Config.TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS)).longValue();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/state/IStateSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/state/IStateSpout.java b/storm-client/src/jvm/org/apache/storm/state/IStateSpout.java
index 19026d2..de523ce 100644
--- a/storm-client/src/jvm/org/apache/storm/state/IStateSpout.java
+++ b/storm-client/src/jvm/org/apache/storm/state/IStateSpout.java
@@ -22,7 +22,7 @@ import java.io.Serializable;
import java.util.Map;
public interface IStateSpout extends Serializable {
- void open(Map conf, TopologyContext context);
+ void open(Map<String, Object> conf, TopologyContext context);
void close();
void nextTuple(StateSpoutOutputCollector collector);
void synchronize(SynchronizeOutputCollector collector);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/state/InMemoryKeyValueStateProvider.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/state/InMemoryKeyValueStateProvider.java b/storm-client/src/jvm/org/apache/storm/state/InMemoryKeyValueStateProvider.java
index bc33241..624f12e 100644
--- a/storm-client/src/jvm/org/apache/storm/state/InMemoryKeyValueStateProvider.java
+++ b/storm-client/src/jvm/org/apache/storm/state/InMemoryKeyValueStateProvider.java
@@ -29,7 +29,7 @@ public class InMemoryKeyValueStateProvider implements StateProvider {
private final ConcurrentHashMap<String, State> states = new ConcurrentHashMap<>();
@Override
- public State newState(String namespace, Map stormConf, TopologyContext context) {
+ public State newState(String namespace, Map<String, Object> topoConf, TopologyContext context) {
State state = states.get(namespace);
if (state == null) {
State newState = new InMemoryKeyValueState<>();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/state/StateFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/state/StateFactory.java b/storm-client/src/jvm/org/apache/storm/state/StateFactory.java
index d5ec7ae..79e6d96 100644
--- a/storm-client/src/jvm/org/apache/storm/state/StateFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/state/StateFactory.java
@@ -37,23 +37,23 @@ public class StateFactory {
* {@link InMemoryKeyValueState} if no provider is configured.
*
* @param namespace the state namespace
- * @param stormConf the storm conf
+ * @param topoConf the storm conf
* @param context the topology context
* @return the state instance
*/
- public static State getState(String namespace, Map stormConf, TopologyContext context) {
+ public static State getState(String namespace, Map<String, Object> topoConf, TopologyContext context) {
State state;
try {
String provider = null;
- if (stormConf.containsKey(Config.TOPOLOGY_STATE_PROVIDER)) {
- provider = (String) stormConf.get(Config.TOPOLOGY_STATE_PROVIDER);
+ if (topoConf.containsKey(Config.TOPOLOGY_STATE_PROVIDER)) {
+ provider = (String) topoConf.get(Config.TOPOLOGY_STATE_PROVIDER);
} else {
provider = DEFAULT_PROVIDER;
}
Class<?> klazz = Class.forName(provider);
Object object = klazz.newInstance();
if (object instanceof StateProvider) {
- state = ((StateProvider) object).newState(namespace, stormConf, context);
+ state = ((StateProvider) object).newState(namespace, topoConf, context);
} else {
String msg = "Invalid state provider '" + provider +
"'. Should implement org.apache.storm.state.StateProvider";
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/state/StateProvider.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/state/StateProvider.java b/storm-client/src/jvm/org/apache/storm/state/StateProvider.java
index 8c09a0e..2b2aad2 100644
--- a/storm-client/src/jvm/org/apache/storm/state/StateProvider.java
+++ b/storm-client/src/jvm/org/apache/storm/state/StateProvider.java
@@ -30,9 +30,9 @@ public interface StateProvider {
* the componentid-task of the task, so that each task can have its own unique state.
*
* @param namespace a namespace of the state
- * @param stormConf the storm topology configuration
+ * @param topoConf the storm topology configuration
* @param context the {@link TopologyContext}
* @return a previously saved state if one exists otherwise a newly initialized state.
*/
- State newState(String namespace, Map stormConf, TopologyContext context);
+ State newState(String namespace, Map<String, Object> topoConf, TopologyContext context);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java
index f1163ca..6f3853d 100644
--- a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBolt.java
@@ -39,8 +39,8 @@ class ProcessorBolt extends BaseRichBolt implements StreamBolt {
}
@Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- delegate.prepare(stormConf, context, collector);
+ public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
+ delegate.prepare(topoConf, context, collector);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java
index 3f3b5c1..1329bcf 100644
--- a/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java
+++ b/storm-client/src/jvm/org/apache/storm/streams/ProcessorBoltDelegate.java
@@ -53,7 +53,7 @@ class ProcessorBoltDelegate implements Serializable {
private final String id;
private final DirectedGraph<Node, Edge> graph;
private final List<ProcessorNode> nodes;
- private Map stormConf;
+ private Map<String, Object> topoConf;
private TopologyContext topologyContext;
private OutputCollector outputCollector;
private final List<ProcessorNode> outgoingProcessors = new ArrayList<>();
@@ -81,8 +81,8 @@ class ProcessorBoltDelegate implements Serializable {
return Collections.unmodifiableList(nodes);
}
- void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- this.stormConf = stormConf;
+ void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
+ this.topoConf = topoConf;
topologyContext = context;
outputCollector = collector;
DirectedSubgraph<Node, Edge> subgraph = new DirectedSubgraph<>(graph, new HashSet<>(nodes), null);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java
index c123658..4e40292 100644
--- a/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/streams/StatefulProcessorBolt.java
@@ -47,8 +47,8 @@ class StatefulProcessorBolt<K, V> extends BaseStatefulBolt<KeyValueState<K, V>>
}
@Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- delegate.prepare(stormConf, context, collector);
+ public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
+ delegate.prepare(topoConf, context, collector);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/streams/StreamState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/streams/StreamState.java b/storm-client/src/jvm/org/apache/storm/streams/StreamState.java
index a4633f7..814f400 100644
--- a/storm-client/src/jvm/org/apache/storm/streams/StreamState.java
+++ b/storm-client/src/jvm/org/apache/storm/streams/StreamState.java
@@ -40,4 +40,4 @@ public class StreamState<K, V> implements Serializable {
ProcessorNode getUpdateStateNode() {
return (ProcessorNode) stream.node;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java b/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java
index 6fc4e69..7207b12 100644
--- a/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/streams/WindowedProcessorBolt.java
@@ -54,8 +54,8 @@ class WindowedProcessorBolt extends BaseWindowedBolt implements StreamBolt {
}
@Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
- delegate.prepare(stormConf, context, collector);
+ public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
+ delegate.prepare(topoConf, context, collector);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
index 9acd655..6614f94 100644
--- a/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
+++ b/storm-client/src/jvm/org/apache/storm/task/GeneralTopologyContext.java
@@ -41,14 +41,14 @@ public class GeneralTopologyContext implements JSONAware {
private Map<String, List<Integer>> _componentToTasks;
private Map<String, Map<String, Fields>> _componentToStreamToFields;
private String _stormId;
- protected Map _stormConf;
+ protected Map _topoConf;
// pass in componentToSortedTasks for the case of running tons of tasks in single executor
- public GeneralTopologyContext(StormTopology topology, Map stormConf,
+ public GeneralTopologyContext(StormTopology topology, Map<String, Object> topoConf,
Map<Integer, String> taskToComponent, Map<String, List<Integer>> componentToSortedTasks,
Map<String, Map<String, Fields>> componentToStreamToFields, String stormId) {
_topology = topology;
- _stormConf = stormConf;
+ _topoConf = topoConf;
_taskToComponent = taskToComponent;
_stormId = stormId;
_componentToTasks = componentToSortedTasks;
@@ -183,13 +183,13 @@ public class GeneralTopologyContext implements JSONAware {
}
public int maxTopologyMessageTimeout() {
- Integer max = ObjectReader.getInt(_stormConf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS));
+ Integer max = ObjectReader.getInt(_topoConf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS));
for(String spout: getRawTopology().get_spouts().keySet()) {
ComponentCommon common = getComponentCommon(spout);
String jsonConf = common.get_json_conf();
if(jsonConf!=null) {
try {
- Map conf = (Map) JSONValue.parseWithException(jsonConf);
+ Map<String, Object> conf = (Map) JSONValue.parseWithException(jsonConf);
Object comp = conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS);
max = Math.max(ObjectReader.getInt(comp, max), max);
} catch (ParseException e) {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/task/IBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/IBolt.java b/storm-client/src/jvm/org/apache/storm/task/IBolt.java
index a2a026e..2268b6a 100644
--- a/storm-client/src/jvm/org/apache/storm/task/IBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/task/IBolt.java
@@ -48,11 +48,11 @@ public interface IBolt extends Serializable {
*
* This includes the:
*
- * @param stormConf The Storm configuration for this bolt. This is the configuration provided to the topology merged in with cluster configuration on this machine.
+ * @param topoConf The Storm configuration for this bolt. This is the configuration provided to the topology merged in with cluster configuration on this machine.
* @param context This object can be used to get information about this task's place within the topology, including the task id and component id of this task, input and output information, etc.
* @param collector The collector is used to emit tuples from this bolt. Tuples can be emitted at any time, including the prepare and cleanup methods. The collector is thread-safe and should be saved as an instance variable of this bolt object.
*/
- void prepare(Map stormConf, TopologyContext context, OutputCollector collector);
+ void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector);
/**
* Process a single tuple of input. The Tuple object contains metadata on it
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/task/ShellBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/ShellBolt.java b/storm-client/src/jvm/org/apache/storm/task/ShellBolt.java
index 0acb409..82b06bc 100644
--- a/storm-client/src/jvm/org/apache/storm/task/ShellBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/task/ShellBolt.java
@@ -124,12 +124,12 @@ public class ShellBolt implements IBolt {
this.changeDirectory = changeDirectory;
}
- public void prepare(Map stormConf, TopologyContext context,
+ public void prepare(Map<String, Object> topoConf, TopologyContext context,
final OutputCollector collector) {
- if (ConfigUtils.isLocalMode(stormConf)) {
+ if (ConfigUtils.isLocalMode(topoConf)) {
_isLocalMode = true;
}
- Object maxPending = stormConf.get(Config.TOPOLOGY_SHELLBOLT_MAX_PENDING);
+ Object maxPending = topoConf.get(Config.TOPOLOGY_SHELLBOLT_MAX_PENDING);
if (maxPending != null) {
this._pendingWrites = new ShellBoltMessageQueue(((Number)maxPending).intValue());
}
@@ -139,10 +139,10 @@ public class ShellBolt implements IBolt {
_context = context;
- if (stormConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) {
- workerTimeoutMills = 1000 * ObjectReader.getInt(stormConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS));
+ if (topoConf.containsKey(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS)) {
+ workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_SUBPROCESS_TIMEOUT_SECS));
} else {
- workerTimeoutMills = 1000 * ObjectReader.getInt(stormConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
+ workerTimeoutMills = 1000 * ObjectReader.getInt(topoConf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS));
}
_process = new ShellProcess(_command);
@@ -151,7 +151,7 @@ public class ShellBolt implements IBolt {
}
//subprocesses must send their pid first thing
- Number subpid = _process.launch(stormConf, context, changeDirectory);
+ Number subpid = _process.launch(topoConf, context, changeDirectory);
LOG.info("Launched subprocess with pid " + subpid);
// reader
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java
index 5ac8347..4c226ef 100644
--- a/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java
+++ b/storm-client/src/jvm/org/apache/storm/task/TopologyContext.java
@@ -60,7 +60,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
public TopologyContext(StormTopology topology,
- Map stormConf,
+ Map<String, Object> topoConf,
Map<Integer, String> taskToComponent,
Map<String, List<Integer>> componentToSortedTasks,
Map<String, Map<String, Fields>> componentToStreamToFields,
@@ -75,7 +75,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
Map<String, Object> executorData,
Map<Integer, Map<Integer, Map<String, IMetric>>> registeredMetrics,
AtomicBoolean openOrPrepareWasCalled) {
- super(topology, stormConf, taskToComponent, componentToSortedTasks,
+ super(topology, topoConf, taskToComponent, componentToSortedTasks,
componentToStreamToFields, stormId, codeDir, pidDir,
workerPort, workerTasks, defaultResources, userResources);
_taskId = taskId;
@@ -261,7 +261,7 @@ public class TopologyContext extends WorkerTopologyContext implements IMetricsCo
}
public void addTaskHook(ITaskHook hook) {
- hook.prepare(_stormConf, this);
+ hook.prepare(_topoConf, this);
_hooks.add(hook);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java b/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
index 89ba72e..2817b65 100644
--- a/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
+++ b/storm-client/src/jvm/org/apache/storm/task/WorkerTopologyContext.java
@@ -37,7 +37,7 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
public WorkerTopologyContext(
StormTopology topology,
- Map stormConf,
+ Map<String, Object> topoConf,
Map<Integer, String> taskToComponent,
Map<String, List<Integer>> componentToSortedTasks,
Map<String, Map<String, Fields>> componentToStreamToFields,
@@ -49,7 +49,7 @@ public class WorkerTopologyContext extends GeneralTopologyContext {
Map<String, Object> defaultResources,
Map<String, Object> userResources
) {
- super(topology, stormConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId);
+ super(topology, topoConf, taskToComponent, componentToSortedTasks, componentToStreamToFields, stormId);
_codeDir = codeDir;
_defaultResources = defaultResources;
_userResources = userResources;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/testing/BatchNumberList.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/testing/BatchNumberList.java b/storm-client/src/jvm/org/apache/storm/testing/BatchNumberList.java
index b958243..6f7b83d 100644
--- a/storm-client/src/jvm/org/apache/storm/testing/BatchNumberList.java
+++ b/storm-client/src/jvm/org/apache/storm/testing/BatchNumberList.java
@@ -48,7 +48,7 @@ public class BatchNumberList extends BaseBatchBolt<Object> {
@Override
- public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
+ public void prepare(Map<String, Object> conf, TopologyContext context, BatchOutputCollector collector, Object id) {
_collector = collector;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/testing/CountingBatchBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/testing/CountingBatchBolt.java b/storm-client/src/jvm/org/apache/storm/testing/CountingBatchBolt.java
index 982eb87..f421aee 100644
--- a/storm-client/src/jvm/org/apache/storm/testing/CountingBatchBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/testing/CountingBatchBolt.java
@@ -32,7 +32,7 @@ public class CountingBatchBolt extends BaseBatchBolt<Object> {
int _count = 0;
@Override
- public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) {
+ public void prepare(Map<String, Object> conf, TopologyContext context, BatchOutputCollector collector, Object id) {
_collector = collector;
_id = id;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/testing/CountingCommitBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/testing/CountingCommitBolt.java b/storm-client/src/jvm/org/apache/storm/testing/CountingCommitBolt.java
index 2e12a02..b76ca88 100644
--- a/storm-client/src/jvm/org/apache/storm/testing/CountingCommitBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/testing/CountingCommitBolt.java
@@ -34,7 +34,7 @@ public class CountingCommitBolt extends BaseTransactionalBolt implements ICommit
int _count = 0;
@Override
- public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, TransactionAttempt id) {
+ public void prepare(Map<String, Object> conf, TopologyContext context, BatchOutputCollector collector, TransactionAttempt id) {
_id = id;
_collector = collector;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/testing/FeederSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/testing/FeederSpout.java b/storm-client/src/jvm/org/apache/storm/testing/FeederSpout.java
index 7801ca0..cb6b0ff 100644
--- a/storm-client/src/jvm/org/apache/storm/testing/FeederSpout.java
+++ b/storm-client/src/jvm/org/apache/storm/testing/FeederSpout.java
@@ -66,7 +66,7 @@ public class FeederSpout extends BaseRichSpout {
InprocMessaging.waitForReader(_id);
}
- public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
_collector = collector;
}
@@ -104,4 +104,4 @@ public class FeederSpout extends BaseRichSpout {
public Map<String, Object> getComponentConfiguration() {
return new HashMap<String, Object>();
}
-}
\ No newline at end of file
+}