diff --git a/bin/storm.py b/bin/storm.py
index 74a7825..0da2ae1 100755
--- a/bin/storm.py
+++ b/bin/storm.py
@@ -326,7 +326,7 @@
     storm blobstore create mytopo:data.tgz -f data.tgz -a u:alice:rwa,u:bob:rw,o::r
     """
     exec_storm_class(
-        "org.apache.storm.command.blobstore",
+        "org.apache.storm.command.Blobstore",
         args=args,
         jvmtype="-client",
         extrajars=[USER_CONF_DIR, STORM_BIN_DIR])
diff --git a/storm-core/src/clj/org/apache/storm/blobstore.clj b/storm-core/src/clj/org/apache/storm/blobstore.clj
deleted file mode 100644
index 92fb44f..0000000
--- a/storm-core/src/clj/org/apache/storm/blobstore.clj
+++ /dev/null
@@ -1,28 +0,0 @@
-;; Licensed to the Apache Software Foundation (ASF) under one
-;; or more contributor license agreements.  See the NOTICE file
-;; distributed with this work for additional information
-;; regarding copyright ownership.  The ASF licenses this file
-;; to you under the Apache License, Version 2.0 (the
-;; "License"); you may not use this file except in compliance
-;; with the License.  You may obtain a copy of the License at
-;;
-;; http://www.apache.org/licenses/LICENSE-2.0
-;;
-;; Unless required by applicable law or agreed to in writing, software
-;; distributed under the License is distributed on an "AS IS" BASIS,
-;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-;; See the License for the specific language governing permissions and
-;; limitations under the License.
-
-(ns org.apache.storm.blobstore
-  (:import [org.apache.storm.utils Utils ConfigUtils])
-  (:import [org.apache.storm.blobstore ClientBlobStore])
-  (:use [org.apache.storm config util]))
-
-(defmacro with-configured-blob-client
-  [client-sym & body]
-  `(let [conf# (clojurify-structure (ConfigUtils/readStormConfig))
-         ^ClientBlobStore ~client-sym (Utils/getClientBlobStore conf#)]
-     (try
-       ~@body
-       (finally (.shutdown ~client-sym)))))
diff --git a/storm-core/src/clj/org/apache/storm/command/blobstore.clj b/storm-core/src/clj/org/apache/storm/command/blobstore.clj
deleted file mode 100644
index 924f825..0000000
--- a/storm-core/src/clj/org/apache/storm/command/blobstore.clj
+++ /dev/null
@@ -1,163 +0,0 @@
-;; Licensed to the Apache Software Foundation (ASF) under one
-;; or more contributor license agreements.  See the NOTICE file
-;; distributed with this work for additional information
-;; regarding copyright ownership.  The ASF licenses this file
-;; to you under the Apache License, Version 2.0 (the
-;; "License"); you may not use this file except in compliance
-;; with the License.  You may obtain a copy of the License at
-;;
-;; http://www.apache.org/licenses/LICENSE-2.0
-;;
-;; Unless required by applicable law or agreed to in writing, software
-;; distributed under the License is distributed on an "AS IS" BASIS,
-;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-;; See the License for the specific language governing permissions and
-;; limitations under the License.
-(ns org.apache.storm.command.blobstore
-  (:import [java.io InputStream OutputStream]
-           [org.apache.storm.generated SettableBlobMeta AccessControl AuthorizationException
-            KeyNotFoundException]
-           [org.apache.storm.blobstore BlobStoreAclHandler]
-           [org.apache.storm.utils Utils])
-  (:use [org.apache.storm config]
-        [clojure.string :only [split]]
-        [clojure.tools.cli :only [cli]]
-        [clojure.java.io :only [copy input-stream output-stream]]
-        [org.apache.storm blobstore log])
-  (:gen-class))
-
-(defn update-blob-from-stream
-  "Update a blob in the blob store from an InputStream"
-  [key ^InputStream in]
-  (with-configured-blob-client blobstore
-    (let [out (.updateBlob blobstore key)]
-      (try 
-        (copy in out)
-        (.close out)
-        (catch Exception e
-          (log-message e)
-          (.cancel out)
-          (throw e))))))
-
-(defn create-blob-from-stream
-  "Create a blob in the blob store from an InputStream"
-  [key ^InputStream in ^SettableBlobMeta meta]
-  (with-configured-blob-client blobstore
-    (let [out (.createBlob blobstore key meta)]
-      (try 
-        (copy in out)
-        (.close out)
-        (catch Exception e
-          (.cancel out)
-          (throw e))))))
-
-(defn read-blob
-  "Read a blob in the blob store and write to an OutputStream"
-  [key ^OutputStream out]
-  (with-configured-blob-client blobstore
-    (with-open [in (.getBlob blobstore key)]
-      (copy in out))))
-
-(defn as-access-control
-  "Convert a parameter to an AccessControl object"
-  [param]
-  (BlobStoreAclHandler/parseAccessControl (str param)))
-
-(defn as-acl
-  [param]
-  (map as-access-control (split param #",")))
-
-(defn access-control-str
-  [^AccessControl acl]
-  (BlobStoreAclHandler/accessControlToString acl))
-
-(defn read-cli [args]
-  (let [[{file :file} [key] _] (cli args ["-f" "--file" :default nil])]
-    (if file
-      (with-open [f (output-stream file)]
-        (read-blob key f))
-      (read-blob key System/out))))
-
-(defn update-cli [args]
-  (let [[{file :file} [key] _] (cli args ["-f" "--file" :default nil])]
-    (if file
-      (with-open [f (input-stream file)]
-        (update-blob-from-stream key f))
-      (update-blob-from-stream key System/in))
-    (log-message "Successfully updated " key)))
-
-(defn create-cli [args]
-  (let [[{file :file acl :acl replication-factor :replication-factor} [key] _] (cli args ["-f" "--file" :default nil]
-                                                  ["-a" "--acl" :default [] :parse-fn as-acl]
-                                                  ["-r" "--replication-factor" :default -1 :parse-fn #(Integer/parseInt %)])
-        meta (doto (SettableBlobMeta. acl)
-                   (.set_replication_factor replication-factor))]
-    (Utils/validateKeyName key)
-    (log-message "Creating " key " with ACL " (pr-str (map access-control-str acl)))
-    (if file
-      (with-open [f (input-stream file)]
-        (create-blob-from-stream key f meta))
-      (create-blob-from-stream key System/in meta))
-    (log-message "Successfully created " key)))
-
-(defn delete-cli [args]
-  (with-configured-blob-client blobstore
-    (doseq [key args]
-      (.deleteBlob blobstore key)
-      (log-message "deleted " key))))
-
-(defn list-cli [args]
-  (with-configured-blob-client blobstore
-    (let [keys (if (empty? args) (iterator-seq (.listKeys blobstore)) args)]
-      (doseq [key keys]
-        (try
-          (let [meta (.getBlobMeta blobstore key)
-                version (.get_version meta)
-                acl (.get_acl (.get_settable meta))]
-            (log-message key " " version " " (pr-str (map access-control-str acl))))
-          (catch AuthorizationException ae
-            (if-not (empty? args) (log-error "ACCESS DENIED to key: " key)))
-          (catch KeyNotFoundException knf
-            (if-not (empty? args) (log-error key " NOT FOUND"))))))))
-
-(defn set-acl-cli [args]
-  (let [[{set-acl :set} [key] _]
-           (cli args ["-s" "--set" :default [] :parse-fn as-acl])]
-    (with-configured-blob-client blobstore
-      (let [meta (.getBlobMeta blobstore key)
-            acl (.get_acl (.get_settable meta))
-            new-acl (if set-acl set-acl acl)
-            new-meta (SettableBlobMeta. new-acl)]
-        (log-message "Setting ACL for " key " to " (pr-str (map access-control-str new-acl)))
-        (.setBlobMeta blobstore key new-meta)))))
-
-(defn rep-cli [args]
-  (let [sub-command (first args)
-        new-args (rest args)]
-    (with-configured-blob-client blobstore
-      (condp = sub-command
-      "--read" (let [key (first new-args)
-                     blob-replication (.getBlobReplication blobstore key)]
-                 (log-message "Current replication factor " blob-replication)
-                 blob-replication)
-      "--update" (let [[{replication-factor :replication-factor} [key] _]
-                        (cli new-args ["-r" "--replication-factor" :parse-fn #(Integer/parseInt %)])]
-                   (if (nil? replication-factor)
-                     (throw (RuntimeException. (str "Please set the replication factor")))
-                     (let [blob-replication (.updateBlobReplication blobstore key replication-factor)]
-                       (log-message "Replication factor is set to " blob-replication)
-                       blob-replication)))
-      :else (throw (RuntimeException. (str sub-command " is not a supported blobstore command")))))))
-
-(defn -main [& args]
-  (let [command (first args)
-        new-args (rest args)]
-    (condp = command
-      "cat" (read-cli new-args)
-      "create" (create-cli new-args)
-      "update" (update-cli new-args)
-      "delete" (delete-cli new-args)
-      "list" (list-cli new-args)
-      "set-acl" (set-acl-cli new-args)
-      "replication" (rep-cli new-args)
-      :else (throw (RuntimeException. (str command " is not a supported blobstore command"))))))
diff --git a/storm-core/src/jvm/org/apache/storm/blobstore/ClientBlobStore.java b/storm-core/src/jvm/org/apache/storm/blobstore/ClientBlobStore.java
index 61e48b3..2cb1cce 100644
--- a/storm-core/src/jvm/org/apache/storm/blobstore/ClientBlobStore.java
+++ b/storm-core/src/jvm/org/apache/storm/blobstore/ClientBlobStore.java
@@ -23,7 +23,9 @@
 import org.apache.storm.generated.SettableBlobMeta;
 import org.apache.storm.generated.KeyAlreadyExistsException;
 import org.apache.storm.generated.KeyNotFoundException;
+import org.apache.storm.utils.ConfigUtils;
 import org.apache.storm.utils.NimbusClient;
+import org.apache.storm.utils.Utils;
 
 import java.util.Iterator;
 import java.util.Map;
@@ -50,6 +52,21 @@
 public abstract class ClientBlobStore implements Shutdownable {
     protected Map conf;
 
+    public interface WithBlobstore {
+        void run(ClientBlobStore blobStore) throws Exception;
+    }
+
+    public static void withConfiguredClient(WithBlobstore withBlobstore) throws Exception {
+        Map<String, Object> conf = ConfigUtils.readStormConfig();
+        ClientBlobStore blobStore = Utils.getClientBlobStore(conf);
+
+        try {
+            withBlobstore.run(blobStore);
+        } finally {
+            blobStore.shutdown();
+        }
+    }
+
     /**
      * Sets up the client API by parsing the configs.
      * @param conf The storm conf containing the config details.
diff --git a/storm-core/src/jvm/org/apache/storm/command/Blobstore.java b/storm-core/src/jvm/org/apache/storm/command/Blobstore.java
new file mode 100644
index 0000000..a5b01fc
--- /dev/null
+++ b/storm-core/src/jvm/org/apache/storm/command/Blobstore.java
@@ -0,0 +1,348 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.command;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.storm.blobstore.AtomicOutputStream;
+import org.apache.storm.blobstore.BlobStoreAclHandler;
+import org.apache.storm.blobstore.ClientBlobStore;
+import org.apache.storm.blobstore.InputStreamWithMeta;
+import org.apache.storm.generated.AccessControl;
+import org.apache.storm.generated.AuthorizationException;
+import org.apache.storm.generated.KeyNotFoundException;
+import org.apache.storm.generated.ReadableBlobMeta;
+import org.apache.storm.generated.SettableBlobMeta;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+public class Blobstore {
+    private static final Logger LOG = LoggerFactory.getLogger(Blobstore.class);
+
+    public static void main(String[] args) throws Exception {
+        if (args.length == 0) {
+            throw new IllegalArgumentException("You should provide command.");
+        }
+
+        String command = args[0];
+        String[] newArgs = Arrays.copyOfRange(args, 1, args.length);
+
+        switch (command) {
+            case "cat":
+                readCli(newArgs);
+                break;
+
+            case "create":
+                createCli(newArgs);
+                break;
+
+            case "update":
+                updateCli(newArgs);
+                break;
+
+            case "delete":
+                deleteCli(newArgs);
+                break;
+
+            case "list":
+                listCli(newArgs);
+                break;
+
+            case "set-acl":
+                setAclCli(newArgs);
+                break;
+
+            case "replication":
+                replicationCli(newArgs);
+                break;
+
+            default:
+                throw new RuntimeException("" + command + " is not a supported blobstore command");
+        }
+    }
+
+    private static void readCli(String[] args) throws Exception {
+        Map<String, Object> cl = CLI.opt("f", "file", null, CLI.AS_STRING)
+                .arg("key", CLI.FIRST_WINS)
+                .parse(args);
+        final String key = (String) cl.get("key");
+        final String file = (String) cl.get("f");
+
+        if (StringUtils.isNotEmpty(file)) {
+            try (BufferedOutputStream f = new BufferedOutputStream(new FileOutputStream(file))) {
+                BlobStoreSupport.readBlob(key, f);
+            }
+        } else {
+            BlobStoreSupport.readBlob(key, System.out);
+        }
+    }
+
+    private static void createCli(String[] args) throws Exception {
+        Map<String, Object> cl = CLI.opt("f", "file", null, CLI.AS_STRING)
+                .opt("a", "acl", Collections.emptyList(), new AsAclParser())
+                .opt("r", "replication-factor", -1, CLI.AS_INT)
+                .arg("key", CLI.FIRST_WINS)
+                .parse(args);
+
+        final String key = (String) cl.get("key");
+        final String file = (String) cl.get("f");
+        final List<AccessControl> acl = (List<AccessControl>) cl.get("a");
+        final Integer replicationFactor = (Integer) cl.get("r");
+
+        SettableBlobMeta meta = new SettableBlobMeta(acl);
+        meta.set_replication_factor(replicationFactor);
+
+        Utils.validateKeyName(key);
+
+        LOG.info("Creating {} with ACL {}", key, generateAccessControlsInfo(acl));
+
+        if (StringUtils.isNotEmpty(file)) {
+            try (BufferedInputStream f = new BufferedInputStream(new FileInputStream(file))) {
+                BlobStoreSupport.createBlobFromStream(key, f, meta);
+            }
+        } else {
+            BlobStoreSupport.createBlobFromStream(key, System.in, meta);
+        }
+
+        LOG.info("Successfully created {}", key);
+    }
+
+    private static void updateCli(String[] args) throws Exception {
+        Map<String, Object> cl = CLI.opt("f", "file", null, CLI.AS_STRING)
+                .arg("key", CLI.FIRST_WINS)
+                .parse(args);
+
+        final String key = (String) cl.get("key");
+        final String file = (String) cl.get("f");
+
+        if (StringUtils.isNotEmpty(file)) {
+            try (BufferedInputStream f = new BufferedInputStream(new FileInputStream(file))) {
+                BlobStoreSupport.updateBlobFromStream(key, f);
+            }
+        } else {
+            BlobStoreSupport.updateBlobFromStream(key, System.in);
+        }
+
+        LOG.info("Successfully updated {}", key);
+    }
+
+    private static void deleteCli(final String[] args) throws Exception {
+        ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+            @Override
+            public void run(ClientBlobStore blobStore) throws Exception {
+                for (String key : args) {
+                    blobStore.deleteBlob(key);
+
+                    LOG.info("deleted {}", key);
+                }
+            }
+        });
+    }
+
+    private static void listCli(final String[] args) throws Exception {
+        ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+            @Override
+            public void run(ClientBlobStore blobStore) throws Exception {
+                Iterator<String> keys;
+                boolean isArgsEmpty = (args == null || args.length == 0);
+                if (isArgsEmpty) {
+                    keys = blobStore.listKeys();
+                } else {
+                    keys = Arrays.asList(args).iterator();
+                }
+
+                while (keys.hasNext()) {
+                    String key = keys.next();
+
+                    try {
+                        ReadableBlobMeta meta = blobStore.getBlobMeta(key);
+                        long version = meta.get_version();
+                        List<AccessControl> acl = meta.get_settable().get_acl();
+
+                        LOG.info("{} {} {}", key, version, generateAccessControlsInfo(acl));
+                    } catch (AuthorizationException ae) {
+                        if (!isArgsEmpty) {
+                            LOG.error("ACCESS DENIED to key: {}", key);
+                        }
+                    } catch (KeyNotFoundException knf) {
+                        if (!isArgsEmpty) {
+                            LOG.error("{} NOT FOUND", key);
+                        }
+                    }
+                }
+            }
+        });
+    }
+
+    private static void setAclCli(String[] args) throws Exception {
+        Map<String, Object> cl = CLI.opt("s", "set", Collections.emptyList(), new AsAclParser())
+                .arg("key", CLI.FIRST_WINS)
+                .parse(args);
+
+        final String key = (String) cl.get("key");
+        final List<AccessControl> setAcl = (List<AccessControl>) cl.get("s");
+
+        ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+            @Override
+            public void run(ClientBlobStore blobStore) throws Exception {
+                ReadableBlobMeta meta = blobStore.getBlobMeta(key);
+                List<AccessControl> acl = meta.get_settable().get_acl();
+                List<AccessControl> newAcl;
+                if (setAcl != null && !setAcl.isEmpty()) {
+                    newAcl = setAcl;
+                } else {
+                    newAcl = acl;
+                }
+
+                SettableBlobMeta newMeta = new SettableBlobMeta(newAcl);
+                LOG.info("Setting ACL for {} to {}", key, generateAccessControlsInfo(newAcl));
+                blobStore.setBlobMeta(key, newMeta);
+            }
+        });
+    }
+
+    private static void replicationCli(String[] args) throws Exception {
+        if (args.length == 0) {
+            throw new IllegalArgumentException("replication command needs at least subcommand as parameter.");
+        }
+        final String subCommand = args[0];
+        final String[] newArgs = Arrays.copyOfRange(args, 1, args.length);
+
+        ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+            @Override
+            public void run(ClientBlobStore blobStore) throws Exception {
+                switch (subCommand) {
+                    case "--read":
+                        if (newArgs.length == 0) {
+                            throw new IllegalArgumentException("replication --read needs key as parameter.");
+                        }
+
+                        String key = newArgs[0];
+                        int blobReplication = blobStore.getBlobReplication(key);
+                        LOG.info("Current replication factor {}", blobReplication);
+                        break;
+
+                    case "--update":
+                        updateReplicationFactor(blobStore, newArgs);
+                        break;
+
+                    default:
+                        throw new RuntimeException("" + subCommand + " is not a supported blobstore command");
+                }
+            }
+
+            private void updateReplicationFactor(ClientBlobStore blobStore, String[] args) throws Exception {
+                Map<String, Object> cl = CLI.opt("r", "replication-factor", null, CLI.AS_INT)
+                        .arg("key", CLI.FIRST_WINS)
+                        .parse(args);
+
+                final String key = (String) cl.get("key");
+                final Integer replicationFactor = (Integer) cl.get("r");
+
+                if (replicationFactor == null) {
+                    throw new RuntimeException("Please set the replication factor");
+                }
+
+                int blobReplication = blobStore.updateBlobReplication(key, replicationFactor);
+                LOG.info("Replication factor is set to {}", blobReplication);
+            }
+        });
+    }
+
+    private static final class BlobStoreSupport {
+        static void readBlob(final String key, final OutputStream os) throws Exception {
+            ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+                @Override
+                public void run(ClientBlobStore blobStore) throws Exception {
+                    try (InputStreamWithMeta is = blobStore.getBlob(key)) {
+                        IOUtils.copy(is, os);
+                    }
+                }
+            });
+        }
+
+        static void createBlobFromStream(final String key, final InputStream is, final SettableBlobMeta meta) throws Exception {
+            ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+                @Override
+                public void run(ClientBlobStore blobStore) throws Exception {
+                    AtomicOutputStream os = blobStore.createBlob(key, meta);
+                    copyInputStreamToBlobOutputStream(is, os);
+                }
+            });
+        }
+
+        static void updateBlobFromStream(final String key, final InputStream is) throws Exception {
+            ClientBlobStore.withConfiguredClient(new ClientBlobStore.WithBlobstore() {
+                @Override
+                public void run(ClientBlobStore blobStore) throws Exception {
+                    AtomicOutputStream os = blobStore.updateBlob(key);
+                    copyInputStreamToBlobOutputStream(is, os);
+                }
+            });
+        }
+
+        static void copyInputStreamToBlobOutputStream(InputStream is, AtomicOutputStream os) throws IOException {
+            try {
+                IOUtils.copy(is, os);
+                os.close();
+            } catch (Exception e) {
+                os.cancel();
+                throw e;
+            }
+        }
+    }
+
+    private static List<String> generateAccessControlsInfo(List<AccessControl> acl) {
+        List<String> accessControlStrings = new ArrayList<>();
+        for (AccessControl ac : acl) {
+            accessControlStrings.add(BlobStoreAclHandler.accessControlToString(ac));
+        }
+        return accessControlStrings;
+    }
+
+    private static final class AsAclParser implements CLI.Parse {
+        @Override
+        public Object parse(String value) {
+            List<AccessControl> accessControls = new ArrayList<>();
+            for (String part : value.split(",")) {
+                accessControls.add(asAccessControl(part));
+            }
+
+            return accessControls;
+        }
+
+        private AccessControl asAccessControl(String param) {
+            return BlobStoreAclHandler.parseAccessControl(param);
+        }
+    }
+}