start worker successfully
Project: http://git-wip-us.apache.org/repos/asf/storm/repo Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/42928c21 Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/42928c21 Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/42928c21 Branch: refs/heads/master Commit: 42928c2182cf2b755c6f98ad039b2e858787dfe4 Parents: b49c995 Author: xiaojian.fxj <[email protected]> Authored: Tue Mar 15 00:16:19 2016 +0800 Committer: xiaojian.fxj <[email protected]> Committed: Tue Mar 15 00:16:19 2016 +0800 ---------------------------------------------------------------------- .../src/clj/org/apache/storm/daemon/nimbus.clj | 4 +-- .../workermanager/DefaultWorkerManager.java | 33 +++++++++++--------- 2 files changed, 21 insertions(+), 16 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/storm/blob/42928c21/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj ---------------------------------------------------------------------- diff --git a/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj b/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj index e6fd0a2..5820ee9 100644 --- a/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj +++ b/storm-core/src/clj/org/apache/storm/daemon/nimbus.clj @@ -1353,8 +1353,8 @@ (str "Failed to submit topology. Topology requests more than " workers-allowed " workers.")))))) (defn nimbus-topology-bases [storm-cluster-state] - map-val #(clojurify-storm-base %) (clojurify-structure - (StormCommon/topologyBases storm-cluster-state))) + (map-val #(clojurify-storm-base %) (clojurify-structure + (StormCommon/topologyBases storm-cluster-state)))) (defn- set-logger-timeouts [log-config] (let [timeout-secs (.get_reset_log_level_timeout_secs log-config) http://git-wip-us.apache.org/repos/asf/storm/blob/42928c21/storm-core/src/jvm/org/apache/storm/daemon/supervisor/workermanager/DefaultWorkerManager.java ---------------------------------------------------------------------- diff --git a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/workermanager/DefaultWorkerManager.java b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/workermanager/DefaultWorkerManager.java index b19fd89..a73a9bd 100644 --- a/storm-core/src/jvm/org/apache/storm/daemon/supervisor/workermanager/DefaultWorkerManager.java +++ b/storm-core/src/jvm/org/apache/storm/daemon/supervisor/workermanager/DefaultWorkerManager.java @@ -340,7 +340,7 @@ public class DefaultWorkerManager implements IWorkerManager { String os = System.getProperty("os.name").replaceAll("\\s+", "_"); String arch = System.getProperty("os.arch"); String archResourceRoot = resourceRoot + Utils.FILE_PATH_SEPARATOR + os + "-" + arch; - String ret = archResourceRoot + Utils.FILE_PATH_SEPARATOR + resourceRoot + Utils.FILE_PATH_SEPARATOR + conf.get(Config.JAVA_LIBRARY_PATH); + String ret = archResourceRoot + Utils.CLASS_PATH_SEPARATOR + resourceRoot + Utils.CLASS_PATH_SEPARATOR + conf.get(Config.JAVA_LIBRARY_PATH); return ret; } @@ -373,23 +373,28 @@ public class DefaultWorkerManager implements IWorkerManager { List<String> rets = new ArrayList<>(); if (value instanceof String) { String string = (String) value; - string = string.replace("%ID%", String.valueOf(port)); - string = string.replace("%WORKER-ID%", workerId); - string = string.replace("%TOPOLOGY-ID%", stormId); - string = string.replace("%WORKER-PORT%", String.valueOf(port)); - string = string.replace("%HEAP-MEM%", String.valueOf(memOnheap)); - String[] strings = string.split("\\s+"); - rets.addAll(Arrays.asList(strings)); + if (StringUtils.isNotBlank(string)){ + string = string.replace("%ID%", String.valueOf(port)); + string = string.replace("%WORKER-ID%", workerId); + string = string.replace("%TOPOLOGY-ID%", stormId); + string = string.replace("%WORKER-PORT%", String.valueOf(port)); + string = string.replace("%HEAP-MEM%", String.valueOf(memOnheap)); + String[] strings = string.split("\\s+"); + rets.addAll(Arrays.asList(strings)); + } + } else if (value instanceof List) { List<Object> objects = (List<Object>) value; for (Object object : objects) { String str = (String) object; - str = str.replace("%ID%", String.valueOf(port)); - str = str.replace("%WORKER-ID%", workerId); - str = str.replace("%TOPOLOGY-ID%", stormId); - str = str.replace("%WORKER-PORT%", String.valueOf(port)); - str = str.replace("%HEAP-MEM%", String.valueOf(memOnheap)); - rets.add(str); + if (StringUtils.isNotBlank(str)){ + str = str.replace("%ID%", String.valueOf(port)); + str = str.replace("%WORKER-ID%", workerId); + str = str.replace("%TOPOLOGY-ID%", stormId); + str = str.replace("%WORKER-PORT%", String.valueOf(port)); + str = str.replace("%HEAP-MEM%", String.valueOf(memOnheap)); + rets.add(str); + } } } return rets;
