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 2015/10/09 16:53:32 UTC
[4/9] storm git commit: [STORM-893] - Resource Aware Scheduler
implementation. [STORM-894] - Basic Resource Aware Scheduling implementation.
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java b/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
index 6c2cc69..f023f63 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyInfo.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-9-3")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-9-30")
public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, TopologyInfo._Fields>, java.io.Serializable, Cloneable, Comparable<TopologyInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TopologyInfo");
@@ -1194,14 +1194,14 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
case 4: // EXECUTORS
if (schemeField.type == org.apache.thrift.protocol.TType.LIST) {
{
- org.apache.thrift.protocol.TList _list298 = iprot.readListBegin();
- struct.executors = new ArrayList<ExecutorSummary>(_list298.size);
- ExecutorSummary _elem299;
- for (int _i300 = 0; _i300 < _list298.size; ++_i300)
+ org.apache.thrift.protocol.TList _list308 = iprot.readListBegin();
+ struct.executors = new ArrayList<ExecutorSummary>(_list308.size);
+ ExecutorSummary _elem309;
+ for (int _i310 = 0; _i310 < _list308.size; ++_i310)
{
- _elem299 = new ExecutorSummary();
- _elem299.read(iprot);
- struct.executors.add(_elem299);
+ _elem309 = new ExecutorSummary();
+ _elem309.read(iprot);
+ struct.executors.add(_elem309);
}
iprot.readListEnd();
}
@@ -1221,26 +1221,26 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
case 6: // ERRORS
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map301 = iprot.readMapBegin();
- struct.errors = new HashMap<String,List<ErrorInfo>>(2*_map301.size);
- String _key302;
- List<ErrorInfo> _val303;
- for (int _i304 = 0; _i304 < _map301.size; ++_i304)
+ org.apache.thrift.protocol.TMap _map311 = iprot.readMapBegin();
+ struct.errors = new HashMap<String,List<ErrorInfo>>(2*_map311.size);
+ String _key312;
+ List<ErrorInfo> _val313;
+ for (int _i314 = 0; _i314 < _map311.size; ++_i314)
{
- _key302 = iprot.readString();
+ _key312 = iprot.readString();
{
- org.apache.thrift.protocol.TList _list305 = iprot.readListBegin();
- _val303 = new ArrayList<ErrorInfo>(_list305.size);
- ErrorInfo _elem306;
- for (int _i307 = 0; _i307 < _list305.size; ++_i307)
+ org.apache.thrift.protocol.TList _list315 = iprot.readListBegin();
+ _val313 = new ArrayList<ErrorInfo>(_list315.size);
+ ErrorInfo _elem316;
+ for (int _i317 = 0; _i317 < _list315.size; ++_i317)
{
- _elem306 = new ErrorInfo();
- _elem306.read(iprot);
- _val303.add(_elem306);
+ _elem316 = new ErrorInfo();
+ _elem316.read(iprot);
+ _val313.add(_elem316);
}
iprot.readListEnd();
}
- struct.errors.put(_key302, _val303);
+ struct.errors.put(_key312, _val313);
}
iprot.readMapEnd();
}
@@ -1252,16 +1252,16 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
case 7: // COMPONENT_DEBUG
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map308 = iprot.readMapBegin();
- struct.component_debug = new HashMap<String,DebugOptions>(2*_map308.size);
- String _key309;
- DebugOptions _val310;
- for (int _i311 = 0; _i311 < _map308.size; ++_i311)
+ org.apache.thrift.protocol.TMap _map318 = iprot.readMapBegin();
+ struct.component_debug = new HashMap<String,DebugOptions>(2*_map318.size);
+ String _key319;
+ DebugOptions _val320;
+ for (int _i321 = 0; _i321 < _map318.size; ++_i321)
{
- _key309 = iprot.readString();
- _val310 = new DebugOptions();
- _val310.read(iprot);
- struct.component_debug.put(_key309, _val310);
+ _key319 = iprot.readString();
+ _val320 = new DebugOptions();
+ _val320.read(iprot);
+ struct.component_debug.put(_key319, _val320);
}
iprot.readMapEnd();
}
@@ -1324,9 +1324,9 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
oprot.writeFieldBegin(EXECUTORS_FIELD_DESC);
{
oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.executors.size()));
- for (ExecutorSummary _iter312 : struct.executors)
+ for (ExecutorSummary _iter322 : struct.executors)
{
- _iter312.write(oprot);
+ _iter322.write(oprot);
}
oprot.writeListEnd();
}
@@ -1341,14 +1341,14 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
oprot.writeFieldBegin(ERRORS_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST, struct.errors.size()));
- for (Map.Entry<String, List<ErrorInfo>> _iter313 : struct.errors.entrySet())
+ for (Map.Entry<String, List<ErrorInfo>> _iter323 : struct.errors.entrySet())
{
- oprot.writeString(_iter313.getKey());
+ oprot.writeString(_iter323.getKey());
{
- oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, _iter313.getValue().size()));
- for (ErrorInfo _iter314 : _iter313.getValue())
+ oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, _iter323.getValue().size()));
+ for (ErrorInfo _iter324 : _iter323.getValue())
{
- _iter314.write(oprot);
+ _iter324.write(oprot);
}
oprot.writeListEnd();
}
@@ -1362,10 +1362,10 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
oprot.writeFieldBegin(COMPONENT_DEBUG_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, struct.component_debug.size()));
- for (Map.Entry<String, DebugOptions> _iter315 : struct.component_debug.entrySet())
+ for (Map.Entry<String, DebugOptions> _iter325 : struct.component_debug.entrySet())
{
- oprot.writeString(_iter315.getKey());
- _iter315.getValue().write(oprot);
+ oprot.writeString(_iter325.getKey());
+ _iter325.getValue().write(oprot);
}
oprot.writeMapEnd();
}
@@ -1413,22 +1413,22 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
oprot.writeI32(struct.uptime_secs);
{
oprot.writeI32(struct.executors.size());
- for (ExecutorSummary _iter316 : struct.executors)
+ for (ExecutorSummary _iter326 : struct.executors)
{
- _iter316.write(oprot);
+ _iter326.write(oprot);
}
}
oprot.writeString(struct.status);
{
oprot.writeI32(struct.errors.size());
- for (Map.Entry<String, List<ErrorInfo>> _iter317 : struct.errors.entrySet())
+ for (Map.Entry<String, List<ErrorInfo>> _iter327 : struct.errors.entrySet())
{
- oprot.writeString(_iter317.getKey());
+ oprot.writeString(_iter327.getKey());
{
- oprot.writeI32(_iter317.getValue().size());
- for (ErrorInfo _iter318 : _iter317.getValue())
+ oprot.writeI32(_iter327.getValue().size());
+ for (ErrorInfo _iter328 : _iter327.getValue())
{
- _iter318.write(oprot);
+ _iter328.write(oprot);
}
}
}
@@ -1450,10 +1450,10 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
if (struct.is_set_component_debug()) {
{
oprot.writeI32(struct.component_debug.size());
- for (Map.Entry<String, DebugOptions> _iter319 : struct.component_debug.entrySet())
+ for (Map.Entry<String, DebugOptions> _iter329 : struct.component_debug.entrySet())
{
- oprot.writeString(_iter319.getKey());
- _iter319.getValue().write(oprot);
+ oprot.writeString(_iter329.getKey());
+ _iter329.getValue().write(oprot);
}
}
}
@@ -1478,55 +1478,55 @@ public class TopologyInfo implements org.apache.thrift.TBase<TopologyInfo, Topol
struct.uptime_secs = iprot.readI32();
struct.set_uptime_secs_isSet(true);
{
- org.apache.thrift.protocol.TList _list320 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- struct.executors = new ArrayList<ExecutorSummary>(_list320.size);
- ExecutorSummary _elem321;
- for (int _i322 = 0; _i322 < _list320.size; ++_i322)
+ org.apache.thrift.protocol.TList _list330 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
+ struct.executors = new ArrayList<ExecutorSummary>(_list330.size);
+ ExecutorSummary _elem331;
+ for (int _i332 = 0; _i332 < _list330.size; ++_i332)
{
- _elem321 = new ExecutorSummary();
- _elem321.read(iprot);
- struct.executors.add(_elem321);
+ _elem331 = new ExecutorSummary();
+ _elem331.read(iprot);
+ struct.executors.add(_elem331);
}
}
struct.set_executors_isSet(true);
struct.status = iprot.readString();
struct.set_status_isSet(true);
{
- org.apache.thrift.protocol.TMap _map323 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST, iprot.readI32());
- struct.errors = new HashMap<String,List<ErrorInfo>>(2*_map323.size);
- String _key324;
- List<ErrorInfo> _val325;
- for (int _i326 = 0; _i326 < _map323.size; ++_i326)
+ org.apache.thrift.protocol.TMap _map333 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.LIST, iprot.readI32());
+ struct.errors = new HashMap<String,List<ErrorInfo>>(2*_map333.size);
+ String _key334;
+ List<ErrorInfo> _val335;
+ for (int _i336 = 0; _i336 < _map333.size; ++_i336)
{
- _key324 = iprot.readString();
+ _key334 = iprot.readString();
{
- org.apache.thrift.protocol.TList _list327 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- _val325 = new ArrayList<ErrorInfo>(_list327.size);
- ErrorInfo _elem328;
- for (int _i329 = 0; _i329 < _list327.size; ++_i329)
+ org.apache.thrift.protocol.TList _list337 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
+ _val335 = new ArrayList<ErrorInfo>(_list337.size);
+ ErrorInfo _elem338;
+ for (int _i339 = 0; _i339 < _list337.size; ++_i339)
{
- _elem328 = new ErrorInfo();
- _elem328.read(iprot);
- _val325.add(_elem328);
+ _elem338 = new ErrorInfo();
+ _elem338.read(iprot);
+ _val335.add(_elem338);
}
}
- struct.errors.put(_key324, _val325);
+ struct.errors.put(_key334, _val335);
}
}
struct.set_errors_isSet(true);
BitSet incoming = iprot.readBitSet(4);
if (incoming.get(0)) {
{
- org.apache.thrift.protocol.TMap _map330 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- struct.component_debug = new HashMap<String,DebugOptions>(2*_map330.size);
- String _key331;
- DebugOptions _val332;
- for (int _i333 = 0; _i333 < _map330.size; ++_i333)
+ org.apache.thrift.protocol.TMap _map340 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
+ struct.component_debug = new HashMap<String,DebugOptions>(2*_map340.size);
+ String _key341;
+ DebugOptions _val342;
+ for (int _i343 = 0; _i343 < _map340.size; ++_i343)
{
- _key331 = iprot.readString();
- _val332 = new DebugOptions();
- _val332.read(iprot);
- struct.component_debug.put(_key331, _val332);
+ _key341 = iprot.readString();
+ _val342 = new DebugOptions();
+ _val342.read(iprot);
+ struct.component_debug.put(_key341, _val342);
}
}
struct.set_component_debug_isSet(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java b/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
index c31e23e..3528ca2 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-9-28")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-10-6")
public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInfo, TopologyPageInfo._Fields>, java.io.Serializable, Cloneable, Comparable<TopologyPageInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TopologyPageInfo");
@@ -1585,16 +1585,16 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
case 9: // ID_TO_SPOUT_AGG_STATS
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map384 = iprot.readMapBegin();
- struct.id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map384.size);
- String _key385;
- ComponentAggregateStats _val386;
- for (int _i387 = 0; _i387 < _map384.size; ++_i387)
+ org.apache.thrift.protocol.TMap _map394 = iprot.readMapBegin();
+ struct.id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map394.size);
+ String _key395;
+ ComponentAggregateStats _val396;
+ for (int _i397 = 0; _i397 < _map394.size; ++_i397)
{
- _key385 = iprot.readString();
- _val386 = new ComponentAggregateStats();
- _val386.read(iprot);
- struct.id_to_spout_agg_stats.put(_key385, _val386);
+ _key395 = iprot.readString();
+ _val396 = new ComponentAggregateStats();
+ _val396.read(iprot);
+ struct.id_to_spout_agg_stats.put(_key395, _val396);
}
iprot.readMapEnd();
}
@@ -1606,16 +1606,16 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
case 10: // ID_TO_BOLT_AGG_STATS
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map388 = iprot.readMapBegin();
- struct.id_to_bolt_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map388.size);
- String _key389;
- ComponentAggregateStats _val390;
- for (int _i391 = 0; _i391 < _map388.size; ++_i391)
+ org.apache.thrift.protocol.TMap _map398 = iprot.readMapBegin();
+ struct.id_to_bolt_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map398.size);
+ String _key399;
+ ComponentAggregateStats _val400;
+ for (int _i401 = 0; _i401 < _map398.size; ++_i401)
{
- _key389 = iprot.readString();
- _val390 = new ComponentAggregateStats();
- _val390.read(iprot);
- struct.id_to_bolt_agg_stats.put(_key389, _val390);
+ _key399 = iprot.readString();
+ _val400 = new ComponentAggregateStats();
+ _val400.read(iprot);
+ struct.id_to_bolt_agg_stats.put(_key399, _val400);
}
iprot.readMapEnd();
}
@@ -1730,10 +1730,10 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
oprot.writeFieldBegin(ID_TO_SPOUT_AGG_STATS_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, struct.id_to_spout_agg_stats.size()));
- for (Map.Entry<String, ComponentAggregateStats> _iter392 : struct.id_to_spout_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter402 : struct.id_to_spout_agg_stats.entrySet())
{
- oprot.writeString(_iter392.getKey());
- _iter392.getValue().write(oprot);
+ oprot.writeString(_iter402.getKey());
+ _iter402.getValue().write(oprot);
}
oprot.writeMapEnd();
}
@@ -1745,10 +1745,10 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
oprot.writeFieldBegin(ID_TO_BOLT_AGG_STATS_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, struct.id_to_bolt_agg_stats.size()));
- for (Map.Entry<String, ComponentAggregateStats> _iter393 : struct.id_to_bolt_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter403 : struct.id_to_bolt_agg_stats.entrySet())
{
- oprot.writeString(_iter393.getKey());
- _iter393.getValue().write(oprot);
+ oprot.writeString(_iter403.getKey());
+ _iter403.getValue().write(oprot);
}
oprot.writeMapEnd();
}
@@ -1874,20 +1874,20 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
if (struct.is_set_id_to_spout_agg_stats()) {
{
oprot.writeI32(struct.id_to_spout_agg_stats.size());
- for (Map.Entry<String, ComponentAggregateStats> _iter394 : struct.id_to_spout_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter404 : struct.id_to_spout_agg_stats.entrySet())
{
- oprot.writeString(_iter394.getKey());
- _iter394.getValue().write(oprot);
+ oprot.writeString(_iter404.getKey());
+ _iter404.getValue().write(oprot);
}
}
}
if (struct.is_set_id_to_bolt_agg_stats()) {
{
oprot.writeI32(struct.id_to_bolt_agg_stats.size());
- for (Map.Entry<String, ComponentAggregateStats> _iter395 : struct.id_to_bolt_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter405 : struct.id_to_bolt_agg_stats.entrySet())
{
- oprot.writeString(_iter395.getKey());
- _iter395.getValue().write(oprot);
+ oprot.writeString(_iter405.getKey());
+ _iter405.getValue().write(oprot);
}
}
}
@@ -1944,32 +1944,32 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
}
if (incoming.get(7)) {
{
- org.apache.thrift.protocol.TMap _map396 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- struct.id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map396.size);
- String _key397;
- ComponentAggregateStats _val398;
- for (int _i399 = 0; _i399 < _map396.size; ++_i399)
+ org.apache.thrift.protocol.TMap _map406 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
+ struct.id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map406.size);
+ String _key407;
+ ComponentAggregateStats _val408;
+ for (int _i409 = 0; _i409 < _map406.size; ++_i409)
{
- _key397 = iprot.readString();
- _val398 = new ComponentAggregateStats();
- _val398.read(iprot);
- struct.id_to_spout_agg_stats.put(_key397, _val398);
+ _key407 = iprot.readString();
+ _val408 = new ComponentAggregateStats();
+ _val408.read(iprot);
+ struct.id_to_spout_agg_stats.put(_key407, _val408);
}
}
struct.set_id_to_spout_agg_stats_isSet(true);
}
if (incoming.get(8)) {
{
- org.apache.thrift.protocol.TMap _map400 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- struct.id_to_bolt_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map400.size);
- String _key401;
- ComponentAggregateStats _val402;
- for (int _i403 = 0; _i403 < _map400.size; ++_i403)
+ org.apache.thrift.protocol.TMap _map410 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
+ struct.id_to_bolt_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map410.size);
+ String _key411;
+ ComponentAggregateStats _val412;
+ for (int _i413 = 0; _i413 < _map410.size; ++_i413)
{
- _key401 = iprot.readString();
- _val402 = new ComponentAggregateStats();
- _val402.read(iprot);
- struct.id_to_bolt_agg_stats.put(_key401, _val402);
+ _key411 = iprot.readString();
+ _val412 = new ComponentAggregateStats();
+ _val412.read(iprot);
+ struct.id_to_bolt_agg_stats.put(_key411, _val412);
}
}
struct.set_id_to_bolt_agg_stats_isSet(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java b/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
index 453ead2..b2f5b23 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
@@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-9-28")
+@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-10-6")
public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, TopologyStats._Fields>, java.io.Serializable, Cloneable, Comparable<TopologyStats> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TopologyStats");
@@ -737,15 +737,15 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
case 1: // WINDOW_TO_EMITTED
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map334 = iprot.readMapBegin();
- struct.window_to_emitted = new HashMap<String,Long>(2*_map334.size);
- String _key335;
- long _val336;
- for (int _i337 = 0; _i337 < _map334.size; ++_i337)
+ org.apache.thrift.protocol.TMap _map344 = iprot.readMapBegin();
+ struct.window_to_emitted = new HashMap<String,Long>(2*_map344.size);
+ String _key345;
+ long _val346;
+ for (int _i347 = 0; _i347 < _map344.size; ++_i347)
{
- _key335 = iprot.readString();
- _val336 = iprot.readI64();
- struct.window_to_emitted.put(_key335, _val336);
+ _key345 = iprot.readString();
+ _val346 = iprot.readI64();
+ struct.window_to_emitted.put(_key345, _val346);
}
iprot.readMapEnd();
}
@@ -757,15 +757,15 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
case 2: // WINDOW_TO_TRANSFERRED
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map338 = iprot.readMapBegin();
- struct.window_to_transferred = new HashMap<String,Long>(2*_map338.size);
- String _key339;
- long _val340;
- for (int _i341 = 0; _i341 < _map338.size; ++_i341)
+ org.apache.thrift.protocol.TMap _map348 = iprot.readMapBegin();
+ struct.window_to_transferred = new HashMap<String,Long>(2*_map348.size);
+ String _key349;
+ long _val350;
+ for (int _i351 = 0; _i351 < _map348.size; ++_i351)
{
- _key339 = iprot.readString();
- _val340 = iprot.readI64();
- struct.window_to_transferred.put(_key339, _val340);
+ _key349 = iprot.readString();
+ _val350 = iprot.readI64();
+ struct.window_to_transferred.put(_key349, _val350);
}
iprot.readMapEnd();
}
@@ -777,15 +777,15 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
case 3: // WINDOW_TO_COMPLETE_LATENCIES_MS
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map342 = iprot.readMapBegin();
- struct.window_to_complete_latencies_ms = new HashMap<String,Double>(2*_map342.size);
- String _key343;
- double _val344;
- for (int _i345 = 0; _i345 < _map342.size; ++_i345)
+ org.apache.thrift.protocol.TMap _map352 = iprot.readMapBegin();
+ struct.window_to_complete_latencies_ms = new HashMap<String,Double>(2*_map352.size);
+ String _key353;
+ double _val354;
+ for (int _i355 = 0; _i355 < _map352.size; ++_i355)
{
- _key343 = iprot.readString();
- _val344 = iprot.readDouble();
- struct.window_to_complete_latencies_ms.put(_key343, _val344);
+ _key353 = iprot.readString();
+ _val354 = iprot.readDouble();
+ struct.window_to_complete_latencies_ms.put(_key353, _val354);
}
iprot.readMapEnd();
}
@@ -797,15 +797,15 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
case 4: // WINDOW_TO_ACKED
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map346 = iprot.readMapBegin();
- struct.window_to_acked = new HashMap<String,Long>(2*_map346.size);
- String _key347;
- long _val348;
- for (int _i349 = 0; _i349 < _map346.size; ++_i349)
+ org.apache.thrift.protocol.TMap _map356 = iprot.readMapBegin();
+ struct.window_to_acked = new HashMap<String,Long>(2*_map356.size);
+ String _key357;
+ long _val358;
+ for (int _i359 = 0; _i359 < _map356.size; ++_i359)
{
- _key347 = iprot.readString();
- _val348 = iprot.readI64();
- struct.window_to_acked.put(_key347, _val348);
+ _key357 = iprot.readString();
+ _val358 = iprot.readI64();
+ struct.window_to_acked.put(_key357, _val358);
}
iprot.readMapEnd();
}
@@ -817,15 +817,15 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
case 5: // WINDOW_TO_FAILED
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
- org.apache.thrift.protocol.TMap _map350 = iprot.readMapBegin();
- struct.window_to_failed = new HashMap<String,Long>(2*_map350.size);
- String _key351;
- long _val352;
- for (int _i353 = 0; _i353 < _map350.size; ++_i353)
+ org.apache.thrift.protocol.TMap _map360 = iprot.readMapBegin();
+ struct.window_to_failed = new HashMap<String,Long>(2*_map360.size);
+ String _key361;
+ long _val362;
+ for (int _i363 = 0; _i363 < _map360.size; ++_i363)
{
- _key351 = iprot.readString();
- _val352 = iprot.readI64();
- struct.window_to_failed.put(_key351, _val352);
+ _key361 = iprot.readString();
+ _val362 = iprot.readI64();
+ struct.window_to_failed.put(_key361, _val362);
}
iprot.readMapEnd();
}
@@ -852,10 +852,10 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
oprot.writeFieldBegin(WINDOW_TO_EMITTED_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, struct.window_to_emitted.size()));
- for (Map.Entry<String, Long> _iter354 : struct.window_to_emitted.entrySet())
+ for (Map.Entry<String, Long> _iter364 : struct.window_to_emitted.entrySet())
{
- oprot.writeString(_iter354.getKey());
- oprot.writeI64(_iter354.getValue());
+ oprot.writeString(_iter364.getKey());
+ oprot.writeI64(_iter364.getValue());
}
oprot.writeMapEnd();
}
@@ -867,10 +867,10 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
oprot.writeFieldBegin(WINDOW_TO_TRANSFERRED_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, struct.window_to_transferred.size()));
- for (Map.Entry<String, Long> _iter355 : struct.window_to_transferred.entrySet())
+ for (Map.Entry<String, Long> _iter365 : struct.window_to_transferred.entrySet())
{
- oprot.writeString(_iter355.getKey());
- oprot.writeI64(_iter355.getValue());
+ oprot.writeString(_iter365.getKey());
+ oprot.writeI64(_iter365.getValue());
}
oprot.writeMapEnd();
}
@@ -882,10 +882,10 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
oprot.writeFieldBegin(WINDOW_TO_COMPLETE_LATENCIES_MS_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.DOUBLE, struct.window_to_complete_latencies_ms.size()));
- for (Map.Entry<String, Double> _iter356 : struct.window_to_complete_latencies_ms.entrySet())
+ for (Map.Entry<String, Double> _iter366 : struct.window_to_complete_latencies_ms.entrySet())
{
- oprot.writeString(_iter356.getKey());
- oprot.writeDouble(_iter356.getValue());
+ oprot.writeString(_iter366.getKey());
+ oprot.writeDouble(_iter366.getValue());
}
oprot.writeMapEnd();
}
@@ -897,10 +897,10 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
oprot.writeFieldBegin(WINDOW_TO_ACKED_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, struct.window_to_acked.size()));
- for (Map.Entry<String, Long> _iter357 : struct.window_to_acked.entrySet())
+ for (Map.Entry<String, Long> _iter367 : struct.window_to_acked.entrySet())
{
- oprot.writeString(_iter357.getKey());
- oprot.writeI64(_iter357.getValue());
+ oprot.writeString(_iter367.getKey());
+ oprot.writeI64(_iter367.getValue());
}
oprot.writeMapEnd();
}
@@ -912,10 +912,10 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
oprot.writeFieldBegin(WINDOW_TO_FAILED_FIELD_DESC);
{
oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, struct.window_to_failed.size()));
- for (Map.Entry<String, Long> _iter358 : struct.window_to_failed.entrySet())
+ for (Map.Entry<String, Long> _iter368 : struct.window_to_failed.entrySet())
{
- oprot.writeString(_iter358.getKey());
- oprot.writeI64(_iter358.getValue());
+ oprot.writeString(_iter368.getKey());
+ oprot.writeI64(_iter368.getValue());
}
oprot.writeMapEnd();
}
@@ -959,50 +959,50 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
if (struct.is_set_window_to_emitted()) {
{
oprot.writeI32(struct.window_to_emitted.size());
- for (Map.Entry<String, Long> _iter359 : struct.window_to_emitted.entrySet())
+ for (Map.Entry<String, Long> _iter369 : struct.window_to_emitted.entrySet())
{
- oprot.writeString(_iter359.getKey());
- oprot.writeI64(_iter359.getValue());
+ oprot.writeString(_iter369.getKey());
+ oprot.writeI64(_iter369.getValue());
}
}
}
if (struct.is_set_window_to_transferred()) {
{
oprot.writeI32(struct.window_to_transferred.size());
- for (Map.Entry<String, Long> _iter360 : struct.window_to_transferred.entrySet())
+ for (Map.Entry<String, Long> _iter370 : struct.window_to_transferred.entrySet())
{
- oprot.writeString(_iter360.getKey());
- oprot.writeI64(_iter360.getValue());
+ oprot.writeString(_iter370.getKey());
+ oprot.writeI64(_iter370.getValue());
}
}
}
if (struct.is_set_window_to_complete_latencies_ms()) {
{
oprot.writeI32(struct.window_to_complete_latencies_ms.size());
- for (Map.Entry<String, Double> _iter361 : struct.window_to_complete_latencies_ms.entrySet())
+ for (Map.Entry<String, Double> _iter371 : struct.window_to_complete_latencies_ms.entrySet())
{
- oprot.writeString(_iter361.getKey());
- oprot.writeDouble(_iter361.getValue());
+ oprot.writeString(_iter371.getKey());
+ oprot.writeDouble(_iter371.getValue());
}
}
}
if (struct.is_set_window_to_acked()) {
{
oprot.writeI32(struct.window_to_acked.size());
- for (Map.Entry<String, Long> _iter362 : struct.window_to_acked.entrySet())
+ for (Map.Entry<String, Long> _iter372 : struct.window_to_acked.entrySet())
{
- oprot.writeString(_iter362.getKey());
- oprot.writeI64(_iter362.getValue());
+ oprot.writeString(_iter372.getKey());
+ oprot.writeI64(_iter372.getValue());
}
}
}
if (struct.is_set_window_to_failed()) {
{
oprot.writeI32(struct.window_to_failed.size());
- for (Map.Entry<String, Long> _iter363 : struct.window_to_failed.entrySet())
+ for (Map.Entry<String, Long> _iter373 : struct.window_to_failed.entrySet())
{
- oprot.writeString(_iter363.getKey());
- oprot.writeI64(_iter363.getValue());
+ oprot.writeString(_iter373.getKey());
+ oprot.writeI64(_iter373.getValue());
}
}
}
@@ -1014,75 +1014,75 @@ public class TopologyStats implements org.apache.thrift.TBase<TopologyStats, Top
BitSet incoming = iprot.readBitSet(5);
if (incoming.get(0)) {
{
- org.apache.thrift.protocol.TMap _map364 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
- struct.window_to_emitted = new HashMap<String,Long>(2*_map364.size);
- String _key365;
- long _val366;
- for (int _i367 = 0; _i367 < _map364.size; ++_i367)
+ org.apache.thrift.protocol.TMap _map374 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
+ struct.window_to_emitted = new HashMap<String,Long>(2*_map374.size);
+ String _key375;
+ long _val376;
+ for (int _i377 = 0; _i377 < _map374.size; ++_i377)
{
- _key365 = iprot.readString();
- _val366 = iprot.readI64();
- struct.window_to_emitted.put(_key365, _val366);
+ _key375 = iprot.readString();
+ _val376 = iprot.readI64();
+ struct.window_to_emitted.put(_key375, _val376);
}
}
struct.set_window_to_emitted_isSet(true);
}
if (incoming.get(1)) {
{
- org.apache.thrift.protocol.TMap _map368 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
- struct.window_to_transferred = new HashMap<String,Long>(2*_map368.size);
- String _key369;
- long _val370;
- for (int _i371 = 0; _i371 < _map368.size; ++_i371)
+ org.apache.thrift.protocol.TMap _map378 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
+ struct.window_to_transferred = new HashMap<String,Long>(2*_map378.size);
+ String _key379;
+ long _val380;
+ for (int _i381 = 0; _i381 < _map378.size; ++_i381)
{
- _key369 = iprot.readString();
- _val370 = iprot.readI64();
- struct.window_to_transferred.put(_key369, _val370);
+ _key379 = iprot.readString();
+ _val380 = iprot.readI64();
+ struct.window_to_transferred.put(_key379, _val380);
}
}
struct.set_window_to_transferred_isSet(true);
}
if (incoming.get(2)) {
{
- org.apache.thrift.protocol.TMap _map372 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.DOUBLE, iprot.readI32());
- struct.window_to_complete_latencies_ms = new HashMap<String,Double>(2*_map372.size);
- String _key373;
- double _val374;
- for (int _i375 = 0; _i375 < _map372.size; ++_i375)
+ org.apache.thrift.protocol.TMap _map382 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.DOUBLE, iprot.readI32());
+ struct.window_to_complete_latencies_ms = new HashMap<String,Double>(2*_map382.size);
+ String _key383;
+ double _val384;
+ for (int _i385 = 0; _i385 < _map382.size; ++_i385)
{
- _key373 = iprot.readString();
- _val374 = iprot.readDouble();
- struct.window_to_complete_latencies_ms.put(_key373, _val374);
+ _key383 = iprot.readString();
+ _val384 = iprot.readDouble();
+ struct.window_to_complete_latencies_ms.put(_key383, _val384);
}
}
struct.set_window_to_complete_latencies_ms_isSet(true);
}
if (incoming.get(3)) {
{
- org.apache.thrift.protocol.TMap _map376 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
- struct.window_to_acked = new HashMap<String,Long>(2*_map376.size);
- String _key377;
- long _val378;
- for (int _i379 = 0; _i379 < _map376.size; ++_i379)
+ org.apache.thrift.protocol.TMap _map386 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
+ struct.window_to_acked = new HashMap<String,Long>(2*_map386.size);
+ String _key387;
+ long _val388;
+ for (int _i389 = 0; _i389 < _map386.size; ++_i389)
{
- _key377 = iprot.readString();
- _val378 = iprot.readI64();
- struct.window_to_acked.put(_key377, _val378);
+ _key387 = iprot.readString();
+ _val388 = iprot.readI64();
+ struct.window_to_acked.put(_key387, _val388);
}
}
struct.set_window_to_acked_isSet(true);
}
if (incoming.get(4)) {
{
- org.apache.thrift.protocol.TMap _map380 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
- struct.window_to_failed = new HashMap<String,Long>(2*_map380.size);
- String _key381;
- long _val382;
- for (int _i383 = 0; _i383 < _map380.size; ++_i383)
+ org.apache.thrift.protocol.TMap _map390 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.I64, iprot.readI32());
+ struct.window_to_failed = new HashMap<String,Long>(2*_map390.size);
+ String _key391;
+ long _val392;
+ for (int _i393 = 0; _i393 < _map390.size; ++_i393)
{
- _key381 = iprot.readString();
- _val382 = iprot.readI64();
- struct.window_to_failed.put(_key381, _val382);
+ _key391 = iprot.readString();
+ _val392 = iprot.readI64();
+ struct.window_to_failed.put(_key391, _val392);
}
}
struct.set_window_to_failed_isSet(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java b/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java
new file mode 100644
index 0000000..e91e54a
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/networktopography/AbstractDNSToSwitchMapping.java
@@ -0,0 +1,95 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package backtype.storm.networktopography;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * This is a base class for DNS to Switch mappings. <p/> It is not mandatory to
+ * derive {@link DNSToSwitchMapping} implementations from it, but it is strongly
+ * recommended, as it makes it easy for the developers to add new methods
+ * to this base class that are automatically picked up by all implementations.
+ * <p/>
+ *
+ */
+public abstract class AbstractDNSToSwitchMapping
+ implements DNSToSwitchMapping {
+
+ /**
+ * Create an unconfigured instance
+ */
+ protected AbstractDNSToSwitchMapping() {
+ }
+
+ /**
+ * Predicate that indicates that the switch mapping is known to be
+ * single-switch. The base class returns false: it assumes all mappings are
+ * multi-rack. Subclasses may override this with methods that are more aware
+ * of their topologies.
+ *
+ * <p/>
+ *
+ *
+ * @return true if the mapping thinks that it is on a single switch
+ */
+ public boolean isSingleSwitch() {
+ return false;
+ }
+
+ /**
+ * Get a copy of the map (for diagnostics)
+ * @return a clone of the map or null for none known
+ */
+ public Map<String, String> getSwitchMap() {
+ return null;
+ }
+
+ /**
+ * Generate a string listing the switch mapping implementation,
+ * the mapping for every known node and the number of nodes and
+ * unique switches known about -each entry to a separate line.
+ * @return a string that can be presented to the ops team or used in
+ * debug messages.
+ */
+ public String dumpTopology() {
+ Map<String, String> rack = getSwitchMap();
+ StringBuilder builder = new StringBuilder();
+ builder.append("Mapping: ").append(toString()).append("\n");
+ if (rack != null) {
+ builder.append("Map:\n");
+ Set<String> switches = new HashSet<String>();
+ for (Map.Entry<String, String> entry : rack.entrySet()) {
+ builder.append(" ")
+ .append(entry.getKey())
+ .append(" -> ")
+ .append(entry.getValue())
+ .append("\n");
+ switches.add(entry.getValue());
+ }
+ builder.append("Nodes: ").append(rack.size()).append("\n");
+ builder.append("Switches: ").append(switches.size()).append("\n");
+ } else {
+ builder.append("No topology information");
+ }
+ return builder.toString();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/networktopography/DNSToSwitchMapping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/networktopography/DNSToSwitchMapping.java b/storm-core/src/jvm/backtype/storm/networktopography/DNSToSwitchMapping.java
new file mode 100644
index 0000000..fbfbb19
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/networktopography/DNSToSwitchMapping.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package backtype.storm.networktopography;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An interface that must be implemented to allow pluggable
+ * DNS-name/IP-address to RackID resolvers.
+ *
+ */
+public interface DNSToSwitchMapping {
+ public final static String DEFAULT_RACK = "/default-rack";
+
+ /**
+ * Resolves a list of DNS-names/IP-address and returns back a map of DNS-name->switch information ( network paths).
+ * Consider an element in the argument list - x.y.com. The switch information
+ * that is returned must be a network path of the form /foo/rack,
+ * where / is the root, and 'foo' is the switch where 'rack' is connected.
+ * Note the hostname/ip-address is not part of the returned path.
+ * The network topology of the cluster would determine the number of
+ * components in the network path.
+ * <p/>
+ *
+ * If a name cannot be resolved to a rack, the implementation
+ * should return {DEFAULT_RACK}. This
+ * is what the bundled implementations do, though it is not a formal requirement
+ *
+ * @param names the list of hosts to resolve (can be empty)
+ * @return Map of hosts to resolved network paths.
+ * If <i>names</i> is empty, then return empty Map
+ */
+ public Map<String, String> resolve(List<String> names);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java b/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java
new file mode 100644
index 0000000..f3f6152
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/networktopography/DefaultRackDNSToSwitchMapping.java
@@ -0,0 +1,35 @@
+package backtype.storm.networktopography;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This class implements the {@link DNSToSwitchMapping} interface
+ * It returns the DEFAULT_RACK for every host.
+ */
+public final class DefaultRackDNSToSwitchMapping extends AbstractDNSToSwitchMapping {
+
+ private Map<String, String> mappingCache = new ConcurrentHashMap<String, String>();
+
+ @Override
+ public Map<String,String> resolve(List<String> names) {
+
+ Map<String, String> m = new HashMap<String, String>();
+ if (names.isEmpty()) {
+ //name list is empty, return an empty map
+ return m;
+ }
+ for (String name : names) {
+ m.put(name, DEFAULT_RACK);
+ mappingCache.put(name, DEFAULT_RACK);
+ }
+ return m;
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultRackDNSToSwitchMapping (" + mappingCache.size() + " mappings cached)" + dumpTopology();
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
index e0c7cc7..dd15999 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java
@@ -25,12 +25,22 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import backtype.storm.Config;
+import backtype.storm.networktopography.DNSToSwitchMapping;
+import backtype.storm.utils.Utils;
+
public class Cluster {
/**
* key: supervisor id, value: supervisor details
*/
private Map<String, SupervisorDetails> supervisors;
+
+ /**
+ * key: rack, value: nodes in that rack
+ */
+ private Map<String, List<String>> networkTopography;
+
/**
* key: topologyId, value: topology's current assignments.
*/
@@ -44,11 +54,13 @@ public class Cluster {
* a map from hostname to supervisor id.
*/
private Map<String, List<String>> hostToId;
-
+
+ private Map conf = null;
+
private Set<String> blackListedHosts = new HashSet<String>();
private INimbus inimbus;
- public Cluster(INimbus nimbus, Map<String, SupervisorDetails> supervisors, Map<String, SchedulerAssignmentImpl> assignments){
+ public Cluster(INimbus nimbus, Map<String, SupervisorDetails> supervisors, Map<String, SchedulerAssignmentImpl> assignments, Map storm_conf){
this.inimbus = nimbus;
this.supervisors = new HashMap<String, SupervisorDetails>(supervisors.size());
this.supervisors.putAll(supervisors);
@@ -64,6 +76,7 @@ public class Cluster {
}
this.hostToId.get(host).add(nodeId);
}
+ this.conf = storm_conf;
}
public void setBlacklistedHosts(Set<String> hosts) {
@@ -438,6 +451,35 @@ public class Cluster {
return this.supervisors;
}
+ /*
+ * Note: Make sure the proper conf was passed into the Cluster constructor before calling this function
+ * It tries to load the proper network topography detection plugin specified in the config.
+ */
+ public Map<String, List<String>> getNetworkTopography() {
+ if (networkTopography == null) {
+ networkTopography = new HashMap<String, List<String>>();
+ ArrayList<String> supervisorHostNames = new ArrayList<String>();
+ for (SupervisorDetails s : supervisors.values()) {
+ supervisorHostNames.add(s.getHost());
+ }
+
+ String clazz = (String) conf.get(Config.STORM_NETWORK_TOPOGRAPHY_PLUGIN);
+ DNSToSwitchMapping topographyMapper = (DNSToSwitchMapping) Utils.newInstance(clazz);
+
+ Map<String, String> resolvedSuperVisors = topographyMapper.resolve(supervisorHostNames);
+ for (String hostName : resolvedSuperVisors.keySet()) {
+ String rack = resolvedSuperVisors.get(hostName);
+
+ if (!networkTopography.containsKey(rack)) {
+ networkTopography.put(rack, new ArrayList<String>());
+ }
+ List<String> nodesForRack = networkTopography.get(rack);
+ nodesForRack.add(hostName);
+ }
+ }
+ return networkTopography;
+ }
+
public void setStatus(String topologyId, String status) {
this.status.put(topologyId, status);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java b/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java
index 7497f26..65e6e9b 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/SupervisorDetails.java
@@ -20,8 +20,14 @@ package backtype.storm.scheduler;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
+import java.util.Map;
+
+import backtype.storm.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SupervisorDetails {
+ private static final Logger LOG = LoggerFactory.getLogger(SupervisorDetails.class);
String id;
/**
@@ -37,25 +43,46 @@ public class SupervisorDetails {
* all the ports of the supervisor
*/
Set<Integer> allPorts;
+ /**
+ * Map containing a manifest of resources for the node the superivsor resides
+ */
+ private Map<String, Double> _total_resources;
+
+ public SupervisorDetails(String id, String host, Object meta, Object schedulerMeta,
+ Collection<Number> allPorts, Map<String, Double> total_resources){
- public SupervisorDetails(String id, Object meta){
this.id = id;
+ this.host = host;
this.meta = meta;
- allPorts = new HashSet();
+ this.schedulerMeta = schedulerMeta;
+ if(allPorts!=null) {
+ setAllPorts(allPorts);
+ } else {
+ this.allPorts = new HashSet();
+ }
+ this._total_resources = total_resources;
+ LOG.debug("Creating a new supervisor ({}-{}) with resources: {}", this.host, this.id, total_resources);
}
-
+
+ public SupervisorDetails(String id, Object meta){
+ this(id, null, meta, null, null, null);
+ }
+
+ public SupervisorDetails(String id, Object meta, Map<String, Double> total_resources) {
+ this(id, null, meta, null, null, total_resources);
+ }
+
public SupervisorDetails(String id, Object meta, Collection<Number> allPorts){
- this.id = id;
- this.meta = meta;
- setAllPorts(allPorts);
+ this(id, null, meta, null, allPorts, null);
}
- public SupervisorDetails(String id, String host, Object schedulerMeta, Collection<Number> allPorts){
- this.id = id;
- this.host = host;
- this.schedulerMeta = schedulerMeta;
+ public SupervisorDetails(String id, String host, Object schedulerMeta, Collection<Number> allPorts) {
+ this(id, host, null, schedulerMeta, allPorts, null);
+ }
- setAllPorts(allPorts);
+ public SupervisorDetails(String id, String host, Object schedulerMeta,
+ Collection<Number> allPorts, Map<String, Double> total_resources) {
+ this(id, host, null, schedulerMeta, allPorts, total_resources);
}
private void setAllPorts(Collection<Number> allPorts) {
@@ -86,4 +113,16 @@ public class SupervisorDetails {
public Object getSchedulerMeta() {
return this.schedulerMeta;
}
+
+ private Double getTotalResource(String type) {
+ return this._total_resources.get(type);
+ }
+
+ public Double getTotalMemory() {
+ return getTotalResource(Config.SUPERVISOR_MEMORY_CAPACITY_MB);
+ }
+
+ public Double getTotalCPU() {
+ return getTotalResource(Config.SUPERVISOR_CPU_CAPACITY);
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
index 70af1b4..409ecb3 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/Topologies.java
@@ -21,10 +21,13 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import backtype.storm.scheduler.resource.Component;
+
public class Topologies {
Map<String, TopologyDetails> topologies;
Map<String, String> nameToId;
-
+ Map<String, Map<String, Component>> _allComponents;
+
public Topologies(Map<String, TopologyDetails> topologies) {
if(topologies==null) topologies = new HashMap();
this.topologies = new HashMap<String, TopologyDetails>(topologies.size());
@@ -54,4 +57,14 @@ public class Topologies {
public Collection<TopologyDetails> getTopologies() {
return this.topologies.values();
}
+
+ public Map<String, Map<String, Component>> getAllComponents() {
+ if (_allComponents == null) {
+ _allComponents = new HashMap<>();
+ for (Map.Entry<String, TopologyDetails> entry : this.topologies.entrySet()) {
+ _allComponents.put(entry.getKey(), entry.getValue().getComponents());
+ }
+ }
+ return _allComponents;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java b/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java
index 6daf4ed..872a834 100644
--- a/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java
+++ b/storm-core/src/jvm/backtype/storm/scheduler/TopologyDetails.java
@@ -17,12 +17,24 @@
*/
package backtype.storm.scheduler;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import backtype.storm.Config;
+import backtype.storm.generated.Bolt;
+import backtype.storm.generated.GlobalStreamId;
+import backtype.storm.generated.Grouping;
+import backtype.storm.generated.SpoutSpec;
import backtype.storm.generated.StormTopology;
+import backtype.storm.scheduler.resource.Component;
+import backtype.storm.scheduler.resource.ResourceUtils;
+import backtype.storm.utils.Utils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TopologyDetails {
@@ -31,41 +43,45 @@ public class TopologyDetails {
StormTopology topology;
Map<ExecutorDetails, String> executorToComponent;
int numWorkers;
-
+ //<ExecutorDetails - Task, Map<String - Type of resource, Map<String - type of that resource, Double - amount>>>
+ private Map<ExecutorDetails, Map<String, Double>> _resourceList;
+ //Max heap size for a worker used by topology
+ private Double topologyWorkerMaxHeapSize;
+
+ private static final Logger LOG = LoggerFactory.getLogger(TopologyDetails.class);
+
public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology, int numWorkers) {
this.topologyId = topologyId;
this.topologyConf = topologyConf;
this.topology = topology;
this.numWorkers = numWorkers;
}
-
+
public TopologyDetails(String topologyId, Map topologyConf, StormTopology topology, int numWorkers, Map<ExecutorDetails, String> executorToComponents) {
this(topologyId, topologyConf, topology, numWorkers);
this.executorToComponent = new HashMap<ExecutorDetails, String>(0);
if (executorToComponents != null) {
this.executorToComponent.putAll(executorToComponents);
}
+ this.initResourceList();
+ this.initConfigs();
}
-
+
public String getId() {
return topologyId;
}
-
+
public String getName() {
- return (String)this.topologyConf.get(Config.TOPOLOGY_NAME);
+ return (String) this.topologyConf.get(Config.TOPOLOGY_NAME);
}
-
+
public Map getConf() {
return topologyConf;
}
-
+
public int getNumWorkers() {
return numWorkers;
}
-
- public StormTopology getTopology() {
- return topology;
- }
public Map<ExecutorDetails, String> getExecutorToComponent() {
return this.executorToComponent;
@@ -79,11 +95,299 @@ public class TopologyDetails {
ret.put(executor, compId);
}
}
-
+
return ret;
}
-
+
public Collection<ExecutorDetails> getExecutors() {
return this.executorToComponent.keySet();
}
+
+ private void initResourceList() {
+ _resourceList = new HashMap<ExecutorDetails, Map<String, Double>>();
+ // Extract bolt memory info
+ if (this.topology.get_bolts() != null) {
+ for (Map.Entry<String, Bolt> bolt : this.topology.get_bolts().entrySet()) {
+ //the json_conf is populated by TopologyBuilder (e.g. boltDeclarer.setMemoryLoad)
+ Map<String, Double> topology_resources = ResourceUtils.parseResources(bolt
+ .getValue().get_common().get_json_conf());
+ ResourceUtils.checkIntialization(topology_resources, bolt.getValue().toString(), this.topologyConf);
+ for (Map.Entry<ExecutorDetails, String> anExecutorToComponent : executorToComponent.entrySet()) {
+ if (bolt.getKey().equals(anExecutorToComponent.getValue())) {
+ _resourceList.put(anExecutorToComponent.getKey(), topology_resources);
+ }
+ }
+ }
+ }
+ // Extract spout memory info
+ if (this.topology.get_spouts() != null) {
+ for (Map.Entry<String, SpoutSpec> spout : this.topology.get_spouts().entrySet()) {
+ Map<String, Double> topology_resources = ResourceUtils.parseResources(spout
+ .getValue().get_common().get_json_conf());
+ ResourceUtils.checkIntialization(topology_resources, spout.getValue().toString(), this.topologyConf);
+ for (Map.Entry<ExecutorDetails, String> anExecutorToComponent : executorToComponent.entrySet()) {
+ if (spout.getKey().equals(anExecutorToComponent.getValue())) {
+ _resourceList.put(anExecutorToComponent.getKey(), topology_resources);
+ }
+ }
+ }
+ } else {
+ LOG.warn("Topology " + topologyId + " does not seem to have any spouts!");
+ }
+ //schedule tasks that are not part of components returned from topology.get_spout or topology.getbolt (AKA sys tasks most specifically __acker tasks)
+ for(ExecutorDetails exec : this.getExecutors()) {
+ if (_resourceList.containsKey(exec) == false) {
+ LOG.debug(
+ "Scheduling {} {} with memory requirement as 'on heap' - {} and 'off heap' - {} and CPU requirement as {}",
+ this.getExecutorToComponent().get(exec),
+ exec,
+ this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB),
+ this.topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB),
+ this.topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
+ this.addDefaultResforExec(exec);
+ }
+ }
+ }
+
+ private List<ExecutorDetails> componentToExecs(String comp) {
+ List<ExecutorDetails> execs = new ArrayList<>();
+ for (Map.Entry<ExecutorDetails, String> entry : executorToComponent.entrySet()) {
+ if (entry.getValue().equals(comp)) {
+ execs.add(entry.getKey());
+ }
+ }
+ return execs;
+ }
+
+ /**
+ * Returns a representation of the non-system components of the topology graph
+ * Each Component object in the returning map is populated with the list of its
+ * parents, children and execs assigned to that component.
+ * @return a map of components
+ */
+ public Map<String, Component> getComponents() {
+ Map<String, Component> all_comp = new HashMap<String, Component>();
+
+ StormTopology storm_topo = this.topology;
+ // spouts
+ if (storm_topo.get_spouts() != null) {
+ for (Map.Entry<String, SpoutSpec> spoutEntry : storm_topo
+ .get_spouts().entrySet()) {
+ if (!Utils.isSystemId(spoutEntry.getKey())) {
+ Component newComp = null;
+ if (all_comp.containsKey(spoutEntry.getKey())) {
+ newComp = all_comp.get(spoutEntry.getKey());
+ newComp.execs = componentToExecs(newComp.id);
+ } else {
+ newComp = new Component(spoutEntry.getKey());
+ newComp.execs = componentToExecs(newComp.id);
+ all_comp.put(spoutEntry.getKey(), newComp);
+ }
+ newComp.type = Component.ComponentType.SPOUT;
+
+ for (Map.Entry<GlobalStreamId, Grouping> spoutInput : spoutEntry
+ .getValue().get_common().get_inputs()
+ .entrySet()) {
+ newComp.parents.add(spoutInput.getKey()
+ .get_componentId());
+ if (!all_comp.containsKey(spoutInput
+ .getKey().get_componentId())) {
+ all_comp.put(spoutInput.getKey()
+ .get_componentId(),
+ new Component(spoutInput.getKey()
+ .get_componentId()));
+ }
+ all_comp.get(spoutInput.getKey()
+ .get_componentId()).children.add(spoutEntry
+ .getKey());
+ }
+ }
+ }
+ }
+ // bolts
+ if (storm_topo.get_bolts() != null) {
+ for (Map.Entry<String, Bolt> boltEntry : storm_topo.get_bolts()
+ .entrySet()) {
+ if (!Utils.isSystemId(boltEntry.getKey())) {
+ Component newComp = null;
+ if (all_comp.containsKey(boltEntry.getKey())) {
+ newComp = all_comp.get(boltEntry.getKey());
+ newComp.execs = componentToExecs(newComp.id);
+ } else {
+ newComp = new Component(boltEntry.getKey());
+ newComp.execs = componentToExecs(newComp.id);
+ all_comp.put(boltEntry.getKey(), newComp);
+ }
+ newComp.type = Component.ComponentType.BOLT;
+
+ for (Map.Entry<GlobalStreamId, Grouping> boltInput : boltEntry
+ .getValue().get_common().get_inputs()
+ .entrySet()) {
+ newComp.parents.add(boltInput.getKey()
+ .get_componentId());
+ if (!all_comp.containsKey(boltInput
+ .getKey().get_componentId())) {
+ all_comp.put(boltInput.getKey()
+ .get_componentId(),
+ new Component(boltInput.getKey()
+ .get_componentId()));
+ }
+ all_comp.get(boltInput.getKey()
+ .get_componentId()).children.add(boltEntry
+ .getKey());
+ }
+ }
+ }
+ }
+ return all_comp;
+ }
+
+ /**
+ * Gets the on heap memory requirement for a
+ * certain task within a topology
+ * @param exec the executor the inquiry is concerning.
+ * @return Double the amount of on heap memory
+ * requirement for this exec in topology topoId.
+ */
+ public Double getOnHeapMemoryRequirement(ExecutorDetails exec) {
+ Double ret = null;
+ if (hasExecInTopo(exec)) {
+ ret = _resourceList
+ .get(exec)
+ .get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB);
+ }
+ return ret;
+ }
+
+ /**
+ * Gets the off heap memory requirement for a
+ * certain task within a topology
+ * @param exec the executor the inquiry is concerning.
+ * @return Double the amount of off heap memory
+ * requirement for this exec in topology topoId.
+ */
+ public Double getOffHeapMemoryRequirement(ExecutorDetails exec) {
+ Double ret = null;
+ if (hasExecInTopo(exec)) {
+ ret = _resourceList
+ .get(exec)
+ .get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB);
+ }
+ return ret;
+ }
+
+ /**
+ * Gets the total memory requirement for a task
+ * @param exec the executor the inquiry is concerning.
+ * @return Double the total memory requirement
+ * for this exec in topology topoId.
+ */
+ public Double getTotalMemReqTask(ExecutorDetails exec) {
+ if (hasExecInTopo(exec)) {
+ return getOffHeapMemoryRequirement(exec)
+ + getOnHeapMemoryRequirement(exec);
+ }
+ return null;
+ }
+
+ /**
+ * Gets the total memory resource list for a
+ * set of tasks that is part of a topology.
+ * @return Map<ExecutorDetails, Double> a map of the total memory requirement
+ * for all tasks in topology topoId.
+ */
+ public Map<ExecutorDetails, Double> getTotalMemoryResourceList() {
+ Map<ExecutorDetails, Double> ret = new HashMap<ExecutorDetails, Double>();
+ for (ExecutorDetails exec : _resourceList.keySet()) {
+ ret.put(exec, getTotalMemReqTask(exec));
+ }
+ return ret;
+ }
+
+ /**
+ * Get the total CPU requirement for executor
+ * @param exec
+ * @return Double the total about of cpu requirement for executor
+ */
+ public Double getTotalCpuReqTask(ExecutorDetails exec) {
+ if (hasExecInTopo(exec)) {
+ return _resourceList
+ .get(exec)
+ .get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT);
+ }
+ return null;
+ }
+
+ /**
+ * get the resources requirements for a executor
+ * @param exec
+ * @return a map containing the resource requirements for this exec
+ */
+ public Map<String, Double> getTaskResourceReqList(ExecutorDetails exec) {
+ if (hasExecInTopo(exec)) {
+ return _resourceList.get(exec);
+ }
+ return null;
+ }
+
+ /**
+ * Checks if a executor is part of this topology
+ * @return Boolean whether or not a certain ExecutorDetail is included in the _resourceList.
+ */
+ public boolean hasExecInTopo(ExecutorDetails exec) {
+ if (_resourceList != null) { // null is possible if the first constructor of TopologyDetails is used
+ return _resourceList.containsKey(exec);
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * add resource requirements for a executor
+ * @param exec
+ * @param resourceList
+ */
+ public void addResourcesForExec(ExecutorDetails exec, Map<String, Double> resourceList) {
+ if (hasExecInTopo(exec)) {
+ LOG.warn("Executor {} already exists...ResourceList: {}", exec, getTaskResourceReqList(exec));
+ return;
+ }
+ _resourceList.put(exec, resourceList);
+ }
+
+ /**
+ * Add default resource requirements for a executor
+ * @param exec
+ */
+ public void addDefaultResforExec(ExecutorDetails exec) {
+ Map<String, Double> defaultResourceList = new HashMap<String, Double>();
+ defaultResourceList.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT,
+ Utils.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT), null));
+ defaultResourceList.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB,
+ Utils.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB), null));
+ defaultResourceList.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB,
+ Utils.getDouble(topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB), null));
+ LOG.debug("Scheduling Executor: {} with memory requirement as onHeap: {} - offHeap: {} " +
+ "and CPU requirement: {}",
+ exec, topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB),
+ topologyConf.get(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB),
+ topologyConf.get(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT));
+ addResourcesForExec(exec, defaultResourceList);
+ }
+
+ /**
+ * initalizes the scheduler member variable by extracting what scheduler
+ * this topology is going to use from topologyConf
+ */
+ private void initConfigs() {
+ this.topologyWorkerMaxHeapSize = Utils.getDouble(this.topologyConf.get(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB), null);
+ }
+
+ /**
+ * Get the max heap size for a worker used by this topology
+ * @return the worker max heap size
+ */
+ public Double getTopologyWorkerMaxHeapSize() {
+ return this.topologyWorkerMaxHeapSize;
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/121d022b/storm-core/src/jvm/backtype/storm/scheduler/resource/Component.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/scheduler/resource/Component.java b/storm-core/src/jvm/backtype/storm/scheduler/resource/Component.java
new file mode 100644
index 0000000..869b5ce
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/scheduler/resource/Component.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package backtype.storm.scheduler.resource;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import backtype.storm.scheduler.ExecutorDetails;
+
+public class Component {
+ public enum ComponentType {
+ SPOUT(1), BOLT(2);
+ private int value;
+
+ private ComponentType(int value) {
+ this.value = value;
+ }
+ }
+
+ public String id;
+ public List<String> parents = null;
+ public List<String> children = null;
+ public List<ExecutorDetails> execs = null;
+ public ComponentType type = null;
+
+ public Component(String id) {
+ this.parents = new ArrayList<String>();
+ this.children = new ArrayList<String>();
+ this.execs = new ArrayList<ExecutorDetails>();
+ this.id = id;
+ }
+
+ @Override
+ public String toString() {
+ String retVal = "{id: " + this.id + " Parents: " + this.parents.toString() + " Children: " + this.children.toString() + " Execs: " + this.execs + "}";
+ return retVal;
+ }
+}