diff --git a/CHANGELOG.md b/CHANGELOG.md
index 93a4b67..2691fb7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
  * STORM-540: Change default time format in logs to ISO8601 in order to include timezone
  * STORM-511: Storm-Kafka spout keeps sending fetch requests with invalid offset
  * STORM-538: Guava com.google.thirdparty.publicsuffix is not shaded
+ * STORM-497: don't modify the netty server taskToQueueId mapping while the someone could be reading it.
 
 ## 0.9.3-rc1
  * STORM-519: add tuple as an input param to HBaseValueMapper 
diff --git a/DEVELOPER.md b/DEVELOPER.md
index 81a4122..4af1e3c 100644
--- a/DEVELOPER.md
+++ b/DEVELOPER.md
@@ -229,6 +229,9 @@
     # Build the code and run the tests (requires nodejs, python and ruby installed) 
     $ mvn clean install
 
+    # Build the code and run the tests, with specifying default test timeout (in millisecond)
+    $ mvn clean install -DSTORM_TEST_TIMEOUT_MS=10000
+
     # Build the code but skip the tests
     $ mvn clean install -DskipTests=true
 
diff --git a/README.markdown b/README.markdown
index 2f90a4e..2e9286b 100644
--- a/README.markdown
+++ b/README.markdown
@@ -79,6 +79,7 @@
 * Michael G. Noll ([@miguno](https://github.com/miguno))
 * Kishor Patil ([@kishorvpatil](https://github.com/kishorvpatil))
 * Sriharsha Chintalapani([@harshach](https://github.com/harshach))
+* Sean Zhong ([@clockfly] (http://github.com/clockfly))
 
 ## Contributors
 
diff --git a/bin/storm.cmd b/bin/storm.cmd
index 359acb6..0cf6b52 100644
--- a/bin/storm.cmd
+++ b/bin/storm.cmd
@@ -69,7 +69,16 @@
     set STORM_OPTS=%STORM_CLIENT_OPTS% %STORM_OPTS% -Dstorm.jar=%2
     set CLASSPATH=%CLASSPATH%;%2
     set CLASS=%3
-    set storm-command-arguments=%4 %5 %6 %7 %8 %9
+    set args=%4
+    shift
+    :start
+    if [%4] == [] goto done
+    set args=%args% %4
+    shift
+    goto start
+
+    :done
+    set storm-command-arguments=%args%
   )
   
   if not defined STORM_LOG_FILE (
diff --git a/external/storm-kafka/README.md b/external/storm-kafka/README.md
index 73cf334..02d9767 100644
--- a/external/storm-kafka/README.md
+++ b/external/storm-kafka/README.md
@@ -33,9 +33,9 @@
     Broker brokerForPartition1 = new Broker("localhost", 9092);//localhost:9092 but we specified the port explicitly
     Broker brokerForPartition2 = new Broker("localhost:9092");//localhost:9092 specified as one string.
     GlobalPartitionInformation partitionInfo = new GlobalPartitionInformation();
-    partitionInfo.add(0, brokerForPartition0)//mapping form partition 0 to brokerForPartition0
-    partitionInfo.add(1, brokerForPartition1)//mapping form partition 1 to brokerForPartition1
-    partitionInfo.add(2, brokerForPartition2)//mapping form partition 2 to brokerForPartition2
+    partitionInfo.addPartition(0, brokerForPartition0);//mapping form partition 0 to brokerForPartition0
+    partitionInfo.addPartition(1, brokerForPartition1);//mapping form partition 1 to brokerForPartition1
+    partitionInfo.addPartition(2, brokerForPartition2);//mapping form partition 2 to brokerForPartition2
     StaticHosts hosts = new StaticHosts(partitionInfo);
 ```
 
@@ -121,7 +121,6 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka_2.10</artifactId>
             <version>0.8.1.1</version>
-            <scope>provided</scope>
             <exclusions>
                 <exclusion>
                     <groupId>org.apache.zookeeper</groupId>
diff --git a/storm-core/src/clj/backtype/storm/daemon/supervisor.clj b/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
index cce89bc..b370eb7 100644
--- a/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
+++ b/storm-core/src/clj/backtype/storm/daemon/supervisor.clj
@@ -502,6 +502,8 @@
     :distributed [supervisor storm-id port worker-id]
     (let [conf (:conf supervisor)
           storm-home (System/getProperty "storm.home")
+          storm-options (System/getProperty "storm.options")
+          storm-conf-file (System/getProperty "storm.conf.file")
           storm-log-dir (or (System/getProperty "storm.log.dir") (str storm-home file-path-separator "logs"))
           stormroot (supervisor-stormdist-root conf storm-id)
           jlp (jlp stormroot conf)
@@ -528,6 +530,8 @@
                     [(str "-Djava.library.path=" jlp)
                      (str "-Dlogfile.name=" logfilename)
                      (str "-Dstorm.home=" storm-home)
+                     (str "-Dstorm.conf.file=" storm-conf-file)
+                     (str "-Dstorm.options=" storm-options)
                      (str "-Dstorm.log.dir=" storm-log-dir)
                      (str "-Dlogback.configurationFile=" storm-home file-path-separator "logback" file-path-separator "cluster.xml")
                      (str "-Dstorm.id=" storm-id)
@@ -625,4 +629,4 @@
         ))))
 
 (defn -main []
-  (-launch (standalone-supervisor)))
\ No newline at end of file
+  (-launch (standalone-supervisor)))
diff --git a/storm-core/src/clj/backtype/storm/testing.clj b/storm-core/src/clj/backtype/storm/testing.clj
index 798dccc..79d3277 100644
--- a/storm-core/src/clj/backtype/storm/testing.clj
+++ b/storm-core/src/clj/backtype/storm/testing.clj
@@ -187,7 +187,9 @@
       ;; on windows, the host process still holds lock on the logfile
       (catch Exception e (log-message (.getMessage e)))) ))
 
-(def TEST-TIMEOUT-MS 5000)
+(def TEST-TIMEOUT-MS
+  (let [timeout (System/getProperty "STORM_TEST_TIMEOUT_MS")]
+    (parse-int (if timeout timeout "5000"))))
 
 (defmacro while-timeout [timeout-ms condition & body]
   `(let [end-time# (+ (System/currentTimeMillis) ~timeout-ms)]
@@ -594,7 +596,7 @@
                            (not= (global-amt track-id "transferred")                                 
                                  (global-amt track-id "processed"))
                            ))]
-        (while-timeout TEST-TIMEOUT-MS (waiting?)
+        (while-timeout timeout-ms (waiting?)
                        ;; (println "Spout emitted: " (global-amt track-id "spout-emitted"))
                        ;; (println "Processed: " (global-amt track-id "processed"))
                        ;; (println "Transferred: " (global-amt track-id "transferred"))
diff --git a/storm-core/src/clj/backtype/storm/testing4j.clj b/storm-core/src/clj/backtype/storm/testing4j.clj
index ff939a9..7790b6b 100644
--- a/storm-core/src/clj/backtype/storm/testing4j.clj
+++ b/storm-core/src/clj/backtype/storm/testing4j.clj
@@ -44,6 +44,7 @@
              ^:static [mkTrackedTopology [backtype.storm.ILocalCluster backtype.storm.generated.StormTopology] backtype.storm.testing.TrackedTopology]
              ^:static [trackedWait [backtype.storm.testing.TrackedTopology] void]
              ^:static [trackedWait [backtype.storm.testing.TrackedTopology Integer] void]
+             ^:static [trackedWait [backtype.storm.testing.TrackedTopology Integer Integer] void]
              ^:static [advanceClusterTime [backtype.storm.ILocalCluster Integer Integer] void]
              ^:static [advanceClusterTime [backtype.storm.ILocalCluster Integer] void]
              ^:static [multiseteq [java.util.Collection java.util.Collection] boolean]
@@ -123,6 +124,8 @@
       (TrackedTopology.)))
 
 (defn -trackedWait
+  ([^TrackedTopology trackedTopology ^Integer amt ^Integer timeout-ms]
+    (tracked-wait trackedTopology amt timeout-ms))
   ([^TrackedTopology trackedTopology ^Integer amt]
    (tracked-wait trackedTopology amt))
   ([^TrackedTopology trackedTopology]
diff --git a/storm-core/src/dev/resources/storm.py b/storm-core/src/dev/resources/storm.py
index d2a3082..a2aa976 100755
--- a/storm-core/src/dev/resources/storm.py
+++ b/storm-core/src/dev/resources/storm.py
@@ -210,8 +210,12 @@
             while True:
                 tup = readTuple()
                 ANCHOR_TUPLE = tup
-                self.process(tup)
-                ack(tup)
+                try:
+                    self.process(tup)
+                    ack(tup)
+                except Exception, e:
+                    reportError(traceback.format_exc(e))
+                    fail(tup)
         except Exception, e:
             reportError(traceback.format_exc(e))
 
diff --git a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
index 20a147d..2499e65 100644
--- a/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
+++ b/storm-core/src/jvm/backtype/storm/messaging/netty/Server.java
@@ -56,7 +56,7 @@
     final ServerBootstrap bootstrap;
     
     private int queueCount;
-    HashMap<Integer, Integer> taskToQueueId = null;
+    private volatile HashMap<Integer, Integer> taskToQueueId = null;
     int roundRobinQueueId;
 	
     boolean closing = false;
@@ -131,18 +131,18 @@
     
     private Integer getMessageQueueId(int task) {
       // try to construct the map from taskId -> queueId in round robin manner.
-      
       Integer queueId = taskToQueueId.get(task);
       if (null == queueId) {
-        synchronized(taskToQueueId) {
-          //assgin task to queue in round-robin manner
-          if (null == taskToQueueId.get(task)) {
+        synchronized (this) {
+          queueId = taskToQueueId.get(task);
+          if (queueId == null) {
             queueId = roundRobinQueueId++;
-            
-            taskToQueueId.put(task, queueId);
             if (roundRobinQueueId == queueCount) {
               roundRobinQueueId = 0;
             }
+            HashMap<Integer, Integer> newRef = new HashMap<Integer, Integer>(taskToQueueId);
+            newRef.put(task, queueId);
+            taskToQueueId = newRef;
           }
         }
       }
diff --git a/storm-core/src/jvm/backtype/storm/utils/LocalState.java b/storm-core/src/jvm/backtype/storm/utils/LocalState.java
index 0d0ae07..14a45da 100644
--- a/storm-core/src/jvm/backtype/storm/utils/LocalState.java
+++ b/storm-core/src/jvm/backtype/storm/utils/LocalState.java
@@ -18,6 +18,8 @@
 package backtype.storm.utils;
 
 import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.util.Map;
@@ -30,28 +32,42 @@
  * Every read/write hits disk.
  */
 public class LocalState {
+    public static Logger LOG = LoggerFactory.getLogger(LocalState.class);
+
     private VersionedStore _vs;
     
     public LocalState(String backingDir) throws IOException {
         _vs = new VersionedStore(backingDir);
     }
-    
+
     public synchronized Map<Object, Object> snapshot() throws IOException {
         int attempts = 0;
         while(true) {
-            String latestPath = _vs.mostRecentVersionPath();
-            if(latestPath==null) return new HashMap<Object, Object>();
             try {
-                return (Map<Object, Object>) Utils.deserialize(FileUtils.readFileToByteArray(new File(latestPath)));
-            } catch(IOException e) {
+                return deserializeLatestVersion();
+            } catch (IOException e) {
                 attempts++;
-                if(attempts >= 10) {
+                if (attempts >= 10) {
                     throw e;
                 }
             }
         }
     }
-    
+
+    private Map<Object, Object> deserializeLatestVersion() throws IOException {
+        String latestPath = _vs.mostRecentVersionPath();
+        Map<Object, Object> result = new HashMap<Object, Object>();
+        if (latestPath != null) {
+            byte[] serialized = FileUtils.readFileToByteArray(new File(latestPath));
+            if (serialized.length == 0) {
+                LOG.warn("LocalState file '{}' contained no data, resetting state", latestPath);
+            } else {
+                result = (Map<Object, Object>) Utils.deserialize(serialized);
+            }
+        }
+        return result;
+    }
+
     public Object get(Object key) throws IOException {
         return snapshot().get(key);
     }
diff --git a/storm-core/src/multilang/py/storm.py b/storm-core/src/multilang/py/storm.py
index d2a3082..a2aa976 100755
--- a/storm-core/src/multilang/py/storm.py
+++ b/storm-core/src/multilang/py/storm.py
@@ -210,8 +210,12 @@
             while True:
                 tup = readTuple()
                 ANCHOR_TUPLE = tup
-                self.process(tup)
-                ack(tup)
+                try:
+                    self.process(tup)
+                    ack(tup)
+                except Exception, e:
+                    reportError(traceback.format_exc(e))
+                    fail(tup)
         except Exception, e:
             reportError(traceback.format_exc(e))
 
diff --git a/storm-core/test/clj/backtype/storm/local_state_test.clj b/storm-core/test/clj/backtype/storm/local_state_test.clj
index ba2b969..4bd58ec 100644
--- a/storm-core/test/clj/backtype/storm/local_state_test.clj
+++ b/storm-core/test/clj/backtype/storm/local_state_test.clj
@@ -16,7 +16,9 @@
 (ns backtype.storm.local-state-test
   (:use [clojure test])
   (:use [backtype.storm testing])
-  (:import [backtype.storm.utils LocalState]))
+  (:import [backtype.storm.utils LocalState]
+           [org.apache.commons.io FileUtils]
+           [java.io File]))
 
 (deftest test-local-state
   (with-local-tmp [dir1 dir2]
@@ -41,3 +43,13 @@
       (.put ls2 "b" 8)
       (is (= 8 (.get ls2 "b")))
       )))
+
+(deftest empty-state
+  (with-local-tmp [dir]
+    (let [ls (LocalState. dir)
+          data (FileUtils/openOutputStream (File. dir "12345"))
+          version (FileUtils/openOutputStream (File. dir "12345.version"))]
+      (is (= nil (.get ls "c")))
+      (.put ls "a" 1)
+      (is (= 1 (.get ls "a")))
+  )))
diff --git a/storm-core/test/clj/backtype/storm/supervisor_test.clj b/storm-core/test/clj/backtype/storm/supervisor_test.clj
index ba74d88..0bb47f3 100644
--- a/storm-core/test/clj/backtype/storm/supervisor_test.clj
+++ b/storm-core/test/clj/backtype/storm/supervisor_test.clj
@@ -255,6 +255,8 @@
                                ["-Djava.library.path="
                                 (str "-Dlogfile.name=worker-" mock-port ".log")
                                 "-Dstorm.home="
+                                "-Dstorm.conf.file="
+                                "-Dstorm.options="
                                 (str "-Dstorm.log.dir=" file-path-separator "logs")
                                 (str "-Dlogback.configurationFile=" file-path-separator "logback" file-path-separator "cluster.xml")
                                 (str "-Dstorm.id=" mock-storm-id)
@@ -469,4 +471,4 @@
      (validate-launched-once (:launched changed)
                              {"sup1" [3 4]}
                              (get-storm-id (:storm-cluster-state cluster) "topology2"))
-     )))
\ No newline at end of file
+     )))