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:35 UTC
[08/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/executor/bolt/BoltExecutor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java
index c3a18ad..9e52cb2 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/bolt/BoltExecutor.java
@@ -50,7 +50,7 @@ public class BoltExecutor extends Executor {
public BoltExecutor(WorkerState workerData, List<Long> executorId, Map<String, String> credentials) {
super(workerData, executorId, credentials);
- this.executeSampler = ConfigUtils.mkStatsSampler(stormConf);
+ this.executeSampler = ConfigUtils.mkStatsSampler(topoConf);
}
public void init(Map<Integer, Task> idToTask) {
@@ -63,25 +63,25 @@ public class BoltExecutor extends Executor {
Task taskData = entry.getValue();
IBolt boltObject = (IBolt) taskData.getTaskObject();
TopologyContext userContext = taskData.getUserContext();
- taskData.getBuiltInMetrics().registerAll(stormConf, userContext);
+ taskData.getBuiltInMetrics().registerAll(topoConf, userContext);
if (boltObject instanceof ICredentialsListener) {
((ICredentialsListener) boltObject).setCredentials(credentials);
}
if (Constants.SYSTEM_COMPONENT_ID.equals(componentId)) {
Map<String, DisruptorQueue> map = ImmutableMap.of("sendqueue", transferQueue, "receive", receiveQueue,
"transfer", workerData.getTransferQueue());
- BuiltinMetricsUtil.registerQueueMetrics(map, stormConf, userContext);
+ BuiltinMetricsUtil.registerQueueMetrics(map, topoConf, userContext);
Map cachedNodePortToSocket = (Map) workerData.getCachedNodeToPortSocket().get();
- BuiltinMetricsUtil.registerIconnectionClientMetrics(cachedNodePortToSocket, stormConf, userContext);
- BuiltinMetricsUtil.registerIconnectionServerMetric(workerData.getReceiver(), stormConf, userContext);
+ BuiltinMetricsUtil.registerIconnectionClientMetrics(cachedNodePortToSocket, topoConf, userContext);
+ BuiltinMetricsUtil.registerIconnectionServerMetric(workerData.getReceiver(), topoConf, userContext);
} else {
Map<String, DisruptorQueue> map = ImmutableMap.of("sendqueue", transferQueue, "receive", receiveQueue);
- BuiltinMetricsUtil.registerQueueMetrics(map, stormConf, userContext);
+ BuiltinMetricsUtil.registerQueueMetrics(map, topoConf, userContext);
}
IOutputCollector outputCollector = new BoltOutputCollectorImpl(this, taskData, entry.getKey(), rand, hasEventLoggers, isDebug);
- boltObject.prepare(stormConf, userContext, new OutputCollector(outputCollector));
+ boltObject.prepare(topoConf, userContext, new OutputCollector(outputCollector));
}
openOrPrepareWasCalled.set(true);
LOG.info("Prepared bolt {}:{}", componentId, idToTask.keySet());
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/executor/error/ReportError.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/executor/error/ReportError.java b/storm-client/src/jvm/org/apache/storm/executor/error/ReportError.java
index 3272b85..d1c80e3 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/error/ReportError.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/error/ReportError.java
@@ -34,7 +34,7 @@ public class ReportError implements IReportError {
private static final Logger LOG = LoggerFactory.getLogger(ReportError.class);
- private final Map stormConf;
+ private final Map<String, Object> topoConf;
private final IStormClusterState stormClusterState;
private final String stormId;
private final String componentId;
@@ -45,14 +45,14 @@ public class ReportError implements IReportError {
private AtomicInteger intervalStartTime;
private AtomicInteger intervalErrors;
- public ReportError(Map stormConf, IStormClusterState stormClusterState, String stormId, String componentId, WorkerTopologyContext workerTopologyContext) {
- this.stormConf = stormConf;
+ public ReportError(Map<String, Object> topoConf, IStormClusterState stormClusterState, String stormId, String componentId, WorkerTopologyContext workerTopologyContext) {
+ this.topoConf = topoConf;
this.stormClusterState = stormClusterState;
this.stormId = stormId;
this.componentId = componentId;
this.workerTopologyContext = workerTopologyContext;
- this.errorIntervalSecs = ObjectReader.getInt(stormConf.get(Config.TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS));
- this.maxPerInterval = ObjectReader.getInt(stormConf.get(Config.TOPOLOGY_MAX_ERROR_REPORT_PER_INTERVAL));
+ this.errorIntervalSecs = ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS));
+ this.maxPerInterval = ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_MAX_ERROR_REPORT_PER_INTERVAL));
this.intervalStartTime = new AtomicInteger(Time.currentTimeSecs());
this.intervalErrors = new AtomicInteger(0);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java
index 9e7622c..36f1892 100644
--- a/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java
+++ b/storm-client/src/jvm/org/apache/storm/executor/spout/SpoutExecutor.java
@@ -70,13 +70,13 @@ public class SpoutExecutor extends Executor {
public SpoutExecutor(final WorkerState workerData, final List<Long> executorId, Map<String, String> credentials) {
super(workerData, executorId, credentials);
- this.spoutWaitStrategy = ReflectionUtils.newInstance((String) stormConf.get(Config.TOPOLOGY_SPOUT_WAIT_STRATEGY));
- this.spoutWaitStrategy.prepare(stormConf);
+ this.spoutWaitStrategy = ReflectionUtils.newInstance((String) topoConf.get(Config.TOPOLOGY_SPOUT_WAIT_STRATEGY));
+ this.spoutWaitStrategy.prepare(topoConf);
- this.backPressureEnabled = ObjectReader.getBoolean(stormConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false);
+ this.backPressureEnabled = ObjectReader.getBoolean(topoConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false);
this.lastActive = new AtomicBoolean(false);
- this.hasAckers = StormCommon.hasAckers(stormConf);
+ this.hasAckers = StormCommon.hasAckers(topoConf);
this.emittedCount = new MutableLong(0);
this.emptyEmitStreak = new MutableLong(0);
this.spoutThrottlingMetrics = new SpoutThrottlingMetrics();
@@ -89,7 +89,7 @@ public class SpoutExecutor extends Executor {
LOG.info("Opening spout {}:{}", componentId, idToTask.keySet());
this.idToTask = idToTask;
- this.maxSpoutPending = ObjectReader.getInt(stormConf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 0) * idToTask.size();
+ this.maxSpoutPending = ObjectReader.getInt(topoConf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING), 0) * idToTask.size();
this.spouts = new ArrayList<>();
for (Task task : idToTask.values()) {
this.spouts.add((ISpout) task.getTaskObject());
@@ -105,7 +105,7 @@ public class SpoutExecutor extends Executor {
}
});
- this.spoutThrottlingMetrics.registerAll(stormConf, idToTask.values().iterator().next().getUserContext());
+ this.spoutThrottlingMetrics.registerAll(topoConf, idToTask.values().iterator().next().getUserContext());
this.outputCollectors = new ArrayList<>();
for (Map.Entry<Integer, Task> entry : idToTask.entrySet()) {
Task taskData = entry.getValue();
@@ -116,14 +116,14 @@ public class SpoutExecutor extends Executor {
SpoutOutputCollector outputCollector = new SpoutOutputCollector(spoutOutputCollector);
this.outputCollectors.add(outputCollector);
- taskData.getBuiltInMetrics().registerAll(stormConf, taskData.getUserContext());
+ taskData.getBuiltInMetrics().registerAll(topoConf, taskData.getUserContext());
Map<String, DisruptorQueue> map = ImmutableMap.of("sendqueue", transferQueue, "receive", receiveQueue);
- BuiltinMetricsUtil.registerQueueMetrics(map, stormConf, taskData.getUserContext());
+ BuiltinMetricsUtil.registerQueueMetrics(map, topoConf, taskData.getUserContext());
if (spoutObject instanceof ICredentialsListener) {
((ICredentialsListener) spoutObject).setCredentials(credentials);
}
- spoutObject.open(stormConf, taskData.getUserContext(), outputCollector);
+ spoutObject.open(topoConf, taskData.getUserContext(), outputCollector);
}
openOrPrepareWasCalled.set(true);
LOG.info("Opened spout {}:{}", componentId, idToTask.keySet());
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/generated/TopologyPageInfo.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/generated/TopologyPageInfo.java b/storm-client/src/jvm/org/apache/storm/generated/TopologyPageInfo.java
index a0e25ce..25bda95 100644
--- a/storm-client/src/jvm/org/apache/storm/generated/TopologyPageInfo.java
+++ b/storm-client/src/jvm/org/apache/storm/generated/TopologyPageInfo.java
@@ -62,7 +62,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
private static final org.apache.thrift.protocol.TField NUM_TASKS_FIELD_DESC = new org.apache.thrift.protocol.TField("num_tasks", org.apache.thrift.protocol.TType.I32, (short)5);
private static final org.apache.thrift.protocol.TField NUM_WORKERS_FIELD_DESC = new org.apache.thrift.protocol.TField("num_workers", org.apache.thrift.protocol.TType.I32, (short)6);
private static final org.apache.thrift.protocol.TField NUM_EXECUTORS_FIELD_DESC = new org.apache.thrift.protocol.TField("num_executors", org.apache.thrift.protocol.TType.I32, (short)7);
- private static final org.apache.thrift.protocol.TField TOPOLOGY_CONF_FIELD_DESC = new org.apache.thrift.protocol.TField("topology_conf", org.apache.thrift.protocol.TType.STRING, (short)8);
+ private static final org.apache.thrift.protocol.TField TOPOLOGY_CONF_FIELD_DESC = new org.apache.thrift.protocol.TField("topoConf", org.apache.thrift.protocol.TType.STRING, (short)8);
private static final org.apache.thrift.protocol.TField ID_TO_SPOUT_AGG_STATS_FIELD_DESC = new org.apache.thrift.protocol.TField("id_to_spout_agg_stats", org.apache.thrift.protocol.TType.MAP, (short)9);
private static final org.apache.thrift.protocol.TField ID_TO_BOLT_AGG_STATS_FIELD_DESC = new org.apache.thrift.protocol.TField("id_to_bolt_agg_stats", org.apache.thrift.protocol.TType.MAP, (short)10);
private static final org.apache.thrift.protocol.TField SCHED_STATUS_FIELD_DESC = new org.apache.thrift.protocol.TField("sched_status", org.apache.thrift.protocol.TType.STRING, (short)11);
@@ -91,7 +91,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
private int num_tasks; // optional
private int num_workers; // optional
private int num_executors; // optional
- private String topology_conf; // optional
+ private String topoConf; // optional
private Map<String,ComponentAggregateStats> id_to_spout_agg_stats; // optional
private Map<String,ComponentAggregateStats> id_to_bolt_agg_stats; // optional
private String sched_status; // optional
@@ -116,7 +116,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
NUM_TASKS((short)5, "num_tasks"),
NUM_WORKERS((short)6, "num_workers"),
NUM_EXECUTORS((short)7, "num_executors"),
- TOPOLOGY_CONF((short)8, "topology_conf"),
+ TOPOLOGY_CONF((short)8, "topoConf"),
ID_TO_SPOUT_AGG_STATS((short)9, "id_to_spout_agg_stats"),
ID_TO_BOLT_AGG_STATS((short)10, "id_to_bolt_agg_stats"),
SCHED_STATUS((short)11, "sched_status"),
@@ -259,7 +259,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
tmpMap.put(_Fields.NUM_EXECUTORS, new org.apache.thrift.meta_data.FieldMetaData("num_executors", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
- tmpMap.put(_Fields.TOPOLOGY_CONF, new org.apache.thrift.meta_data.FieldMetaData("topology_conf", org.apache.thrift.TFieldRequirementType.OPTIONAL,
+ tmpMap.put(_Fields.TOPOLOGY_CONF, new org.apache.thrift.meta_data.FieldMetaData("topoConf", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
tmpMap.put(_Fields.ID_TO_SPOUT_AGG_STATS, new org.apache.thrift.meta_data.FieldMetaData("id_to_spout_agg_stats", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.MapMetaData(org.apache.thrift.protocol.TType.MAP,
@@ -326,8 +326,8 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
this.num_tasks = other.num_tasks;
this.num_workers = other.num_workers;
this.num_executors = other.num_executors;
- if (other.is_set_topology_conf()) {
- this.topology_conf = other.topology_conf;
+ if (other.is_set_topoConf()) {
+ this.topoConf = other.topoConf;
}
if (other.is_set_id_to_spout_agg_stats()) {
Map<String,ComponentAggregateStats> __this__id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(other.id_to_spout_agg_stats.size());
@@ -404,7 +404,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
this.num_workers = 0;
set_num_executors_isSet(false);
this.num_executors = 0;
- this.topology_conf = null;
+ this.topoConf = null;
this.id_to_spout_agg_stats = null;
this.id_to_bolt_agg_stats = null;
this.sched_status = null;
@@ -585,26 +585,26 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
__isset_bitfield = EncodingUtils.setBit(__isset_bitfield, __NUM_EXECUTORS_ISSET_ID, value);
}
- public String get_topology_conf() {
- return this.topology_conf;
+ public String get_topoConf() {
+ return this.topoConf;
}
- public void set_topology_conf(String topology_conf) {
- this.topology_conf = topology_conf;
+ public void set_topoConf(String topoConf) {
+ this.topoConf = topoConf;
}
- public void unset_topology_conf() {
- this.topology_conf = null;
+ public void unset_topoConf() {
+ this.topoConf = null;
}
- /** Returns true if field topology_conf is set (has been assigned a value) and false otherwise */
- public boolean is_set_topology_conf() {
- return this.topology_conf != null;
+ /** Returns true if field topoConf is set (has been assigned a value) and false otherwise */
+ public boolean is_set_topoConf() {
+ return this.topoConf != null;
}
- public void set_topology_conf_isSet(boolean value) {
+ public void set_topoConf_isSet(boolean value) {
if (!value) {
- this.topology_conf = null;
+ this.topoConf = null;
}
}
@@ -1020,9 +1020,9 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
case TOPOLOGY_CONF:
if (value == null) {
- unset_topology_conf();
+ unset_topoConf();
} else {
- set_topology_conf((String)value);
+ set_topoConf((String)value);
}
break;
@@ -1165,7 +1165,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
return get_num_executors();
case TOPOLOGY_CONF:
- return get_topology_conf();
+ return get_topoConf();
case ID_TO_SPOUT_AGG_STATS:
return get_id_to_spout_agg_stats();
@@ -1235,7 +1235,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
case NUM_EXECUTORS:
return is_set_num_executors();
case TOPOLOGY_CONF:
- return is_set_topology_conf();
+ return is_set_topoConf();
case ID_TO_SPOUT_AGG_STATS:
return is_set_id_to_spout_agg_stats();
case ID_TO_BOLT_AGG_STATS:
@@ -1344,12 +1344,12 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
return false;
}
- boolean this_present_topology_conf = true && this.is_set_topology_conf();
- boolean that_present_topology_conf = true && that.is_set_topology_conf();
- if (this_present_topology_conf || that_present_topology_conf) {
- if (!(this_present_topology_conf && that_present_topology_conf))
+ boolean this_present_topoConf = true && this.is_set_topoConf();
+ boolean that_present_topoConf = true && that.is_set_topoConf();
+ if (this_present_topoConf || that_present_topoConf) {
+ if (!(this_present_topoConf && that_present_topoConf))
return false;
- if (!this.topology_conf.equals(that.topology_conf))
+ if (!this.topoConf.equals(that.topoConf))
return false;
}
@@ -1521,10 +1521,10 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
if (present_num_executors)
list.add(num_executors);
- boolean present_topology_conf = true && (is_set_topology_conf());
- list.add(present_topology_conf);
- if (present_topology_conf)
- list.add(topology_conf);
+ boolean present_topoConf = true && (is_set_topoConf());
+ list.add(present_topoConf);
+ if (present_topoConf)
+ list.add(topoConf);
boolean present_id_to_spout_agg_stats = true && (is_set_id_to_spout_agg_stats());
list.add(present_id_to_spout_agg_stats);
@@ -1677,12 +1677,12 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
return lastComparison;
}
}
- lastComparison = Boolean.valueOf(is_set_topology_conf()).compareTo(other.is_set_topology_conf());
+ lastComparison = Boolean.valueOf(is_set_topoConf()).compareTo(other.is_set_topoConf());
if (lastComparison != 0) {
return lastComparison;
}
- if (is_set_topology_conf()) {
- lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.topology_conf, other.topology_conf);
+ if (is_set_topoConf()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.topoConf, other.topoConf);
if (lastComparison != 0) {
return lastComparison;
}
@@ -1898,13 +1898,13 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
sb.append(this.num_executors);
first = false;
}
- if (is_set_topology_conf()) {
+ if (is_set_topoConf()) {
if (!first) sb.append(", ");
- sb.append("topology_conf:");
- if (this.topology_conf == null) {
+ sb.append("topoConf:");
+ if (this.topoConf == null) {
sb.append("null");
} else {
- sb.append(this.topology_conf);
+ sb.append(this.topoConf);
}
first = false;
}
@@ -2133,8 +2133,8 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
break;
case 8: // TOPOLOGY_CONF
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
- struct.topology_conf = iprot.readString();
- struct.set_topology_conf_isSet(true);
+ struct.topoConf = iprot.readString();
+ struct.set_topoConf_isSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
@@ -2342,10 +2342,10 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
oprot.writeI32(struct.num_executors);
oprot.writeFieldEnd();
}
- if (struct.topology_conf != null) {
- if (struct.is_set_topology_conf()) {
+ if (struct.topoConf != null) {
+ if (struct.is_set_topoConf()) {
oprot.writeFieldBegin(TOPOLOGY_CONF_FIELD_DESC);
- oprot.writeString(struct.topology_conf);
+ oprot.writeString(struct.topoConf);
oprot.writeFieldEnd();
}
}
@@ -2493,7 +2493,7 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
if (struct.is_set_num_executors()) {
optionals.set(5);
}
- if (struct.is_set_topology_conf()) {
+ if (struct.is_set_topoConf()) {
optionals.set(6);
}
if (struct.is_set_id_to_spout_agg_stats()) {
@@ -2557,8 +2557,8 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
if (struct.is_set_num_executors()) {
oprot.writeI32(struct.num_executors);
}
- if (struct.is_set_topology_conf()) {
- oprot.writeString(struct.topology_conf);
+ if (struct.is_set_topoConf()) {
+ oprot.writeString(struct.topoConf);
}
if (struct.is_set_id_to_spout_agg_stats()) {
{
@@ -2655,8 +2655,8 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
struct.set_num_executors_isSet(true);
}
if (incoming.get(6)) {
- struct.topology_conf = iprot.readString();
- struct.set_topology_conf_isSet(true);
+ struct.topoConf = iprot.readString();
+ struct.set_topoConf_isSet(true);
}
if (incoming.get(7)) {
{
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/hooks/BaseTaskHook.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/hooks/BaseTaskHook.java b/storm-client/src/jvm/org/apache/storm/hooks/BaseTaskHook.java
index 3eee8d8..e09602b 100644
--- a/storm-client/src/jvm/org/apache/storm/hooks/BaseTaskHook.java
+++ b/storm-client/src/jvm/org/apache/storm/hooks/BaseTaskHook.java
@@ -28,7 +28,7 @@ import java.util.Map;
public class BaseTaskHook implements ITaskHook {
@Override
- public void prepare(Map conf, TopologyContext context) {
+ public void prepare(Map<String, Object> conf, TopologyContext context) {
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/hooks/BaseWorkerHook.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/hooks/BaseWorkerHook.java b/storm-client/src/jvm/org/apache/storm/hooks/BaseWorkerHook.java
index e178280..526a5df 100644
--- a/storm-client/src/jvm/org/apache/storm/hooks/BaseWorkerHook.java
+++ b/storm-client/src/jvm/org/apache/storm/hooks/BaseWorkerHook.java
@@ -33,11 +33,11 @@ public class BaseWorkerHook implements IWorkerHook, Serializable {
/**
* This method is called when a worker is started
*
- * @param stormConf The Storm configuration for this worker
+ * @param topoConf The Storm configuration for this worker
* @param context This object can be used to get information about this worker's place within the topology
*/
@Override
- public void start(Map stormConf, WorkerTopologyContext context) {
+ public void start(Map<String, Object> topoConf, WorkerTopologyContext context) {
// NOOP
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/hooks/ITaskHook.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/hooks/ITaskHook.java b/storm-client/src/jvm/org/apache/storm/hooks/ITaskHook.java
index 519af0c..ae65f46 100644
--- a/storm-client/src/jvm/org/apache/storm/hooks/ITaskHook.java
+++ b/storm-client/src/jvm/org/apache/storm/hooks/ITaskHook.java
@@ -27,7 +27,7 @@ import org.apache.storm.task.TopologyContext;
import java.util.Map;
public interface ITaskHook {
- void prepare(Map conf, TopologyContext context);
+ void prepare(Map<String, Object> conf, TopologyContext context);
void cleanup();
void emit(EmitInfo info);
void spoutAck(SpoutAckInfo info);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java b/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java
index 5f49fe3..6318e24 100644
--- a/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java
+++ b/storm-client/src/jvm/org/apache/storm/hooks/IWorkerHook.java
@@ -32,10 +32,10 @@ public interface IWorkerHook extends Serializable {
/**
* This method is called when a worker is started
*
- * @param stormConf The Storm configuration for this worker
+ * @param topoConf The Storm configuration for this worker
* @param context This object can be used to get information about this worker's place within the topology
*/
- void start(Map stormConf, WorkerTopologyContext context);
+ void start(Map<String, Object> topoConf, WorkerTopologyContext context);
/**
* This method is called right before a worker shuts down
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/lambda/LambdaSpout.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/lambda/LambdaSpout.java b/storm-client/src/jvm/org/apache/storm/lambda/LambdaSpout.java
index 6d0ba3a..35abc41 100644
--- a/storm-client/src/jvm/org/apache/storm/lambda/LambdaSpout.java
+++ b/storm-client/src/jvm/org/apache/storm/lambda/LambdaSpout.java
@@ -35,7 +35,7 @@ public class LambdaSpout extends BaseRichSpout {
}
@Override
- public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
+ public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
index 8f415c2..75e58d9 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
@@ -41,7 +41,7 @@ public class DeserializingConnectionCallback implements IConnectionCallback {
}
};
- public DeserializingConnectionCallback(final Map conf, final GeneralTopologyContext context, WorkerState.ILocalTransferCallback callback) {
+ public DeserializingConnectionCallback(final Map<String, Object> conf, final GeneralTopologyContext context, WorkerState.ILocalTransferCallback callback) {
_conf = conf;
_context = context;
_cb = callback;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/IContext.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/IContext.java b/storm-client/src/jvm/org/apache/storm/messaging/IContext.java
index 72812b1..97148ce 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/IContext.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/IContext.java
@@ -25,15 +25,15 @@ import java.util.Map;
* Messaging plugin is specified via Storm config parameter, storm.messaging.transport.
*
* A messaging plugin should have a default constructor and implements IContext interface.
- * Upon construction, we will invoke IContext::prepare(storm_conf) to enable context to be configured
+ * Upon construction, we will invoke IContext::prepare(topoConf) to enable context to be configured
* according to storm configuration.
*/
public interface IContext {
/**
* This method is invoked at the startup of messaging plugin
- * @param storm_conf storm configuration
+ * @param topoConf storm configuration
*/
- public void prepare(Map storm_conf);
+ public void prepare(Map<String, Object> topoConf);
/**
* This method is invoked when a worker is unload a messaging plugin
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/TransportFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/TransportFactory.java b/storm-client/src/jvm/org/apache/storm/messaging/TransportFactory.java
index 511257d..e98ea64 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/TransportFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/TransportFactory.java
@@ -26,10 +26,10 @@ import org.apache.storm.Config;
public class TransportFactory {
public static final Logger LOG = LoggerFactory.getLogger(TransportFactory.class);
- public static IContext makeContext(Map storm_conf) {
+ public static IContext makeContext(Map<String, Object> topoConf) {
//get factory class name
- String transport_plugin_klassName = (String)storm_conf.get(Config.STORM_MESSAGING_TRANSPORT);
+ String transport_plugin_klassName = (String)topoConf.get(Config.STORM_MESSAGING_TRANSPORT);
LOG.info("Storm peer transport plugin:"+transport_plugin_klassName);
IContext transport;
@@ -42,12 +42,12 @@ public class TransportFactory {
//case 1: plugin is a IContext class
transport = (IContext)obj;
//initialize with storm configuration
- transport.prepare(storm_conf);
+ transport.prepare(topoConf);
} else {
- //case 2: Non-IContext plugin must have a makeContext(storm_conf) method that returns IContext object
+ //case 2: Non-IContext plugin must have a makeContext(topoConf) method that returns IContext object
Method method = klass.getMethod("makeContext", Map.class);
LOG.debug("object:"+obj+" method:"+method);
- transport = (IContext) method.invoke(obj, storm_conf);
+ transport = (IContext) method.invoke(obj, topoConf);
}
} catch(Exception e) {
throw new RuntimeException("Fail to construct messaging plugin from plugin "+transport_plugin_klassName, e);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/local/Context.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/local/Context.java b/storm-client/src/jvm/org/apache/storm/messaging/local/Context.java
index 7300847..06318b8 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/local/Context.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/local/Context.java
@@ -196,7 +196,7 @@ public class Context implements IContext {
@SuppressWarnings("rawtypes")
@Override
- public void prepare(Map storm_conf) {
+ public void prepare(Map<String, Object> topoConf) {
//NOOP
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java
index 8463af6..52fdda8 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java
@@ -77,7 +77,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private static final long NO_DELAY_MS = 0L;
private static final Timer timer = new Timer("Netty-ChannelAlive-Timer", true);
- private final Map stormConf;
+ private final Map<String, Object> topoConf;
private final StormBoundedExponentialBackoffRetry retryPolicy;
private final ClientBootstrap bootstrap;
private final InetSocketAddress dstAddress;
@@ -139,24 +139,24 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
private final Object writeLock = new Object();
@SuppressWarnings("rawtypes")
- Client(Map stormConf, ChannelFactory factory, HashedWheelTimer scheduler, String host, int port, Context context) {
- this.stormConf = stormConf;
+ Client(Map<String, Object> topoConf, ChannelFactory factory, HashedWheelTimer scheduler, String host, int port, Context context) {
+ this.topoConf = topoConf;
closing = false;
this.scheduler = scheduler;
this.context = context;
- int bufferSize = ObjectReader.getInt(stormConf.get(Config.STORM_MESSAGING_NETTY_BUFFER_SIZE));
+ int bufferSize = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_BUFFER_SIZE));
// if SASL authentication is disabled, saslChannelReady is initialized as true; otherwise false
- saslChannelReady.set(!ObjectReader.getBoolean(stormConf.get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION), false));
+ saslChannelReady.set(!ObjectReader.getBoolean(topoConf.get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION), false));
LOG.info("creating Netty Client, connecting to {}:{}, bufferSize: {}", host, port, bufferSize);
- int messageBatchSize = ObjectReader.getInt(stormConf.get(Config.STORM_NETTY_MESSAGE_BATCH_SIZE), 262144);
+ int messageBatchSize = ObjectReader.getInt(topoConf.get(Config.STORM_NETTY_MESSAGE_BATCH_SIZE), 262144);
- int maxReconnectionAttempts = ObjectReader.getInt(stormConf.get(Config.STORM_MESSAGING_NETTY_MAX_RETRIES));
- int minWaitMs = ObjectReader.getInt(stormConf.get(Config.STORM_MESSAGING_NETTY_MIN_SLEEP_MS));
- int maxWaitMs = ObjectReader.getInt(stormConf.get(Config.STORM_MESSAGING_NETTY_MAX_SLEEP_MS));
+ int maxReconnectionAttempts = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_MAX_RETRIES));
+ int minWaitMs = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_MIN_SLEEP_MS));
+ int maxWaitMs = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_MAX_SLEEP_MS));
retryPolicy = new StormBoundedExponentialBackoffRetry(minWaitMs, maxWaitMs, maxReconnectionAttempts);
// Initiate connection to remote destination
- bootstrap = createClientBootstrap(factory, bufferSize, stormConf);
+ bootstrap = createClientBootstrap(factory, bufferSize, topoConf);
dstAddress = new InetSocketAddress(host, port);
dstAddressPrefixedName = prefixedName(dstAddress);
launchChannelAliveThread();
@@ -189,12 +189,12 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
}, 0, CHANNEL_ALIVE_INTERVAL_MS);
}
- private ClientBootstrap createClientBootstrap(ChannelFactory factory, int bufferSize, Map stormConf) {
+ private ClientBootstrap createClientBootstrap(ChannelFactory factory, int bufferSize, Map<String, Object> topoConf) {
ClientBootstrap bootstrap = new ClientBootstrap(factory);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("sendBufferSize", bufferSize);
bootstrap.setOption("keepAlive", true);
- bootstrap.setPipelineFactory(new StormClientPipelineFactory(this, stormConf));
+ bootstrap.setPipelineFactory(new StormClientPipelineFactory(this, topoConf));
return bootstrap;
}
@@ -495,7 +495,7 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
}
public Map getConfig() {
- return stormConf;
+ return topoConf;
}
/** ISaslClient interface **/
@@ -508,11 +508,11 @@ public class Client extends ConnectionWithStatus implements IStatefulObject, ISa
}
public String name() {
- return (String)stormConf.get(Config.TOPOLOGY_NAME);
+ return (String)topoConf.get(Config.TOPOLOGY_NAME);
}
public String secretKey() {
- return SaslUtils.getSecretKey(stormConf);
+ return SaslUtils.getSecretKey(topoConf);
}
/** end **/
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/Context.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Context.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Context.java
index 0f7c66b..10ffc83 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Context.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/Context.java
@@ -31,7 +31,7 @@ import org.apache.storm.messaging.IContext;
public class Context implements IContext {
@SuppressWarnings("rawtypes")
- private Map storm_conf;
+ private Map<String, Object> topoConf;
private Map<String, IConnection> connections;
private NioClientSocketChannelFactory clientChannelFactory;
@@ -41,12 +41,12 @@ public class Context implements IContext {
* initialization per Storm configuration
*/
@SuppressWarnings("rawtypes")
- public void prepare(Map storm_conf) {
- this.storm_conf = storm_conf;
+ public void prepare(Map<String, Object> topoConf) {
+ this.topoConf = topoConf;
connections = new HashMap<>();
//each context will have a single client channel factory
- int maxWorkers = ObjectReader.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS));
+ int maxWorkers = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS));
ThreadFactory bossFactory = new NettyRenameThreadFactory("client" + "-boss");
ThreadFactory workerFactory = new NettyRenameThreadFactory("client" + "-worker");
if (maxWorkers > 0) {
@@ -64,7 +64,7 @@ public class Context implements IContext {
* establish a server with a binding port
*/
public synchronized IConnection bind(String storm_id, int port) {
- IConnection server = new Server(storm_conf, port);
+ IConnection server = new Server(topoConf, port);
connections.put(key(storm_id, port), server);
return server;
}
@@ -78,7 +78,7 @@ public class Context implements IContext {
{
return connection;
}
- IConnection client = new Client(storm_conf, clientChannelFactory,
+ IConnection client = new Client(topoConf, clientChannelFactory,
clientScheduleService, host, port, this);
connections.put(key(host, port), client);
return client;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslClientHandler.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslClientHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslClientHandler.java
index d7b20d6..abaadfb 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslClientHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslClientHandler.java
@@ -35,13 +35,13 @@ public class KerberosSaslClientHandler extends SimpleChannelUpstreamHandler {
private ISaslClient client;
long start_time;
/** Used for client or server's token to send or receive from each other. */
- private Map storm_conf;
+ private Map<String, Object> topoConf;
private String jaas_section;
private String host;
- public KerberosSaslClientHandler(ISaslClient client, Map storm_conf, String jaas_section, String host) throws IOException {
+ public KerberosSaslClientHandler(ISaslClient client, Map<String, Object> topoConf, String jaas_section, String host) throws IOException {
this.client = client;
- this.storm_conf = storm_conf;
+ this.topoConf = topoConf;
this.jaas_section = jaas_section;
this.host = host;
start_time = System.currentTimeMillis();
@@ -64,7 +64,7 @@ public class KerberosSaslClientHandler extends SimpleChannelUpstreamHandler {
if (saslNettyClient == null) {
LOG.debug("Creating saslNettyClient now for channel: {}",
channel);
- saslNettyClient = new KerberosSaslNettyClient(storm_conf, jaas_section, host);
+ saslNettyClient = new KerberosSaslNettyClient(topoConf, jaas_section, host);
KerberosSaslNettyClientState.getKerberosSaslNettyClient.set(channel,
saslNettyClient);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
index 2efcb19..0ebb9f8 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
@@ -58,7 +58,7 @@ public class KerberosSaslNettyClient {
/**
* Create a KerberosSaslNettyClient for authentication with servers.
*/
- public KerberosSaslNettyClient(Map storm_conf, String jaas_section, String host) {
+ public KerberosSaslNettyClient(Map<String, Object> topoConf, String jaas_section, String host) {
LOG.debug("KerberosSaslNettyClient: Creating SASL {} client to authenticate to server ",
SaslUtils.KERBEROS);
@@ -66,7 +66,7 @@ public class KerberosSaslNettyClient {
Configuration login_conf;
try {
- login_conf = AuthUtils.GetConfiguration(storm_conf);
+ login_conf = AuthUtils.GetConfiguration(topoConf);
}
catch (Throwable t) {
LOG.error("Failed to get login_conf: ", t);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
index 72486ef..a36ac47 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
@@ -53,12 +53,12 @@ class KerberosSaslNettyServer {
private Subject subject;
private List<String> authorizedUsers;
- KerberosSaslNettyServer(Map storm_conf, String jaas_section, List<String> authorizedUsers) {
+ KerberosSaslNettyServer(Map<String, Object> topoConf, String jaas_section, List<String> authorizedUsers) {
this.authorizedUsers = authorizedUsers;
LOG.debug("Getting Configuration.");
Configuration login_conf;
try {
- login_conf = AuthUtils.GetConfiguration(storm_conf);
+ login_conf = AuthUtils.GetConfiguration(topoConf);
}
catch (Throwable t) {
LOG.error("Failed to get login_conf: ", t);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslServerHandler.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslServerHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslServerHandler.java
index 14ac172..d57b8ab 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslServerHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslServerHandler.java
@@ -33,16 +33,16 @@ public class KerberosSaslServerHandler extends SimpleChannelUpstreamHandler {
ISaslServer server;
/** Used for client or server's token to send or receive from each other. */
- private Map storm_conf;
+ private Map<String, Object> topoConf;
private String jaas_section;
private List<String> authorizedUsers;
private static final Logger LOG = LoggerFactory
.getLogger(KerberosSaslServerHandler.class);
- public KerberosSaslServerHandler(ISaslServer server, Map storm_conf, String jaas_section, List<String> authorizedUsers) throws IOException {
+ public KerberosSaslServerHandler(ISaslServer server, Map<String, Object> topoConf, String jaas_section, List<String> authorizedUsers) throws IOException {
this.server = server;
- this.storm_conf = storm_conf;
+ this.topoConf = topoConf;
this.jaas_section = jaas_section;
this.authorizedUsers = authorizedUsers;
}
@@ -70,7 +70,7 @@ public class KerberosSaslServerHandler extends SimpleChannelUpstreamHandler {
if (saslNettyServer == null) {
LOG.debug("No saslNettyServer for {} yet; creating now, with topology token: ", channel);
try {
- saslNettyServer = new KerberosSaslNettyServer(storm_conf, jaas_section, authorizedUsers);
+ saslNettyServer = new KerberosSaslNettyServer(topoConf, jaas_section, authorizedUsers);
KerberosSaslNettyServerState.getKerberosSaslNettyServer.set(channel,
saslNettyServer);
} catch (RuntimeException ioe) {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslUtils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslUtils.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslUtils.java
index b5c6b6b..9308100 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslUtils.java
@@ -61,7 +61,7 @@ class SaslUtils {
return new String(Base64.encodeBase64(identifier), Charsets.UTF_8);
}
- static String getSecretKey(Map conf) {
+ static String getSecretKey(Map<String, Object> conf) {
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/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java
index fce00e8..376006f 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java
@@ -51,7 +51,7 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe
private static final Logger LOG = LoggerFactory.getLogger(Server.class);
@SuppressWarnings("rawtypes")
- Map storm_conf;
+ Map<String, Object> topoConf;
int port;
private final ConcurrentHashMap<String, AtomicInteger> messagesEnqueued = new ConcurrentHashMap<>();
private final AtomicInteger messagesDequeued = new AtomicInteger(0);
@@ -66,15 +66,15 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe
private IConnectionCallback _cb = null;
@SuppressWarnings("rawtypes")
- Server(Map storm_conf, int port) {
- this.storm_conf = storm_conf;
+ Server(Map<String, Object> topoConf, int port) {
+ this.topoConf = topoConf;
this.port = port;
- _ser = new KryoValuesSerializer(storm_conf);
+ _ser = new KryoValuesSerializer(topoConf);
// Configure the server.
- int buffer_size = ObjectReader.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_BUFFER_SIZE));
- int backlog = ObjectReader.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_SOCKET_BACKLOG), 500);
- int maxWorkers = ObjectReader.getInt(storm_conf.get(Config.STORM_MESSAGING_NETTY_SERVER_WORKER_THREADS));
+ int buffer_size = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_BUFFER_SIZE));
+ int backlog = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_SOCKET_BACKLOG), 500);
+ int maxWorkers = ObjectReader.getInt(topoConf.get(Config.STORM_MESSAGING_NETTY_SERVER_WORKER_THREADS));
ThreadFactory bossFactory = new NettyRenameThreadFactory(netty_name() + "-boss");
ThreadFactory workerFactory = new NettyRenameThreadFactory(netty_name() + "-worker");
@@ -256,11 +256,11 @@ class Server extends ConnectionWithStatus implements IStatefulObject, ISaslServe
}
public String name() {
- return (String)storm_conf.get(Config.TOPOLOGY_NAME);
+ return (String)topoConf.get(Config.TOPOLOGY_NAME);
}
public String secretKey() {
- return SaslUtils.getSecretKey(storm_conf);
+ return SaslUtils.getSecretKey(topoConf);
}
public void authenticated(Channel c) {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientHandler.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientHandler.java
index a28b076..3487003 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientHandler.java
@@ -41,7 +41,7 @@ public class StormClientHandler extends SimpleChannelUpstreamHandler {
private Client client;
private KryoValuesDeserializer _des;
- StormClientHandler(Client client, Map conf) {
+ StormClientHandler(Client client, Map<String, Object> conf) {
this.client = client;
_des = new KryoValuesDeserializer(conf);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java
index e36eb65..3fd7641 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java
@@ -26,9 +26,9 @@ import java.util.Map;
class StormClientPipelineFactory implements ChannelPipelineFactory {
private Client client;
- private Map conf;
+ private Map<String, Object> conf;
- StormClientPipelineFactory(Client client, Map conf) {
+ StormClientPipelineFactory(Client client, Map<String, Object> conf) {
this.client = client;
this.conf = conf;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
index d28790b..2e4f418 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
@@ -39,7 +39,7 @@ class StormServerPipelineFactory implements ChannelPipelineFactory {
// Encoder
pipeline.addLast("encoder", new MessageEncoder());
- boolean isNettyAuth = (Boolean) this.server.storm_conf
+ boolean isNettyAuth = (Boolean) this.server.topoConf
.get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
if (isNettyAuth) {
// Authenticate: Removed after authentication completes
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java b/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
index 2950e08..901375d 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/EventLoggerBolt.java
@@ -42,10 +42,10 @@ public class EventLoggerBolt implements IBolt {
private IEventLogger eventLogger;
@Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
LOG.info("EventLoggerBolt prepare called");
eventLogger = new FileBasedEventLogger();
- eventLogger.prepare(stormConf, context);
+ eventLogger.prepare(topoConf, context);
}
@Override
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
index f6822ff..8f5632a 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/FileBasedEventLogger.java
@@ -79,7 +79,7 @@ public class FileBasedEventLogger implements IEventLogger {
@Override
- public void prepare(Map stormConf, TopologyContext context) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context) {
String stormId = context.getStormId();
int port = context.getThisWorkerPort();
@@ -87,7 +87,7 @@ public class FileBasedEventLogger implements IEventLogger {
* Include the topology name & worker port in the file name so that
* multiple event loggers can log independently.
*/
- String workersArtifactRoot = ConfigUtils.workerArtifactsRoot(stormConf, stormId, port);
+ String workersArtifactRoot = ConfigUtils.workerArtifactsRoot(topoConf, stormId, port);
Path path = Paths.get(workersArtifactRoot, "events.log");
File dir = path.toFile().getParentFile();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java b/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
index ce81602..70c5a8d 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/IEventLogger.java
@@ -56,7 +56,7 @@ public interface IEventLogger {
}
}
- void prepare(Map stormConf, TopologyContext context);
+ void prepare(Map<String, Object> topoConf, TopologyContext context);
/**
* This method would be invoked when the {@link EventLoggerBolt} receives a tuple from the spouts or bolts that has
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java b/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java
index 4957501..934bc92 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/LoggingMetricsConsumer.java
@@ -49,7 +49,7 @@ public class LoggingMetricsConsumer implements IMetricsConsumer {
public static final Logger LOG = LoggerFactory.getLogger(LoggingMetricsConsumer.class);
@Override
- public void prepare(Map stormConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter) { }
+ public void prepare(Map<String, Object> topoConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter) { }
static private String padding = " ";
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/MetricsConsumerBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/MetricsConsumerBolt.java b/storm-client/src/jvm/org/apache/storm/metric/MetricsConsumerBolt.java
index 8feed00..8fdb69e 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/MetricsConsumerBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/MetricsConsumerBolt.java
@@ -68,14 +68,14 @@ public class MetricsConsumerBolt implements IBolt {
}
@Override
- public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
+ public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
try {
_metricsConsumer = (IMetricsConsumer)Class.forName(_consumerClassName).newInstance();
} catch (Exception e) {
throw new RuntimeException("Could not instantiate a class listed in config under section " +
Config.TOPOLOGY_METRICS_CONSUMER_REGISTER + " with fully qualified name " + _consumerClassName, e);
}
- _metricsConsumer.prepare(stormConf, _registrationArgument, context, collector);
+ _metricsConsumer.prepare(topoConf, _registrationArgument, context, collector);
_collector = collector;
_taskExecuteThread = new Thread(new MetricsHandlerRunnable());
_taskExecuteThread.setDaemon(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/SystemBolt.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/SystemBolt.java b/storm-client/src/jvm/org/apache/storm/metric/SystemBolt.java
index f1426f7..67be2c9 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/SystemBolt.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/SystemBolt.java
@@ -91,7 +91,7 @@ public class SystemBolt implements IBolt {
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
- public void prepare(final Map topoConf, TopologyContext context, OutputCollector collector) {
+ public void prepare(final Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
if(_prepareWasCalled && !"local".equals(topoConf.get(Config.STORM_CLUSTER_MODE))) {
throw new RuntimeException("A single worker should have 1 SystemBolt instance.");
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java b/storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java
index f2623be..def04a3 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/api/IMetricsConsumer.java
@@ -80,7 +80,7 @@ public interface IMetricsConsumer {
}
}
- void prepare(Map stormConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter);
+ void prepare(Map<String, Object> topoConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter);
void handleDataPoints(TaskInfo taskInfo, Collection<DataPoint> dataPoints);
void cleanup();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/metric/filter/FilterByMetricName.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/metric/filter/FilterByMetricName.java b/storm-client/src/jvm/org/apache/storm/metric/filter/FilterByMetricName.java
index 9c95428..05fe861 100644
--- a/storm-client/src/jvm/org/apache/storm/metric/filter/FilterByMetricName.java
+++ b/storm-client/src/jvm/org/apache/storm/metric/filter/FilterByMetricName.java
@@ -107,4 +107,4 @@ public class FilterByMetricName implements MetricsFilter {
return !valueWhenMatched;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/multilang/ISerializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/multilang/ISerializer.java b/storm-client/src/jvm/org/apache/storm/multilang/ISerializer.java
index e3e6651..e4dce96 100644
--- a/storm-client/src/jvm/org/apache/storm/multilang/ISerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/multilang/ISerializer.java
@@ -49,7 +49,7 @@ public interface ISerializer extends Serializable {
* @param context topology context
* @return process pid
*/
- Number connect(Map conf, TopologyContext context) throws IOException,
+ Number connect(Map<String, Object> conf, TopologyContext context) throws IOException,
NoOutputException;
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/multilang/JsonSerializer.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/multilang/JsonSerializer.java b/storm-client/src/jvm/org/apache/storm/multilang/JsonSerializer.java
index 8800573..0353d83 100644
--- a/storm-client/src/jvm/org/apache/storm/multilang/JsonSerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/multilang/JsonSerializer.java
@@ -57,7 +57,7 @@ public class JsonSerializer implements ISerializer {
}
}
- public Number connect(Map conf, TopologyContext context)
+ public Number connect(Map<String, Object> conf, TopologyContext context)
throws IOException, NoOutputException {
JSONObject setupInfo = new JSONObject();
setupInfo.put("pidDir", context.getPIDDir());
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
index cbbfe7f..be09b3e 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -65,7 +65,7 @@ public class PacemakerClient implements ISaslClient {
public PacemakerClient() {
bootstrap = new ClientBootstrap();
}
- public PacemakerClient(Map config, String host) {
+ public PacemakerClient(Map<String, Object> config, String host) {
int port = (int)config.get(Config.PACEMAKER_PORT);
client_name = (String)config.get(Config.TOPOLOGY_NAME);
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
index 1ec0ee4..0069b59 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
@@ -34,9 +34,9 @@ public class PacemakerClientPool {
private ConcurrentHashMap<String, PacemakerClient> clientForServer = new ConcurrentHashMap<>();
private ConcurrentLinkedQueue<String> servers;
- private Map config;
+ private Map<String, Object> config;
- public PacemakerClientPool(Map config) {
+ public PacemakerClientPool(Map<String, Object> config) {
this.config = config;
List<String> serverList = (List<String>)config.get(Config.PACEMAKER_SERVERS);
if(serverList == null) {
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
index 79c6ea4..42a68fd 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
@@ -46,13 +46,13 @@ public class ThriftNettyClientCodec {
private PacemakerClient client;
private AuthMethod authMethod;
- private Map storm_conf;
+ private Map<String, Object> topoConf;
private String host;
- public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map storm_conf, AuthMethod authMethod, String host) {
+ public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf, AuthMethod authMethod, String host) {
client = pacemaker_client;
this.authMethod = authMethod;
- this.storm_conf = storm_conf;
+ this.topoConf = topoConf;
this.host = host;
}
@@ -68,7 +68,7 @@ public class ThriftNettyClientCodec {
LOG.debug("Adding KerberosSaslClientHandler to pacemaker client pipeline.");
pipeline.addLast(KERBEROS_HANDLER,
new KerberosSaslClientHandler(client,
- storm_conf,
+ topoConf,
AuthUtils.LOGIN_CONTEXT_PACEMAKER_CLIENT,
host));
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/Cluster.java b/storm-client/src/jvm/org/apache/storm/scheduler/Cluster.java
index b04a659..968c68a 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/Cluster.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/Cluster.java
@@ -73,12 +73,12 @@ public class Cluster {
*/
private Map<String, List<String>> hostToId;
- private Map conf = null;
+ private Map<String, Object> conf = null;
private Set<String> blackListedHosts = new HashSet<String>();
private INimbus inimbus;
- public Cluster(INimbus nimbus, Map<String, SupervisorDetails> supervisors, Map<String, SchedulerAssignmentImpl> assignments, Map storm_conf){
+ public Cluster(INimbus nimbus, Map<String, SupervisorDetails> supervisors, Map<String, SchedulerAssignmentImpl> assignments, Map<String, Object> topoConf){
this.inimbus = nimbus;
this.supervisors = new HashMap<String, SupervisorDetails>(supervisors.size());
this.supervisors.putAll(supervisors);
@@ -98,7 +98,7 @@ public class Cluster {
}
this.hostToId.get(host).add(nodeId);
}
- this.conf = storm_conf;
+ this.conf = topoConf;
}
/**
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/INimbus.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/INimbus.java b/storm-client/src/jvm/org/apache/storm/scheduler/INimbus.java
index 6b0fda1..5f64eaa 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/INimbus.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/INimbus.java
@@ -22,7 +22,7 @@ import java.util.Map;
import java.util.Set;
public interface INimbus {
- void prepare(Map stormConf, String schedulerLocalDir);
+ void prepare(Map<String, Object> topoConf, String schedulerLocalDir);
/**
* Returns all slots that are available for the next round of scheduling. A slot is available for scheduling
* if it is free and can be assigned to, or if it is used and can be reassigned.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/IScheduler.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/IScheduler.java b/storm-client/src/jvm/org/apache/storm/scheduler/IScheduler.java
index ef9268b..f97755b 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/IScheduler.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/IScheduler.java
@@ -22,7 +22,7 @@ import java.util.Map;
public interface IScheduler {
- void prepare(Map conf);
+ void prepare(Map<String, Object> conf);
/**
* Set assignments for the topologies which needs scheduling. The new assignments is available
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/ISupervisor.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/ISupervisor.java b/storm-client/src/jvm/org/apache/storm/scheduler/ISupervisor.java
index 2ca3aa6..8b44f9e 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/ISupervisor.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/ISupervisor.java
@@ -22,7 +22,7 @@ import java.util.Collection;
public interface ISupervisor {
- void prepare(Map stormConf, String schedulerLocalDir);
+ void prepare(Map<String, Object> topoConf, String schedulerLocalDir);
// for mesos, this is {hostname}-{topologyid}
/**
* The id used for writing metadata into ZK.
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java b/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
index a103cc6..9d63311 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/TopologyDetails.java
@@ -42,7 +42,7 @@ import org.slf4j.LoggerFactory;
public class TopologyDetails {
private String topologyId;
- private Map topologyConf;
+ private Map<String, Object> topologyConf;
private StormTopology topology;
private Map<ExecutorDetails, String> executorToComponent;
private int numWorkers;
@@ -57,16 +57,16 @@ public class TopologyDetails {
private static final Logger LOG = LoggerFactory.getLogger(TopologyDetails.class);
- public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology, int numWorkers) {
+ public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology, int numWorkers) {
this(topologyId, topologyConf, topology, numWorkers, null, 0);
}
- public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology,
+ public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology,
int numWorkers, Map<ExecutorDetails, String> executorToComponents) {
this(topologyId, topologyConf, topology, numWorkers, executorToComponents, 0);
}
- public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology,
+ public TopologyDetails(String topologyId, Map<String, Object> topologyConf, StormTopology topology,
int numWorkers, Map<ExecutorDetails, String> executorToComponents, int launchTime) {
this.topologyId = topologyId;
this.topologyConf = topologyConf;
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
index a36bd36..85c04f2 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/resource/ResourceUtils.java
@@ -42,7 +42,7 @@ import java.util.Map;
public class ResourceUtils {
private static final Logger LOG = LoggerFactory.getLogger(ResourceUtils.class);
- public static Map<String, Map<String, Double>> getBoltsResources(StormTopology topology, Map topologyConf) {
+ public static Map<String, Map<String, Double>> getBoltsResources(StormTopology topology, Map<String, Object> topologyConf) {
Map<String, Map<String, Double>> boltResources = new HashMap<String, Map<String, Double>>();
if (topology.get_bolts() != null) {
for (Map.Entry<String, Bolt> bolt : topology.get_bolts().entrySet()) {
@@ -54,7 +54,7 @@ public class ResourceUtils {
return boltResources;
}
- public static Map<String, Map<String, Double>> getSpoutsResources(StormTopology topology, Map topologyConf) {
+ public static Map<String, Map<String, Double>> getSpoutsResources(StormTopology topology, Map<String, Object> topologyConf) {
Map<String, Map<String, Double>> spoutResources = new HashMap<String, Map<String, Double>>();
if (topology.get_spouts() != null) {
for (Map.Entry<String, SpoutSpec> spout : topology.get_spouts().entrySet()) {
@@ -66,12 +66,12 @@ public class ResourceUtils {
return spoutResources;
}
- public static void checkIntialization(Map<String, Double> topology_resources, String Com, Map topologyConf) {
+ public static void checkIntialization(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
checkInitMem(topology_resources, Com, topologyConf);
checkInitCPU(topology_resources, Com, topologyConf);
}
- public static void checkInitMem(Map<String, Double> topology_resources, String Com, Map topologyConf) {
+ public static void checkInitMem(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
if (!topology_resources.containsKey(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB)) {
topology_resources.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB,
ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null));
@@ -84,7 +84,7 @@ public class ResourceUtils {
}
}
- public static void checkInitCPU(Map<String, Double> topology_resources, String Com, Map topologyConf) {
+ public static void checkInitCPU(Map<String, Double> topology_resources, String Com, Map<String, Object> topologyConf) {
if (!topology_resources.containsKey(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT)) {
topology_resources.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT,
ObjectReader.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null));
@@ -123,7 +123,7 @@ public class ResourceUtils {
return topology_resources;
}
- private static void debugMessage(String memoryType, String Com, Map topologyConf) {
+ private static void debugMessage(String memoryType, String Com, Map<String, Object> topologyConf) {
if (memoryType.equals("ONHEAP")) {
LOG.debug(
"Unable to extract resource requirement for Component {} \n Resource : Memory Type : On Heap set to default {}",
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java b/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
index 8a28ac0..80f5e9b 100644
--- a/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
+++ b/storm-client/src/jvm/org/apache/storm/scheduler/resource/SchedulingState.java
@@ -33,9 +33,9 @@ public class SchedulingState {
public final Cluster cluster;
public final Topologies topologies;
public final RAS_Nodes nodes;
- public final Map conf = new Config();
+ public final Map<String, Object> conf = new Config();
- public SchedulingState(Map<String, User> userMap, Cluster cluster, Topologies topologies, Map conf) {
+ public SchedulingState(Map<String, User> userMap, Cluster cluster, Topologies topologies, Map<String, Object> conf) {
for (Map.Entry<String, User> userMapEntry : userMap.entrySet()) {
String userId = userMapEntry.getKey();
User user = userMapEntry.getValue();
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/INimbusCredentialPlugin.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/INimbusCredentialPlugin.java b/storm-client/src/jvm/org/apache/storm/security/INimbusCredentialPlugin.java
index 4de0cfa..1378f0a 100644
--- a/storm-client/src/jvm/org/apache/storm/security/INimbusCredentialPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/INimbusCredentialPlugin.java
@@ -32,7 +32,7 @@ public interface INimbusCredentialPlugin extends Shutdownable {
* this method will be called when nimbus initializes.
* @param conf
*/
- void prepare(Map conf);
+ void prepare(Map<String, Object> conf);
/**
* Method that will be called on nimbus as part of submit topology. This plugin will be called
@@ -43,5 +43,5 @@ public interface INimbusCredentialPlugin extends Shutdownable {
* @param conf topology configuration
* @return
*/
- void populateCredentials(Map<String, String> credentials, Map conf);
+ void populateCredentials(Map<String, String> credentials, Map<String, Object> conf);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/AuthUtils.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/AuthUtils.java b/storm-client/src/jvm/org/apache/storm/security/auth/AuthUtils.java
index e282040..e87c701 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/AuthUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/AuthUtils.java
@@ -56,14 +56,14 @@ public class AuthUtils {
/**
* Construct a JAAS configuration object per storm configuration file
- * @param storm_conf Storm configuration
+ * @param topoConf Storm configuration
* @return JAAS configuration object
*/
- public static Configuration GetConfiguration(Map storm_conf) {
+ public static Configuration GetConfiguration(Map<String, Object> topoConf) {
Configuration login_conf = null;
//find login file configuration from Storm configuration
- String loginConfigurationFile = (String)storm_conf.get("java.security.auth.login.config");
+ String loginConfigurationFile = (String)topoConf.get("java.security.auth.login.config");
if ((loginConfigurationFile != null) && (loginConfigurationFile.length()>0)) {
File config_file = new File(loginConfigurationFile);
if (!config_file.canRead()) {
@@ -151,13 +151,13 @@ public class AuthUtils {
/**
* Construct a principal to local plugin
- * @param storm_conf storm configuration
+ * @param topoConf storm configuration
* @return the plugin
*/
- public static IPrincipalToLocal GetPrincipalToLocalPlugin(Map<String, Object> storm_conf) {
+ public static IPrincipalToLocal GetPrincipalToLocalPlugin(Map<String, Object> topoConf) {
IPrincipalToLocal ptol = null;
try {
- String ptol_klassName = (String) storm_conf.get(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN);
+ String ptol_klassName = (String) topoConf.get(Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN);
if (ptol_klassName == null) {
LOG.warn("No principal to local given {}", Config.STORM_PRINCIPAL_TO_LOCAL_PLUGIN);
} else {
@@ -165,7 +165,7 @@ public class AuthUtils {
//TODO this can only ever be null if someone is doing something odd with mocking
// We should really fix the mocking and remove this
if (ptol != null) {
- ptol.prepare(storm_conf);
+ ptol.prepare(topoConf);
}
}
} catch (Exception e) {
@@ -202,7 +202,7 @@ public class AuthUtils {
* @param conf the storm configuration to use.
* @return the configured credential renewers.
*/
- public static Collection<ICredentialsRenewer> GetCredentialRenewers(Map conf) {
+ public static Collection<ICredentialsRenewer> GetCredentialRenewers(Map<String, Object> conf) {
try {
Set<ICredentialsRenewer> ret = new HashSet<>();
Collection<String> clazzes = (Collection<String>)conf.get(Config.NIMBUS_CREDENTIAL_RENEWERS);
@@ -224,7 +224,7 @@ public class AuthUtils {
* @param conf nimbus configuration to use.
* @return nimbus auto credential plugins.
*/
- public static Collection<INimbusCredentialPlugin> getNimbusAutoCredPlugins(Map conf) {
+ public static Collection<INimbusCredentialPlugin> getNimbusAutoCredPlugins(Map<String, Object> conf) {
try {
Set<INimbusCredentialPlugin> ret = new HashSet<>();
Collection<String> clazzes = (Collection<String>)conf.get(Config.NIMBUS_AUTO_CRED_PLUGINS);
@@ -243,17 +243,17 @@ public class AuthUtils {
/**
* Get all of the configured AutoCredential Plugins.
- * @param storm_conf the storm configuration to use.
+ * @param topoConf the storm configuration to use.
* @return the configured auto credentials.
*/
- public static Collection<IAutoCredentials> GetAutoCredentials(Map storm_conf) {
+ public static Collection<IAutoCredentials> GetAutoCredentials(Map<String, Object> topoConf) {
try {
Set<IAutoCredentials> autos = new HashSet<>();
- Collection<String> clazzes = (Collection<String>)storm_conf.get(Config.TOPOLOGY_AUTO_CREDENTIALS);
+ Collection<String> clazzes = (Collection<String>)topoConf.get(Config.TOPOLOGY_AUTO_CREDENTIALS);
if (clazzes != null) {
for (String clazz : clazzes) {
IAutoCredentials a = ReflectionUtils.newInstance(clazz);
- a.prepare(storm_conf);
+ a.prepare(topoConf);
autos.add(a);
}
}
@@ -308,11 +308,11 @@ public class AuthUtils {
/**
* Construct a transport plugin per storm configuration
*/
- public static ITransportPlugin GetTransportPlugin(ThriftConnectionType type, Map storm_conf, Configuration login_conf) {
+ public static ITransportPlugin GetTransportPlugin(ThriftConnectionType type, Map<String, Object> topoConf, Configuration login_conf) {
try {
- String transport_plugin_klassName = type.getTransportPlugin(storm_conf);
+ String transport_plugin_klassName = type.getTransportPlugin(topoConf);
ITransportPlugin transportPlugin = ReflectionUtils.newInstance(transport_plugin_klassName);
- transportPlugin.prepare(type, storm_conf, login_conf);
+ transportPlugin.prepare(type, topoConf, login_conf);
return transportPlugin;
} catch (Exception e) {
throw new RuntimeException(e);
@@ -339,7 +339,7 @@ public class AuthUtils {
* @param conf storm configuration
* @return the plugin
*/
- public static IHttpCredentialsPlugin GetUiHttpCredentialsPlugin(Map conf) {
+ public static IHttpCredentialsPlugin GetUiHttpCredentialsPlugin(Map<String, Object> conf) {
String klassName = (String)conf.get(Config.UI_HTTP_CREDS_PLUGIN);
return AuthUtils.GetHttpCredentialsPlugin(conf, klassName);
}
@@ -350,7 +350,7 @@ public class AuthUtils {
* @param conf storm configuration
* @return the plugin
*/
- public static IHttpCredentialsPlugin GetDrpcHttpCredentialsPlugin(Map conf) {
+ public static IHttpCredentialsPlugin GetDrpcHttpCredentialsPlugin(Map<String, Object> conf) {
String klassName = (String)conf.get(Config.DRPC_HTTP_CREDS_PLUGIN);
return klassName == null ? null : AuthUtils.GetHttpCredentialsPlugin(conf, klassName);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/64e29f36/storm-client/src/jvm/org/apache/storm/security/auth/AutoSSL.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/AutoSSL.java b/storm-client/src/jvm/org/apache/storm/security/auth/AutoSSL.java
index 7838855..aa68153 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/AutoSSL.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/AutoSSL.java
@@ -45,23 +45,23 @@ import java.util.Map;
*/
public class AutoSSL implements IAutoCredentials {
private static final Logger LOG = LoggerFactory.getLogger(AutoSSL.class);
- private Map conf;
+ private Map<String, Object> conf;
private String writeDir = "./";
public static final String SSL_FILES_CONF = "ssl.credential.files";
- public void prepare(Map conf) {
+ public void prepare(Map<String, Object> conf) {
this.conf = conf;
writeDir = getSSLWriteDirFromConf(this.conf);
}
@VisibleForTesting
- protected String getSSLWriteDirFromConf(Map conf) {
+ protected String getSSLWriteDirFromConf(Map<String, Object> conf) {
return "./";
}
@VisibleForTesting
- Collection<String> getSSLFilesFromConf(Map conf) {
+ Collection<String> getSSLFilesFromConf(Map<String, Object> conf) {
Object sslConf = conf.get(SSL_FILES_CONF);
if (sslConf == null) {
LOG.info("No ssl files requested, if you want to use SSL please set {} to the " +