storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bo...@apache.org
Subject [05/16] storm git commit: add support for worker lifecycle hooks
Date Tue, 24 Nov 2015 17:57:12 GMT
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()


Mime
View raw message