diff --git a/README.markdown b/README.markdown
index 5da35b0..9e3be66 100644
--- a/README.markdown
+++ b/README.markdown
@@ -10,8 +10,33 @@
 
 ## Getting help
 
-Feel free to ask questions on Storm's mailing list: http://groups.google.com/group/storm-user
+__NOTE:__ The google groups account storm-user@googlegroups.com is now officially deprecated in favor of the Apache-hosted user/dev mailing lists.
 
+### Storm Users
+Storm users should send messages and subscribe to [user@storm.incubator.apache.org](mailto:user@storm.incubator.apache.org).
+
+You can subscribe to this list by sending an email to [user-subscribe@storm.incubator.apache.org](mailto:user-subscribe@storm.incubator.apache.org). Likewise, you can cancel a subscription by sending an email to [user-unsubscribe@storm.incubator.apache.org](mailto:user-unsubscribe@storm.incubator.apache.org).
+
+You can view the archives of the mailing list [here](http://mail-archives.apache.org/mod_mbox/incubator-storm-user/).
+
+### Storm Developers
+Storm developers should send messages and subscribe to [dev@storm.incubator.apache.org](mailto:dev@storm.incubator.apache.org).
+
+You can subscribe to this list by sending an email to [dev-subscribe@storm.incubator.apache.org](mailto:dev-subscribe@storm.incubator.apache.org). Likewise, you can cancel a subscription by sending an email to [dev-unsubscribe@storm.incubator.apache.org](mailto:dev-unsubscribe@storm.incubator.apache.org).
+
+You can view the archives of the mailing list [here](http://mail-archives.apache.org/mod_mbox/incubator-storm-dev/).
+
+### Which list should I send/subscribe to?
+If you are using a pre-built binary distribution of Storm, then chances are you should send questions, comments, storm-related announcements, etc. to [user@storm.apache.incubator.org](user@storm.apache.incubator.org). 
+
+If you are building storm from source, developing new features, or otherwise hacking storm source code, then [dev@storm.incubator.apache.org](dev@storm.incubator.apache.org) is more appropriate. 
+
+### What will happen with storm-user@googlegroups.com?
+All existing messages will remain archived there, and can be accessed/searched [here](https://groups.google.com/forum/#!forum/storm-user).
+
+New messages sent to storm-user@googlegroups.com will either be rejected/bounced or replied to with a message to direct the email to the appropriate Apache-hosted group.
+
+### IRC
 You can also come to the #storm-user room on [freenode](http://freenode.net/). You can usually find a Storm developer there to help you out.
 
 ## License
@@ -27,11 +52,14 @@
 
 * Nathan Marz ([@nathanmarz](http://twitter.com/nathanmarz))
 
-## Core contributors
+## Committers
 
 * James Xu ([@xumingming](https://github.com/xumingming))
 * Jason Jackson ([@jason_j](http://twitter.com/jason_j))
 * Andy Feng ([@anfeng](https://github.com/anfeng))
+* Flip Kromer ([@mrflip](https://github.com/mrflip))
+* David Lao ([@davidlao2k](https://github.com/davidlao2k))
+* P. Taylor Goetz ([@ptgoetz](https://github.com/ptgoetz))
 
 ## Contributors
 
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index ede4034..08c7889 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -40,6 +40,7 @@
 
 logviewer.port: 8000
 logviewer.childopts: "-Xmx128m"
+logviewer.appender.name: "A1"
 
 
 drpc.port: 3772
diff --git a/storm-core/src/clj/backtype/storm/daemon/executor.clj b/storm-core/src/clj/backtype/storm/daemon/executor.clj
index 431bf35..40f5561 100644
--- a/storm-core/src/clj/backtype/storm/daemon/executor.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/executor.clj
@@ -283,9 +283,10 @@
         receive-queue (:receive-queue executor-data)
         context (:worker-context executor-data)]
     (when tick-time-secs
-      (if (and (not (storm-conf TOPOLOGY-ENABLE-MESSAGE-TIMEOUTS))
-               (= :spout (:type executor-data)))
-        (log-message "Timeouts disabled for executor " (:executor-id executor-data))
+      (if (or (system-id? (:component-id executor-data))
+              (and (not (storm-conf TOPOLOGY-ENABLE-MESSAGE-TIMEOUTS))
+                   (= :spout (:type executor-data))))
+        (log-message "Timeouts disabled for executor " (:component-id executor-data) ":" (:executor-id executor-data))
         (schedule-recurring
           (:user-timer worker)
           tick-time-secs
diff --git a/storm-core/src/clj/backtype/storm/daemon/logviewer.clj b/storm-core/src/clj/backtype/storm/daemon/logviewer.clj
index 3cc8b39..c116937 100644
--- a/storm-core/src/clj/backtype/storm/daemon/logviewer.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/logviewer.clj
@@ -3,11 +3,16 @@
   (:use [hiccup core page-helpers])
   (:use [backtype.storm config util log])
   (:use [ring.adapter.jetty :only [run-jetty]])
+  (:import [org.slf4j LoggerFactory])
+  (:import [ch.qos.logback.classic Logger])
   (:import [org.apache.commons.logging LogFactory])
   (:import [org.apache.commons.logging.impl Log4JLogger])
+  (:import [ch.qos.logback.core FileAppender])
   (:import [org.apache.log4j Level])
+  (:import [java.io File])
   (:require [compojure.route :as route]
-            [compojure.handler :as handler])
+            [compojure.handler :as handler]
+            [clojure.string :as string])
   (:gen-class))
 
 (defn tail-file [path tail]
@@ -25,8 +30,19 @@
       (.toString output))
     ))
 
-(defn log-page [file tail grep]
-  (let [path (str (System/getProperty "storm.home") "/logs/" file)
+(defn log-root-dir
+  "Given an appender name, as configured, get the parent directory of the appender's log file.
+
+Note that if anything goes wrong, this will throw an Error and exit."
+  [appender-name]
+  (let [appender (.getAppender (LoggerFactory/getLogger Logger/ROOT_LOGGER_NAME) appender-name)]
+    (if (and appender-name appender (instance? FileAppender appender))
+      (.getParent (File. (.getFile appender)))
+      (throw
+       (RuntimeException. "Log viewer could not find configured appender, or the appender is not a FileAppender. Please check that the appender name configured in storm and logback agree.")))))
+
+(defn log-page [file tail grep root-dir]
+  (let [path (.getCanonicalPath (File. root-dir file))
         tail (if tail
                (min 10485760 (Integer/parseInt tail))
                10240)
@@ -59,8 +75,8 @@
     ]))
 
 (defroutes log-routes
-  (GET "/log" [:as {cookies :cookies} & m]
-       (log-template (log-page (:file m) (:tail m) (:grep m))))
+  (GET "/log" [:as req & m]
+       (log-template (log-page (:file m) (:tail m) (:grep m) (:log-root req))))
   (GET "/loglevel" [:as {cookies :cookies} & m]
        (log-template (log-level-page (:name m) (:level m))))
   (route/resources "/")
@@ -70,10 +86,16 @@
   (handler/site log-routes)
  )
 
-(defn start-logviewer [port]
-  (run-jetty logapp {:port port}))
+(defn conf-middleware
+  "For passing the storm configuration with each request."
+  [app log-root]
+  (fn [req]
+    (app (assoc req :log-root log-root))))
+
+(defn start-logviewer [port log-root]
+  (run-jetty (conf-middleware logapp log-root) {:port port}))
 
 (defn -main []
-  (let [conf (read-storm-config)]
-    (start-logviewer (int (conf LOGVIEWER-PORT)))))
-
+  (let [conf (read-storm-config)
+        log-root (log-root-dir (conf LOGVIEWER-APPENDER-NAME))]
+    (start-logviewer (int (conf LOGVIEWER-PORT)) log-root)))
diff --git a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
index 04731dc..caac996 100644
--- a/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/nimbus.clj
@@ -326,7 +326,7 @@
 ;; Does not assume that clocks are synchronized. Executor heartbeat is only used so that
 ;; nimbus knows when it's received a new heartbeat. All timing is done by nimbus and
 ;; tracked through heartbeat-cache
-(defn- update-executor-cache [curr hb]
+(defn- update-executor-cache [curr hb timeout]
   (let [reported-time (:time-secs hb)
         {last-nimbus-time :nimbus-time
          last-reported-time :executor-reported-time} curr
@@ -338,15 +338,18 @@
                       (current-time-secs)
                       last-nimbus-time
                       )]
-      {:nimbus-time nimbus-time
+      {:is-timed-out (and
+                       nimbus-time
+                       (>= (time-delta nimbus-time) timeout))
+       :nimbus-time nimbus-time
        :executor-reported-time reported-time}))
 
-(defn update-heartbeat-cache [cache executor-beats all-executors]
+(defn update-heartbeat-cache [cache executor-beats all-executors timeout]
   (let [cache (select-keys cache all-executors)]
     (into {}
       (for [executor all-executors :let [curr (cache executor)]]
         [executor
-         (update-executor-cache curr (get executor-beats executor))]
+         (update-executor-cache curr (get executor-beats executor) timeout)]
          ))))
 
 (defn update-heartbeats! [nimbus storm-id all-executors existing-assignment]
@@ -355,7 +358,8 @@
         executor-beats (.executor-beats storm-cluster-state storm-id (:executor->node+port existing-assignment))
         cache (update-heartbeat-cache (@(:heartbeats-cache nimbus) storm-id)
                                       executor-beats
-                                      all-executors)]
+                                      all-executors
+                                      ((:conf nimbus) NIMBUS-TASK-TIMEOUT-SECS))]
       (swap! (:heartbeats-cache nimbus) assoc storm-id cache)))
 
 (defn- update-all-heartbeats! [nimbus existing-assignments topology->executors]
@@ -380,14 +384,12 @@
     (->> all-executors
         (filter (fn [executor]
           (let [start-time (get executor-start-times executor)
-                nimbus-time (-> heartbeats-cache (get executor) :nimbus-time)]
+                is-timed-out (-> heartbeats-cache (get executor) :is-timed-out)]
             (if (and start-time
                    (or
                     (< (time-delta start-time)
                        (conf NIMBUS-TASK-LAUNCH-SECS))
-                    (not nimbus-time)
-                    (< (time-delta nimbus-time)
-                       (conf NIMBUS-TASK-TIMEOUT-SECS))
+                    (not is-timed-out)
                     ))
               true
               (do
diff --git a/storm-core/src/jvm/backtype/storm/Config.java b/storm-core/src/jvm/backtype/storm/Config.java
index 89e17d2..13e8d02 100644
--- a/storm-core/src/jvm/backtype/storm/Config.java
+++ b/storm-core/src/jvm/backtype/storm/Config.java
@@ -313,6 +313,12 @@
     public static final Object LOGVIEWER_CHILDOPTS_SCHEMA = String.class;
 
     /**
+     * Appender name used by log viewer to determine log directory.
+     */
+    public static final String LOGVIEWER_APPENDER_NAME = "logviewer.appender.name";
+    public static final Object LOGVIEWER_APPENDER_NAME_SCHEMA = String.class;
+
+    /**
      * Childopts for Storm UI Java process.
      */
     public static final String UI_CHILDOPTS = "ui.childopts";
@@ -727,6 +733,12 @@
     public static final Object TOPOLOGY_NAME_SCHEMA = String.class;
 
     /**
+     * Max pending tuples in one ShellBolt
+     */
+    public static final String TOPOLOGY_SHELLBOLT_MAX_PENDING="topology.shellbolt.max.pending";
+    public static final Object TOPOLOGY_SHELLBOLT_MAX_PENDING_SCHEMA = Number.class;
+
+    /**
      * The root directory in ZooKeeper for metadata about TransactionalSpouts.
      */
     public static final String TRANSACTIONAL_ZOOKEEPER_ROOT="transactional.zookeeper.root";
diff --git a/storm-core/src/jvm/backtype/storm/task/ShellBolt.java b/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
index 854aa8f..3fa6741 100644
--- a/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
+++ b/storm-core/src/jvm/backtype/storm/task/ShellBolt.java
@@ -1,5 +1,6 @@
 package backtype.storm.task;
 
+import backtype.storm.Config;
 import backtype.storm.generated.ShellComponent;
 import backtype.storm.tuple.MessageId;
 import backtype.storm.tuple.Tuple;
@@ -71,6 +72,10 @@
 
     public void prepare(Map stormConf, TopologyContext context,
                         final OutputCollector collector) {
+        Object maxPending = stormConf.get(Config.TOPOLOGY_SHELLBOLT_MAX_PENDING);
+        if (maxPending != null) {
+           this._pendingWrites = new LinkedBlockingQueue(((Number)maxPending).intValue());
+        }
         _rand = new Random();
         _process = new ShellProcess(_command);
         _collector = collector;
diff --git a/storm-core/test/clj/backtype/storm/tick_tuple_test.clj b/storm-core/test/clj/backtype/storm/tick_tuple_test.clj
new file mode 100644
index 0000000..b1679c5
--- /dev/null
+++ b/storm-core/test/clj/backtype/storm/tick_tuple_test.clj
@@ -0,0 +1,35 @@
+(ns backtype.storm.tick-tuple-test
+  (:use [clojure test])
+  (:use [backtype.storm bootstrap testing])
+  (:use [backtype.storm.daemon common]))
+
+(bootstrap)
+
+(defbolt noop-bolt ["tuple"] {:prepare true}
+  [conf context collector]
+  (bolt
+   (execute [tuple])))
+
+(defspout noop-spout ["tuple"]
+  [conf context collector]
+  (spout
+   (nextTuple [])))
+
+(deftest test-tick-tuple-works-with-system-bolt
+  (with-simulated-time-local-cluster [cluster]
+    (let [topology (thrift/mk-topology
+                    {"1" (thrift/mk-spout-spec noop-spout)}
+                    {"2" (thrift/mk-bolt-spec {"1" ["tuple"]} noop-bolt)})]
+      (try
+        (submit-local-topology (:nimbus cluster)
+                               "test"
+                               {TOPOLOGY-TICK-TUPLE-FREQ-SECS 1}
+                               topology)
+        (advance-cluster-time cluster 2)
+        ;; if reaches here, it means everything works ok.
+        (is true)
+        (catch Exception e
+          (is false))))))
+
+
+
diff --git a/storm-netty/src/jvm/backtype/storm/messaging/netty/Client.java b/storm-netty/src/jvm/backtype/storm/messaging/netty/Client.java
index be8ab38..00431d4 100644
--- a/storm-netty/src/jvm/backtype/storm/messaging/netty/Client.java
+++ b/storm-netty/src/jvm/backtype/storm/messaging/netty/Client.java
@@ -93,9 +93,12 @@
      */
     private int getSleepTimeMs()
     {
-        int sleepMs = base_sleep_ms * Math.max(1, random.nextInt(1 << retries.get()));
+        int backoff = 1 << Math.max(1, retries.get());
+        int sleepMs = base_sleep_ms * Math.max(1, random.nextInt(backoff));
         if ( sleepMs > max_sleep_ms )
             sleepMs = max_sleep_ms;
+        if ( sleepMs < base_sleep_ms )
+          sleepMs = base_sleep_ms;
         return sleepMs;
     }