Merge remote-tracking branch 'ooyala/0.8.0-better-hooks'

Nathan Marz · Nathan Marz · commit 58ce22604e97 · 2012-08-01T14:48:22.000-07:00
diff --git a/src/clj/backtype/storm/daemon/executor.clj b/src/clj/backtype/storm/daemon/executor.clj
@@ -275,22 +275,24 @@
         )))
 
 (defn- fail-spout-msg [executor-data task-data msg-id tuple-info time-delta]
-  (let [^ISpout spout (:object task-data)]
+  (let [^ISpout spout (:object task-data)
+        task-id (:task-id task-data)]
     ;;TODO: need to throttle these when there's lots of failures
     (log-message "Failing message " msg-id ": " tuple-info)
     (.fail spout msg-id)
-    (task/apply-hooks (:user-context task-data) .spoutFail (SpoutFailInfo. msg-id time-delta))
+    (task/apply-hooks (:user-context task-data) .spoutFail (SpoutFailInfo. msg-id task-id time-delta))
     (when time-delta
       (stats/spout-failed-tuple! (:stats executor-data) (:stream tuple-info) time-delta)
       )))
 
 (defn- ack-spout-msg [executor-data task-data msg-id tuple-info time-delta]
   (let [storm-conf (:storm-conf executor-data)
-        ^ISpout spout (:object task-data)]
+        ^ISpout spout (:object task-data)
+        task-id (:task-id task-data)]
     (when (= true (storm-conf TOPOLOGY-DEBUG))
       (log-message "Acking message " msg-id))
     (.ack spout msg-id)
-    (task/apply-hooks (:user-context task-data) .spoutAck (SpoutAckInfo. msg-id time-delta))
+    (task/apply-hooks (:user-context task-data) .spoutAck (SpoutAckInfo. msg-id task-id time-delta))
     (when time-delta
       (stats/spout-acked-tuple! (:stats executor-data) (:stream tuple-info) time-delta)
       )))
@@ -525,7 +527,7 @@
                                                  [root (bit-xor id ack-val)])
                            ))
                        (let [delta (tuple-time-delta! tuple)]
-                         (task/apply-hooks user-context .boltAck (BoltAckInfo. tuple delta))
+                         (task/apply-hooks user-context .boltAck (BoltAckInfo. tuple task-id delta))
                          (when delta
                            (stats/bolt-acked-tuple! executor-stats
                                                     (.getSourceComponent tuple)
@@ -538,7 +540,7 @@
                                                ACKER-FAIL-STREAM-ID
                                                [root]))
                        (let [delta (tuple-time-delta! tuple)]
-                         (task/apply-hooks user-context .boltFail (BoltFailInfo. tuple delta))
+                         (task/apply-hooks user-context .boltFail (BoltFailInfo. tuple task-id delta))
                          (when delta
                            (stats/bolt-failed-tuple! executor-stats
                                                      (.getSourceComponent tuple)
diff --git a/src/clj/backtype/storm/daemon/task.clj b/src/clj/backtype/storm/daemon/task.clj
@@ -103,7 +103,8 @@
         stream->component->grouper (:stream->component->grouper executor-data)
         user-context (:user-context task-data)
         executor-stats (:stats executor-data)
-        debug? (= true (storm-conf TOPOLOGY-DEBUG))]
+        debug? (= true (storm-conf TOPOLOGY-DEBUG))
+        task-id (:task-id task-data)]
     (fn ([^Integer out-task-id ^String stream ^List values]
           (when debug?
             (log-message "Emitting direct: " out-task-id "; " component-id " " stream " " values))
@@ -113,7 +114,7 @@
                 out-task-id (if grouping out-task-id)]
             (when (and (not-nil? grouping) (not= :direct grouping))
               (throw (IllegalArgumentException. "Cannot emitDirect to a task expecting a regular grouping")))                          
-            (apply-hooks user-context .emit (EmitInfo. values stream [out-task-id]))
+            (apply-hooks user-context .emit (EmitInfo. values stream task-id [out-task-id]))
             (when (emit-sampler)
               (stats/emitted-tuple! executor-stats stream)
               (if out-task-id
@@ -133,7 +134,7 @@
                    (.addAll out-tasks comp-tasks)
                    (.add out-tasks comp-tasks)
                    )))
-             (apply-hooks user-context .emit (EmitInfo. values stream out-tasks))
+             (apply-hooks user-context .emit (EmitInfo. values stream task-id out-tasks))
              (when (emit-sampler)
                (stats/emitted-tuple! executor-stats stream)
                (stats/transferred-tuples! executor-stats stream (count out-tasks)))
diff --git a/src/jvm/backtype/storm/hooks/info/BoltAckInfo.java b/src/jvm/backtype/storm/hooks/info/BoltAckInfo.java
@@ -4,10 +4,12 @@
 
 public class BoltAckInfo {
     public Tuple tuple;
+    public int ackingTaskId;
     public Long processLatencyMs; // null if it wasn't sampled
     
-    public BoltAckInfo(Tuple tuple, Long processLatencyMs) {
+    public BoltAckInfo(Tuple tuple, int ackingTaskId, Long processLatencyMs) {
         this.tuple = tuple;
+        this.ackingTaskId = ackingTaskId;
         this.processLatencyMs = processLatencyMs;
     }
 }
diff --git a/src/jvm/backtype/storm/hooks/info/BoltFailInfo.java b/src/jvm/backtype/storm/hooks/info/BoltFailInfo.java
@@ -4,10 +4,12 @@
 
 public class BoltFailInfo {
     public Tuple tuple;
+    public int failingTaskId;
     public Long failLatencyMs; // null if it wasn't sampled
     
-    public BoltFailInfo(Tuple tuple, Long failLatencyMs) {
+    public BoltFailInfo(Tuple tuple, int failingTaskId, Long failLatencyMs) {
         this.tuple = tuple;
+        this.failingTaskId = failingTaskId;
         this.failLatencyMs = failLatencyMs;
     }
 }
diff --git a/src/jvm/backtype/storm/hooks/info/EmitInfo.java b/src/jvm/backtype/storm/hooks/info/EmitInfo.java
@@ -6,11 +6,13 @@
 public class EmitInfo {
     public List<Object> values;
     public String stream;
+    public int taskId;
     public Collection<Integer> outTasks;
     
-    public EmitInfo(List<Object> values, String stream, Collection<Integer> outTasks) {
+    public EmitInfo(List<Object> values, String stream, int taskId, Collection<Integer> outTasks) {
         this.values = values;
         this.stream = stream;
+        this.taskId = taskId;
         this.outTasks = outTasks;
     }
 }
diff --git a/src/jvm/backtype/storm/hooks/info/SpoutAckInfo.java b/src/jvm/backtype/storm/hooks/info/SpoutAckInfo.java
@@ -2,10 +2,12 @@
 
 public class SpoutAckInfo {
     public Object messageId;
+    public int spoutTaskId;
     public Long completeLatencyMs; // null if it wasn't sampled
     
-    public SpoutAckInfo(Object messageId, Long completeLatencyMs) {
+    public SpoutAckInfo(Object messageId, int spoutTaskId, Long completeLatencyMs) {
         this.messageId = messageId;
+        this.spoutTaskId = spoutTaskId;
         this.completeLatencyMs = completeLatencyMs;
     }
 }
diff --git a/src/jvm/backtype/storm/hooks/info/SpoutFailInfo.java b/src/jvm/backtype/storm/hooks/info/SpoutFailInfo.java
@@ -2,10 +2,12 @@
 
 public class SpoutFailInfo {
     public Object messageId;
+    public int spoutTaskId;
     public Long failLatencyMs; // null if it wasn't sampled
     
-    public SpoutFailInfo(Object messageId, Long failLatencyMs) {
+    public SpoutFailInfo(Object messageId, int spoutTaskId, Long failLatencyMs) {
         this.messageId = messageId;
+        this.spoutTaskId = spoutTaskId;
         this.failLatencyMs = failLatencyMs;
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -4,10 +4,12 @@`
`4`	`4`
`5`	`5`	`public class BoltAckInfo {`
`6`	`6`	`public Tuple tuple;`
	`7`	`+ public int ackingTaskId;`
`7`	`8`	`public Long processLatencyMs; // null if it wasn't sampled`
`8`	`9`
`9`		`- public BoltAckInfo(Tuple tuple, Long processLatencyMs) {`
	`10`	`+ public BoltAckInfo(Tuple tuple, int ackingTaskId, Long processLatencyMs) {`
`10`	`11`	`this.tuple = tuple;`
	`12`	`+ this.ackingTaskId = ackingTaskId;`
`11`	`13`	`this.processLatencyMs = processLatencyMs;`
`12`	`14`	`}`
`13`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -4,10 +4,12 @@`
`4`	`4`
`5`	`5`	`public class BoltFailInfo {`
`6`	`6`	`public Tuple tuple;`
	`7`	`+ public int failingTaskId;`
`7`	`8`	`public Long failLatencyMs; // null if it wasn't sampled`
`8`	`9`
`9`		`- public BoltFailInfo(Tuple tuple, Long failLatencyMs) {`
	`10`	`+ public BoltFailInfo(Tuple tuple, int failingTaskId, Long failLatencyMs) {`
`10`	`11`	`this.tuple = tuple;`
	`12`	`+ this.failingTaskId = failingTaskId;`
`11`	`13`	`this.failLatencyMs = failLatencyMs;`
`12`	`14`	`}`
`13`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,11 +6,13 @@`
`6`	`6`	`public class EmitInfo {`
`7`	`7`	`public List<Object> values;`
`8`	`8`	`public String stream;`
	`9`	`+ public int taskId;`
`9`	`10`	`public Collection<Integer> outTasks;`
`10`	`11`
`11`		`- public EmitInfo(List<Object> values, String stream, Collection<Integer> outTasks) {`
	`12`	`+ public EmitInfo(List<Object> values, String stream, int taskId, Collection<Integer> outTasks) {`
`12`	`13`	`this.values = values;`
`13`	`14`	`this.stream = stream;`
	`15`	`+ this.taskId = taskId;`
`14`	`16`	`this.outTasks = outTasks;`
`15`	`17`	`}`
`16`	`18`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,10 +2,12 @@`
`2`	`2`
`3`	`3`	`public class SpoutAckInfo {`
`4`	`4`	`public Object messageId;`
	`5`	`+ public int spoutTaskId;`
`5`	`6`	`public Long completeLatencyMs; // null if it wasn't sampled`
`6`	`7`
`7`		`- public SpoutAckInfo(Object messageId, Long completeLatencyMs) {`
	`8`	`+ public SpoutAckInfo(Object messageId, int spoutTaskId, Long completeLatencyMs) {`
`8`	`9`	`this.messageId = messageId;`
	`10`	`+ this.spoutTaskId = spoutTaskId;`
`9`	`11`	`this.completeLatencyMs = completeLatencyMs;`
`10`	`12`	`}`
`11`	`13`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,10 +2,12 @@`
`2`	`2`
`3`	`3`	`public class SpoutFailInfo {`
`4`	`4`	`public Object messageId;`
	`5`	`+ public int spoutTaskId;`
`5`	`6`	`public Long failLatencyMs; // null if it wasn't sampled`
`6`	`7`
`7`		`- public SpoutFailInfo(Object messageId, Long failLatencyMs) {`
	`8`	`+ public SpoutFailInfo(Object messageId, int spoutTaskId, Long failLatencyMs) {`
`8`	`9`	`this.messageId = messageId;`
	`10`	`+ this.spoutTaskId = spoutTaskId;`
`9`	`11`	`this.failLatencyMs = failLatencyMs;`
`10`	`12`	`}`
`11`	`13`	`}`