You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2015/12/01 19:08:52 UTC
[08/50] [abbrv] storm git commit: add support for worker lifecycle
hooks
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/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 4813fde..c943cac 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyPageInfo.java
@@ -2041,16 +2041,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 _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)
+ org.apache.thrift.protocol.TMap _map402 = iprot.readMapBegin();
+ struct.id_to_spout_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map402.size);
+ String _key403;
+ ComponentAggregateStats _val404;
+ for (int _i405 = 0; _i405 < _map402.size; ++_i405)
{
- _key395 = iprot.readString();
- _val396 = new ComponentAggregateStats();
- _val396.read(iprot);
- struct.id_to_spout_agg_stats.put(_key395, _val396);
+ _key403 = iprot.readString();
+ _val404 = new ComponentAggregateStats();
+ _val404.read(iprot);
+ struct.id_to_spout_agg_stats.put(_key403, _val404);
}
iprot.readMapEnd();
}
@@ -2062,16 +2062,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 _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)
+ org.apache.thrift.protocol.TMap _map406 = iprot.readMapBegin();
+ struct.id_to_bolt_agg_stats = new HashMap<String,ComponentAggregateStats>(2*_map406.size);
+ String _key407;
+ ComponentAggregateStats _val408;
+ for (int _i409 = 0; _i409 < _map406.size; ++_i409)
{
- _key399 = iprot.readString();
- _val400 = new ComponentAggregateStats();
- _val400.read(iprot);
- struct.id_to_bolt_agg_stats.put(_key399, _val400);
+ _key407 = iprot.readString();
+ _val408 = new ComponentAggregateStats();
+ _val408.read(iprot);
+ struct.id_to_bolt_agg_stats.put(_key407, _val408);
}
iprot.readMapEnd();
}
@@ -2234,10 +2234,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> _iter402 : struct.id_to_spout_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter410 : struct.id_to_spout_agg_stats.entrySet())
{
- oprot.writeString(_iter402.getKey());
- _iter402.getValue().write(oprot);
+ oprot.writeString(_iter410.getKey());
+ _iter410.getValue().write(oprot);
}
oprot.writeMapEnd();
}
@@ -2249,10 +2249,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> _iter403 : struct.id_to_bolt_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter411 : struct.id_to_bolt_agg_stats.entrySet())
{
- oprot.writeString(_iter403.getKey());
- _iter403.getValue().write(oprot);
+ oprot.writeString(_iter411.getKey());
+ _iter411.getValue().write(oprot);
}
oprot.writeMapEnd();
}
@@ -2426,20 +2426,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> _iter404 : struct.id_to_spout_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter412 : struct.id_to_spout_agg_stats.entrySet())
{
- oprot.writeString(_iter404.getKey());
- _iter404.getValue().write(oprot);
+ oprot.writeString(_iter412.getKey());
+ _iter412.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> _iter405 : struct.id_to_bolt_agg_stats.entrySet())
+ for (Map.Entry<String, ComponentAggregateStats> _iter413 : struct.id_to_bolt_agg_stats.entrySet())
{
- oprot.writeString(_iter405.getKey());
- _iter405.getValue().write(oprot);
+ oprot.writeString(_iter413.getKey());
+ _iter413.getValue().write(oprot);
}
}
}
@@ -2514,32 +2514,32 @@ public class TopologyPageInfo implements org.apache.thrift.TBase<TopologyPageInf
}
if (incoming.get(7)) {
{
- 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)
+ org.apache.thrift.protocol.TMap _map414 = 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*_map414.size);
+ String _key415;
+ ComponentAggregateStats _val416;
+ for (int _i417 = 0; _i417 < _map414.size; ++_i417)
{
- _key407 = iprot.readString();
- _val408 = new ComponentAggregateStats();
- _val408.read(iprot);
- struct.id_to_spout_agg_stats.put(_key407, _val408);
+ _key415 = iprot.readString();
+ _val416 = new ComponentAggregateStats();
+ _val416.read(iprot);
+ struct.id_to_spout_agg_stats.put(_key415, _val416);
}
}
struct.set_id_to_spout_agg_stats_isSet(true);
}
if (incoming.get(8)) {
{
- 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)
+ org.apache.thrift.protocol.TMap _map418 = 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*_map418.size);
+ String _key419;
+ ComponentAggregateStats _val420;
+ for (int _i421 = 0; _i421 < _map418.size; ++_i421)
{
- _key411 = iprot.readString();
- _val412 = new ComponentAggregateStats();
- _val412.read(iprot);
- struct.id_to_bolt_agg_stats.put(_key411, _val412);
+ _key419 = iprot.readString();
+ _val420 = new ComponentAggregateStats();
+ _val420.read(iprot);
+ struct.id_to_bolt_agg_stats.put(_key419, _val420);
}
}
struct.set_id_to_bolt_agg_stats_isSet(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/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 99f3922..aa598e4 100644
--- a/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
+++ b/storm-core/src/jvm/backtype/storm/generated/TopologyStats.java
@@ -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 _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)
+ org.apache.thrift.protocol.TMap _map352 = iprot.readMapBegin();
+ struct.window_to_emitted = new HashMap<String,Long>(2*_map352.size);
+ String _key353;
+ long _val354;
+ for (int _i355 = 0; _i355 < _map352.size; ++_i355)
{
- _key345 = iprot.readString();
- _val346 = iprot.readI64();
- struct.window_to_emitted.put(_key345, _val346);
+ _key353 = iprot.readString();
+ _val354 = iprot.readI64();
+ struct.window_to_emitted.put(_key353, _val354);
}
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 _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)
+ org.apache.thrift.protocol.TMap _map356 = iprot.readMapBegin();
+ struct.window_to_transferred = new HashMap<String,Long>(2*_map356.size);
+ String _key357;
+ long _val358;
+ for (int _i359 = 0; _i359 < _map356.size; ++_i359)
{
- _key349 = iprot.readString();
- _val350 = iprot.readI64();
- struct.window_to_transferred.put(_key349, _val350);
+ _key357 = iprot.readString();
+ _val358 = iprot.readI64();
+ struct.window_to_transferred.put(_key357, _val358);
}
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 _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)
+ org.apache.thrift.protocol.TMap _map360 = iprot.readMapBegin();
+ struct.window_to_complete_latencies_ms = new HashMap<String,Double>(2*_map360.size);
+ String _key361;
+ double _val362;
+ for (int _i363 = 0; _i363 < _map360.size; ++_i363)
{
- _key353 = iprot.readString();
- _val354 = iprot.readDouble();
- struct.window_to_complete_latencies_ms.put(_key353, _val354);
+ _key361 = iprot.readString();
+ _val362 = iprot.readDouble();
+ struct.window_to_complete_latencies_ms.put(_key361, _val362);
}
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 _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)
+ org.apache.thrift.protocol.TMap _map364 = iprot.readMapBegin();
+ struct.window_to_acked = new HashMap<String,Long>(2*_map364.size);
+ String _key365;
+ long _val366;
+ for (int _i367 = 0; _i367 < _map364.size; ++_i367)
{
- _key357 = iprot.readString();
- _val358 = iprot.readI64();
- struct.window_to_acked.put(_key357, _val358);
+ _key365 = iprot.readString();
+ _val366 = iprot.readI64();
+ struct.window_to_acked.put(_key365, _val366);
}
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 _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)
+ org.apache.thrift.protocol.TMap _map368 = iprot.readMapBegin();
+ struct.window_to_failed = new HashMap<String,Long>(2*_map368.size);
+ String _key369;
+ long _val370;
+ for (int _i371 = 0; _i371 < _map368.size; ++_i371)
{
- _key361 = iprot.readString();
- _val362 = iprot.readI64();
- struct.window_to_failed.put(_key361, _val362);
+ _key369 = iprot.readString();
+ _val370 = iprot.readI64();
+ struct.window_to_failed.put(_key369, _val370);
}
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> _iter364 : struct.window_to_emitted.entrySet())
+ for (Map.Entry<String, Long> _iter372 : struct.window_to_emitted.entrySet())
{
- oprot.writeString(_iter364.getKey());
- oprot.writeI64(_iter364.getValue());
+ oprot.writeString(_iter372.getKey());
+ oprot.writeI64(_iter372.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> _iter365 : struct.window_to_transferred.entrySet())
+ for (Map.Entry<String, Long> _iter373 : struct.window_to_transferred.entrySet())
{
- oprot.writeString(_iter365.getKey());
- oprot.writeI64(_iter365.getValue());
+ oprot.writeString(_iter373.getKey());
+ oprot.writeI64(_iter373.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> _iter366 : struct.window_to_complete_latencies_ms.entrySet())
+ for (Map.Entry<String, Double> _iter374 : struct.window_to_complete_latencies_ms.entrySet())
{
- oprot.writeString(_iter366.getKey());
- oprot.writeDouble(_iter366.getValue());
+ oprot.writeString(_iter374.getKey());
+ oprot.writeDouble(_iter374.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> _iter367 : struct.window_to_acked.entrySet())
+ for (Map.Entry<String, Long> _iter375 : struct.window_to_acked.entrySet())
{
- oprot.writeString(_iter367.getKey());
- oprot.writeI64(_iter367.getValue());
+ oprot.writeString(_iter375.getKey());
+ oprot.writeI64(_iter375.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> _iter368 : struct.window_to_failed.entrySet())
+ for (Map.Entry<String, Long> _iter376 : struct.window_to_failed.entrySet())
{
- oprot.writeString(_iter368.getKey());
- oprot.writeI64(_iter368.getValue());
+ oprot.writeString(_iter376.getKey());
+ oprot.writeI64(_iter376.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> _iter369 : struct.window_to_emitted.entrySet())
+ for (Map.Entry<String, Long> _iter377 : struct.window_to_emitted.entrySet())
{
- oprot.writeString(_iter369.getKey());
- oprot.writeI64(_iter369.getValue());
+ oprot.writeString(_iter377.getKey());
+ oprot.writeI64(_iter377.getValue());
}
}
}
if (struct.is_set_window_to_transferred()) {
{
oprot.writeI32(struct.window_to_transferred.size());
- for (Map.Entry<String, Long> _iter370 : struct.window_to_transferred.entrySet())
+ for (Map.Entry<String, Long> _iter378 : struct.window_to_transferred.entrySet())
{
- oprot.writeString(_iter370.getKey());
- oprot.writeI64(_iter370.getValue());
+ oprot.writeString(_iter378.getKey());
+ oprot.writeI64(_iter378.getValue());
}
}
}
if (struct.is_set_window_to_complete_latencies_ms()) {
{
oprot.writeI32(struct.window_to_complete_latencies_ms.size());
- for (Map.Entry<String, Double> _iter371 : struct.window_to_complete_latencies_ms.entrySet())
+ for (Map.Entry<String, Double> _iter379 : struct.window_to_complete_latencies_ms.entrySet())
{
- oprot.writeString(_iter371.getKey());
- oprot.writeDouble(_iter371.getValue());
+ oprot.writeString(_iter379.getKey());
+ oprot.writeDouble(_iter379.getValue());
}
}
}
if (struct.is_set_window_to_acked()) {
{
oprot.writeI32(struct.window_to_acked.size());
- for (Map.Entry<String, Long> _iter372 : struct.window_to_acked.entrySet())
+ for (Map.Entry<String, Long> _iter380 : struct.window_to_acked.entrySet())
{
- oprot.writeString(_iter372.getKey());
- oprot.writeI64(_iter372.getValue());
+ oprot.writeString(_iter380.getKey());
+ oprot.writeI64(_iter380.getValue());
}
}
}
if (struct.is_set_window_to_failed()) {
{
oprot.writeI32(struct.window_to_failed.size());
- for (Map.Entry<String, Long> _iter373 : struct.window_to_failed.entrySet())
+ for (Map.Entry<String, Long> _iter381 : struct.window_to_failed.entrySet())
{
- oprot.writeString(_iter373.getKey());
- oprot.writeI64(_iter373.getValue());
+ oprot.writeString(_iter381.getKey());
+ oprot.writeI64(_iter381.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 _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)
+ org.apache.thrift.protocol.TMap _map382 = 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*_map382.size);
+ String _key383;
+ long _val384;
+ for (int _i385 = 0; _i385 < _map382.size; ++_i385)
{
- _key375 = iprot.readString();
- _val376 = iprot.readI64();
- struct.window_to_emitted.put(_key375, _val376);
+ _key383 = iprot.readString();
+ _val384 = iprot.readI64();
+ struct.window_to_emitted.put(_key383, _val384);
}
}
struct.set_window_to_emitted_isSet(true);
}
if (incoming.get(1)) {
{
- 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)
+ 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_transferred = new HashMap<String,Long>(2*_map386.size);
+ String _key387;
+ long _val388;
+ for (int _i389 = 0; _i389 < _map386.size; ++_i389)
{
- _key379 = iprot.readString();
- _val380 = iprot.readI64();
- struct.window_to_transferred.put(_key379, _val380);
+ _key387 = iprot.readString();
+ _val388 = iprot.readI64();
+ struct.window_to_transferred.put(_key387, _val388);
}
}
struct.set_window_to_transferred_isSet(true);
}
if (incoming.get(2)) {
{
- 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)
+ org.apache.thrift.protocol.TMap _map390 = 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*_map390.size);
+ String _key391;
+ double _val392;
+ for (int _i393 = 0; _i393 < _map390.size; ++_i393)
{
- _key383 = iprot.readString();
- _val384 = iprot.readDouble();
- struct.window_to_complete_latencies_ms.put(_key383, _val384);
+ _key391 = iprot.readString();
+ _val392 = iprot.readDouble();
+ struct.window_to_complete_latencies_ms.put(_key391, _val392);
}
}
struct.set_window_to_complete_latencies_ms_isSet(true);
}
if (incoming.get(3)) {
{
- 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)
+ org.apache.thrift.protocol.TMap _map394 = 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*_map394.size);
+ String _key395;
+ long _val396;
+ for (int _i397 = 0; _i397 < _map394.size; ++_i397)
{
- _key387 = iprot.readString();
- _val388 = iprot.readI64();
- struct.window_to_acked.put(_key387, _val388);
+ _key395 = iprot.readString();
+ _val396 = iprot.readI64();
+ struct.window_to_acked.put(_key395, _val396);
}
}
struct.set_window_to_acked_isSet(true);
}
if (incoming.get(4)) {
{
- 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)
+ org.apache.thrift.protocol.TMap _map398 = 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*_map398.size);
+ String _key399;
+ long _val400;
+ for (int _i401 = 0; _i401 < _map398.size; ++_i401)
{
- _key391 = iprot.readString();
- _val392 = iprot.readI64();
- struct.window_to_failed.put(_key391, _val392);
+ _key399 = iprot.readString();
+ _val400 = iprot.readI64();
+ struct.window_to_failed.put(_key399, _val400);
}
}
struct.set_window_to_failed_isSet(true);
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/storm-core/src/jvm/backtype/storm/hooks/BaseWorkerHook.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/hooks/BaseWorkerHook.java b/storm-core/src/jvm/backtype/storm/hooks/BaseWorkerHook.java
new file mode 100644
index 0000000..6fe9f19
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/hooks/BaseWorkerHook.java
@@ -0,0 +1,34 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hooks;
+
+import backtype.storm.task.WorkerTopologyContext;
+
+import java.util.List;
+import java.util.Map;
+
+public class BaseWorkerHook implements IWorkerHook {
+ @Override
+ public void start(Map stormConf, WorkerTopologyContext context, List taskIds) {
+
+ }
+
+ @Override
+ public void shutdown() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/storm-core/src/jvm/backtype/storm/hooks/IWorkerHook.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/hooks/IWorkerHook.java b/storm-core/src/jvm/backtype/storm/hooks/IWorkerHook.java
new file mode 100644
index 0000000..6c2bab2
--- /dev/null
+++ b/storm-core/src/jvm/backtype/storm/hooks/IWorkerHook.java
@@ -0,0 +1,29 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.hooks;
+
+import backtype.storm.task.WorkerTopologyContext;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+public interface IWorkerHook extends Serializable {
+ void start(Map stormConf, WorkerTopologyContext context, List taskIds);
+ void shutdown();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java b/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java
index 9536faa..965540e 100644
--- a/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java
+++ b/storm-core/src/jvm/backtype/storm/topology/TopologyBuilder.java
@@ -18,25 +18,20 @@
package backtype.storm.topology;
import backtype.storm.Config;
-import backtype.storm.generated.Bolt;
-import backtype.storm.generated.ComponentCommon;
-import backtype.storm.generated.ComponentObject;
-import backtype.storm.generated.GlobalStreamId;
-import backtype.storm.generated.Grouping;
-import backtype.storm.generated.NullStruct;
-import backtype.storm.generated.SpoutSpec;
-import backtype.storm.generated.StateSpoutSpec;
-import backtype.storm.generated.StormTopology;
+import backtype.storm.generated.*;
import backtype.storm.grouping.CustomStreamGrouping;
import backtype.storm.grouping.PartialKeyGrouping;
+import backtype.storm.hooks.IWorkerHook;
import backtype.storm.tuple.Fields;
import backtype.storm.utils.Utils;
+import org.json.simple.JSONValue;
+
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-
import backtype.storm.windowing.TupleWindow;
-import org.json.simple.JSONValue;
/**
* TopologyBuilder exposes the Java API for specifying a topology for Storm
@@ -98,11 +93,13 @@ public class TopologyBuilder {
// private Map<String, Map<GlobalStreamId, Grouping>> _inputs = new HashMap<String, Map<GlobalStreamId, Grouping>>();
private Map<String, StateSpoutSpec> _stateSpouts = new HashMap<>();
-
-
+ private List<ByteBuffer> _workerHooks = new ArrayList<>();
+
+
public StormTopology createTopology() {
Map<String, Bolt> boltSpecs = new HashMap<>();
Map<String, SpoutSpec> spoutSpecs = new HashMap<>();
+
for(String boltId: _bolts.keySet()) {
IRichBolt bolt = _bolts.get(boltId);
ComponentCommon common = getComponentCommon(boltId, bolt);
@@ -112,11 +109,15 @@ public class TopologyBuilder {
IRichSpout spout = _spouts.get(spoutId);
ComponentCommon common = getComponentCommon(spoutId, spout);
spoutSpecs.put(spoutId, new SpoutSpec(ComponentObject.serialized_java(Utils.javaSerialize(spout)), common));
-
}
- return new StormTopology(spoutSpecs,
- boltSpecs,
- new HashMap<String, StateSpoutSpec>());
+
+ StormTopology stormTopology = new StormTopology(spoutSpecs,
+ boltSpecs,
+ new HashMap<String, StateSpoutSpec>());
+
+ stormTopology.set_worker_hooks(_workerHooks);
+
+ return stormTopology;
}
/**
@@ -230,6 +231,14 @@ public class TopologyBuilder {
// TODO: finish
}
+ /**
+ * Add a new worker lifecycle hook
+ *
+ * @param workerHook the lifecycle hook to add
+ */
+ public void addWorkerHook(IWorkerHook workerHook) {
+ _workerHooks.add(ByteBuffer.wrap(Utils.javaSerialize(workerHook)));
+ }
private void validateUnusedId(String id) {
if(_bolts.containsKey(id)) {
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/storm-core/src/jvm/backtype/storm/utils/ThriftTopologyUtils.java
----------------------------------------------------------------------
diff --git a/storm-core/src/jvm/backtype/storm/utils/ThriftTopologyUtils.java b/storm-core/src/jvm/backtype/storm/utils/ThriftTopologyUtils.java
index 8306d9b..d5c460f 100644
--- a/storm-core/src/jvm/backtype/storm/utils/ThriftTopologyUtils.java
+++ b/storm-core/src/jvm/backtype/storm/utils/ThriftTopologyUtils.java
@@ -27,30 +27,38 @@ import java.util.Map;
import java.util.Set;
public class ThriftTopologyUtils {
+ public static boolean isWorkerHook(StormTopology._Fields f) {
+ return f.equals(StormTopology._Fields.WORKER_HOOKS);
+ }
+
public static Set<String> getComponentIds(StormTopology topology) {
Set<String> ret = new HashSet<String>();
for(StormTopology._Fields f: StormTopology.metaDataMap.keySet()) {
- Map<String, Object> componentMap = (Map<String, Object>) topology.getFieldValue(f);
- ret.addAll(componentMap.keySet());
+ if(StormTopology.metaDataMap.get(f).valueMetaData.type == org.apache.thrift.protocol.TType.MAP) {
+ Map<String, Object> componentMap = (Map<String, Object>) topology.getFieldValue(f);
+ ret.addAll(componentMap.keySet());
+ }
}
return ret;
}
public static ComponentCommon getComponentCommon(StormTopology topology, String componentId) {
for(StormTopology._Fields f: StormTopology.metaDataMap.keySet()) {
- Map<String, Object> componentMap = (Map<String, Object>) topology.getFieldValue(f);
- if(componentMap.containsKey(componentId)) {
- Object component = componentMap.get(componentId);
- if(component instanceof Bolt) {
- return ((Bolt) component).get_common();
- }
- if(component instanceof SpoutSpec) {
- return ((SpoutSpec) component).get_common();
- }
- if(component instanceof StateSpoutSpec) {
- return ((StateSpoutSpec) component).get_common();
+ if(StormTopology.metaDataMap.get(f).valueMetaData.type == org.apache.thrift.protocol.TType.MAP) {
+ Map<String, Object> componentMap = (Map<String, Object>) topology.getFieldValue(f);
+ if(componentMap.containsKey(componentId)) {
+ Object component = componentMap.get(componentId);
+ if(component instanceof Bolt) {
+ return ((Bolt) component).get_common();
+ }
+ if(component instanceof SpoutSpec) {
+ return ((SpoutSpec) component).get_common();
+ }
+ if(component instanceof StateSpoutSpec) {
+ return ((StateSpoutSpec) component).get_common();
+ }
+ throw new RuntimeException("Unreachable code! No get_common conversion for component " + component);
}
- throw new RuntimeException("Unreachable code! No get_common conversion for component " + component);
}
}
throw new IllegalArgumentException("Could not find component common for " + componentId);
http://git-wip-us.apache.org/repos/asf/storm/blob/b03ce6b2/storm-core/src/py/storm/Nimbus.py
----------------------------------------------------------------------
diff --git a/storm-core/src/py/storm/Nimbus.py b/storm-core/src/py/storm/Nimbus.py
index c1e1b02..c2bb9ac 100644
--- a/storm-core/src/py/storm/Nimbus.py
+++ b/storm-core/src/py/storm/Nimbus.py
@@ -3811,11 +3811,11 @@ class getComponentPendingProfileActions_result:
if fid == 0:
if ftype == TType.LIST:
self.success = []
- (_etype641, _size638) = iprot.readListBegin()
- for _i642 in xrange(_size638):
- _elem643 = ProfileRequest()
- _elem643.read(iprot)
- self.success.append(_elem643)
+ (_etype648, _size645) = iprot.readListBegin()
+ for _i649 in xrange(_size645):
+ _elem650 = ProfileRequest()
+ _elem650.read(iprot)
+ self.success.append(_elem650)
iprot.readListEnd()
else:
iprot.skip(ftype)
@@ -3832,8 +3832,8 @@ class getComponentPendingProfileActions_result:
if self.success is not None:
oprot.writeFieldBegin('success', TType.LIST, 0)
oprot.writeListBegin(TType.STRUCT, len(self.success))
- for iter644 in self.success:
- iter644.write(oprot)
+ for iter651 in self.success:
+ iter651.write(oprot)
oprot.writeListEnd()
oprot.writeFieldEnd()
oprot.writeFieldStop()