diff --git a/bin/storm.py b/bin/storm.py
index 16ddb95..38c5ae2 100755
--- a/bin/storm.py
+++ b/bin/storm.py
@@ -231,7 +231,7 @@
 
 
 def exec_storm_class(klass, storm_config_opts, jvmtype="-server", jvmopts=[],
-                     extrajars=[], args=[], fork=False, daemon=True, client=False, daemonName="",
+                     extrajars=[], main_class_args=[], fork=False, daemon=True, client=False, daemonName="",
                      overriding_conf_file=None):
     storm_log_dir = confvalue("storm.log.dir", storm_config_opts=storm_config_opts,
                               extrapaths=[CLUSTER_CONF_DIR], overriding_conf_file=overriding_conf_file)
@@ -246,7 +246,7 @@
        "-Djava.library.path=" + confvalue("java.library.path", storm_config_opts, extrajars, daemon=daemon),
        "-Dstorm.conf.file=" + (overriding_conf_file if overriding_conf_file else ""),
        "-cp", get_classpath(extrajars, daemon, client=client),
-    ] + jvmopts + [klass] + list(args)
+    ] + jvmopts + [klass] + list(main_class_args)
     print("Running: " + " ".join(all_args))
     sys.stdout.flush()
     exit_code = 0
@@ -282,7 +282,7 @@
         klass, args.storm_config_opts,
         jvmtype="-client",
         extrajars=extra_jars,
-        args=args.main_args,
+        main_class_args=args.main_args,
         daemon=False,
         jvmopts=JAR_JVM_OPTS + extrajvmopts + ["-Dstorm.jar=" + jarfile] +
                 ["-Dstorm.dependency.jars=" + ",".join(local_jars)] +
@@ -1106,7 +1106,7 @@
         "org.apache.storm.sql.StormSqlRunner", storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=extra_jars,
-        args=sql_args,
+        main_class_args=sql_args,
         daemon=False,
         jvmopts=["-Dstorm.dependency.jars=" + ",".join(local_jars)] +
                 ["-Dstorm.dependency.artifacts=" + json.dumps(artifact_to_file_jars)],
@@ -1116,7 +1116,7 @@
 def kill(args):
     exec_storm_class(
         "org.apache.storm.command.KillTopology",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1127,7 +1127,7 @@
         raise argparse.ArgumentTypeError("please provide a list of cred key and value pairs " + cred_list)
     exec_storm_class(
         "org.apache.storm.command.UploadCredentials",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1138,7 +1138,7 @@
         raise argparse.ArgumentTypeError("Replication factor needed when doing blob update")
     exec_storm_class(
         "org.apache.storm.command.Blobstore",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1147,7 +1147,7 @@
 def heartbeats(args):
     exec_storm_class(
         "org.apache.storm.command.Heartbeats",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1156,7 +1156,7 @@
 def activate(args):
     exec_storm_class(
         "org.apache.storm.command.Activate",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1164,7 +1164,7 @@
 def listtopos(args):
     exec_storm_class(
         "org.apache.storm.command.ListTopologies",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1180,7 +1180,7 @@
             raise argparse.ArgumentTypeError("Should be in the form[logger name]=[log level][:optional timeout]")
     exec_storm_class(
         "org.apache.storm.command.SetLogLevel",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1188,7 +1188,7 @@
 def deactivate(args):
     exec_storm_class(
         "org.apache.storm.command.Deactivate",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1205,7 +1205,7 @@
             raise argparse.ArgumentTypeError("Should be in the form component_name:new_executor_count")
     exec_storm_class(
         "org.apache.storm.command.Rebalance",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1214,7 +1214,7 @@
 def get_errors(args):
     exec_storm_class(
         "org.apache.storm.command.GetErrors",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1223,7 +1223,7 @@
 def healthcheck(args):
     exec_storm_class(
         "org.apache.storm.command.HealthCheck",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1232,7 +1232,7 @@
 def kill_workers(args):
     exec_storm_class(
         "org.apache.storm.command.KillWorkers",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1241,7 +1241,7 @@
 def admin(args):
     exec_storm_class(
         "org.apache.storm.command.AdminCommands",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1254,7 +1254,7 @@
     runnerargs.extend(args.args)
     exec_storm_class(
         "org.apache.storm.command.ShellSubmission", storm_config_opts=args.storm_config_opts,
-        args=runnerargs,
+        main_class_args=runnerargs,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR],
         fork=True,
@@ -1401,7 +1401,7 @@
 
     exec_storm_class(
         "org.apache.storm.command.BasicDrpcClient",
-        args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
+        main_class_args=remove_common_options(sys.argv[2:]), storm_config_opts=args.storm_config_opts,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR],
         overriding_conf_file=args.config)
@@ -1464,18 +1464,19 @@
 def monitor(args):
     exec_storm_class(
         "org.apache.storm.command.Monitor", storm_config_opts=args.storm_config_opts,
-        args=remove_common_options(sys.argv[2:]),
+        main_class_args=remove_common_options(sys.argv[2:]),
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR])
 
-
 def main():
     init_storm_env()
     storm_parser = initialize_main_command()
     if len(sys.argv) == 1:
         storm_parser.print_help(sys.stderr)
         sys.exit(1)
-    raw_args = storm_parser.parse_args()
+    raw_args, unknown_args = storm_parser.parse_known_args()
+    if hasattr(raw_args, "main_args"):
+        raw_args.main_args += unknown_args
     raw_args.func(raw_args)
 
 
diff --git a/storm-client/test/py/test_storm_cli.py b/storm-client/test/py/test_storm_cli.py
index bf5001c..5f46475 100644
--- a/storm-client/test/py/test_storm_cli.py
+++ b/storm-client/test/py/test_storm_cli.py
@@ -59,19 +59,41 @@
             './external/storm-redis/storm-redis-1.1.0.jar,./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"', '--artifacts', '"redis.clients:jedis:2.9.0,org.apache.kafka:kafka-clients:1.0.0^org.slf4j:slf4j-api"', '--artifactRepositories', '"jboss-repository^http://repository.jboss.com/maven2,HDPRepo^http://repo.hortonworks.com/content/groups/public/'
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
-                self.java_cmd, '-client','-Ddaemon.name=', '-Dstorm.options=+topology.blobstore.map%3D%27%7B%22key1%22%3A%7B%22localname%22%3A%22blob_file%22%2C+%22uncompress%22%3Afalse%7D%2C%22key2%22%3A%7B%7D%7D%27',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=+topology.blobstore.map%3D%27%7B%22key1%22%3A%7B%22localname%22%3A%22blob_file%22%2C+%22uncompress%22%3Afalse%7D%2C%22key2%22%3A%7B%7D%7D%27',
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:example/storm-starter/storm-starter-topologies-*.jar:' + self.storm_dir + '/conf:' + self.storm_dir + '/bin:./external/storm-redis/storm-redis-1.1.0.jar:./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"', '-Dstorm.jar=example/storm-starter/storm-starter-topologies-*.jar', '-Dstorm.dependency.jars=./external/storm-redis/storm-redis-1.1.0.jar,./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"', '-Dstorm.dependency.artifacts={}',
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir
+                + '/extlib:example/storm-starter/storm-starter-topologies-*.jar:' + self.storm_dir + '/conf:'
+                + self.storm_dir + '/bin:./external/storm-redis/storm-redis-1.1.0.jar:./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"', '-Dstorm.jar=example/storm-starter/storm-starter-topologies-*.jar', '-Dstorm.dependency.jars=./external/storm-redis/storm-redis-1.1.0.jar,./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"', '-Dstorm.dependency.artifacts={}',
                 'org.apache.storm.starter.RollingTopWords', 'blobstore-remote2', 'remote'
             ])
         )
 
+        self.mock_execvp.reset_mock()
+
+        self.base_test([
+            'storm', 'jar', '/path/to/jar.jar', 'some.Topology.Class',
+            '-name', 'run-topology', 'randomArgument', '-randomFlag', 'randomFlagValue', '-rotateSize', '0.0001',
+            '--hdfsConf', 'someOtherHdfsConf', 'dfs.namenode.kerberos.principal.pattern=hdfs/*.EV..COM'
+        ], self.mock_execvp, mock.call(
+            self.java_cmd, [
+                self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir
+                + '/extlib:/path/to/jar.jar:' + self.storm_dir + '/conf:' + self.storm_dir + '/bin:',
+                '-Dstorm.jar=/path/to/jar.jar', '-Dstorm.dependency.jars=', '-Dstorm.dependency.artifacts={}',
+                'some.Topology.Class', '-name', 'run-topology', 'randomArgument', '-randomFlag', 'randomFlagValue',
+                '-rotateSize', '0.0001', '--hdfsConf', 'someOtherHdfsConf',
+                'dfs.namenode.kerberos.principal.pattern=hdfs/*.EV..COM'
+            ])
+        )
+
     def test_localconfvalue_command(self):
         self.base_test(
             ["storm", "localconfvalue", "conf_name"], self.mock_popen, mock.call([
              self.java_cmd, '-client', '-Dstorm.options=',
-             '-Dstorm.conf.file=', '-cp', '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +'/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
+             '-Dstorm.conf.file=', '-cp',  self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +'/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
              'org.apache.storm.command.ConfigValue', 'conf_name'
              ], stdout=-1
             )
@@ -81,7 +103,7 @@
         self.base_test(
             ["storm", "remoteconfvalue", "conf_name"], self.mock_popen, mock.call([
              self.java_cmd, '-client', '-Dstorm.options=',
-             '-Dstorm.conf.file=', '-cp', '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
+             '-Dstorm.conf.file=', '-cp',  self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
              'org.apache.storm.command.ConfigValue', 'conf_name'
              ], stdout=-1
             )
@@ -100,9 +122,9 @@
             ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client','-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:example/storm-starter/storm-starter-topologies-*.jar:' + self.storm_dir +
                 '/conf:' + self.storm_dir +
                 '/bin:./external/storm-redis/storm-redis-1.1.0.jar:./external/storm-kafka-client/storm-kafka-client-1.1.0.jar"',
@@ -121,9 +143,9 @@
             ], self.mock_execvp, mock.call(
                 self.java_cmd,
                 [self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                 '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                 '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                  '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                 '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' +
+                 self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' +
                  self.storm_dir +
                  '/conf:' + self.storm_dir + '/bin:' + self.storm_dir + '/lib-tools/sql/core',\
                  '-Dstorm.dependency.jars=', '-Dstorm.dependency.artifacts={}',
@@ -138,9 +160,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir + '/bin', 'org.apache.storm.command.KillTopology', 'doomed_topology'
             ])
         )
@@ -151,9 +173,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd,  '-client', '-Ddaemon.name=', '-Dstorm.options=test%3Dtest',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=/some/other/storm.yaml',
-                '-cp', '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' +
+                '-cp', self.storm_dir + '/*:' + self.storm_dir + '/lib:' +
                                              self.storm_dir +
                                              '/extlib:' + self.storm_dir + '/extlib-daemon:' +
                                              self.storm_dir + '/conf:' + self.storm_dir +
@@ -167,9 +189,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.Blobstore', 'create',
                 'mytopo:data.tgz', '-f', 'data.tgz', '-a', 'u:alice:rwa,u:bob:rw,o::r'])
@@ -181,10 +203,10 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '',
+                '-Dstorm.home=' + self.storm_dir,
                 '-Dstorm.log.dir=' + self.storm_dir + "/logs", '-Djava.library.path=',
                 '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.Blobstore', 'list'])
         )
@@ -195,10 +217,10 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '',
+                '-Dstorm.home=' + self.storm_dir,
                 '-Dstorm.log.dir=' + self.storm_dir + "/logs", '-Djava.library.path=',
                 '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.Blobstore', 'list', 'wordstotrack'])
         )
@@ -209,9 +231,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.Blobstore', 'update', '-f',
                 '/wordsToTrack.list', 'wordstotrack'])
@@ -223,9 +245,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.Blobstore', 'cat', 'wordstotrack'])
         )
@@ -236,9 +258,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir + '/bin',
                 'org.apache.storm.command.Activate', 'doomed_topology'
             ])
@@ -250,9 +272,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir +
                 '/bin', 'org.apache.storm.command.Deactivate', 'doomed_topology'
             ])
@@ -264,9 +286,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir +
                 '/bin', 'org.apache.storm.command.Rebalance', 'doomed_topology'
             ])
@@ -278,9 +300,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir +
                 '/bin', 'org.apache.storm.command.ListTopologies'
             ])
@@ -292,9 +314,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=nimbus', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=nimbus.log',
                 '-Dlog4j.configurationFile=' + self.storm_dir + '/log4j2/cluster.xml',
@@ -308,9 +330,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=supervisor', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=supervisor.log',
                 '-Dlog4j.configurationFile=' + self.storm_dir + '/log4j2/cluster.xml',
@@ -324,9 +346,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=pacemaker', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=pacemaker.log',
                 '-Dlog4j.configurationFile=' + self.storm_dir + '/log4j2/cluster.xml',
@@ -340,9 +362,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=ui', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir +
                 '/lib-webapp:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=ui.log',
@@ -357,9 +379,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=logviewer', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir +
                 '/lib-webapp:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=logviewer.log',
@@ -374,9 +396,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-server', '-Ddaemon.name=drpc', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir +
                 '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir +
                 '/lib-webapp:' + self.storm_dir + '/conf',
                 '-Djava.deserialization.disabled=true', '-Dlogfile.name=drpc.log',
@@ -391,9 +413,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir +
                 '/bin', 'org.apache.storm.command.BasicDrpcClient', 'exclaim', 'a', 'exclaim', 'b', 'test', 'bar'
             ])
@@ -403,9 +425,9 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs",
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs",
                 '-Djava.library.path=', '-Dstorm.conf.file=', '-cp',
-                '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
+                self.storm_dir + '/*:' + self.storm_dir + '/lib:' + self.storm_dir + '/extlib:' + self.storm_dir +
                 '/extlib-daemon:' + self.storm_dir + '/conf:' + self.storm_dir +
                 '/bin', 'org.apache.storm.command.BasicDrpcClient', '-f', 'exclaim', 'a', 'b'
             ])
@@ -417,8 +439,8 @@
         ], self.mock_execvp, mock.call(
             self.java_cmd, [
                 self.java_cmd, '-client', '-Ddaemon.name=', '-Dstorm.options=',
-                '-Dstorm.home=' + self.storm_dir + '', '-Dstorm.log.dir=' + self.storm_dir + "/logs", '-Djava.library.path=',
-                '-Dstorm.conf.file=', '-cp', '' + self.storm_dir + '/*:' + self.storm_dir + '/lib:' +
+                '-Dstorm.home=' + self.storm_dir, '-Dstorm.log.dir=' + self.storm_dir + "/logs", '-Djava.library.path=',
+                '-Dstorm.conf.file=', '-cp', self.storm_dir + '/*:' + self.storm_dir + '/lib:' +
                 self.storm_dir + '/extlib:' + self.storm_dir + '/extlib-daemon:' + self.storm_dir + '/conf:' +
                 self.storm_dir + '/bin', 'org.apache.storm.command.HealthCheck'
             ])