code: private void setupBlobstore() throws Exception {
          BlobStore blobStore = data.getBlobStore();
          StormClusterState clusterState = data.getStormClusterState();
          Set<String> localSetOfKeys = Sets.newHashSet(blobStore.listKeys());
          Set<String> allKeys = Sets.newHashSet(clusterState.active_keys());
          Set<String> localAvailableActiveKeys = Sets.intersection(localSetOfKeys, allKeys);
          // keys on local but not on zk, we will delete it
          Set<String> keysToDelete = Sets.difference(localSetOfKeys, allKeys);
          LOG.debug("deleting keys not on zookeeper {}", keysToDelete);
          for (String key : keysToDelete) {
              blobStore.deleteBlob(key);
          }
          LOG.debug("Creating list of key entries for blobstore inside zookeeper {} local {}",
                  allKeys, localAvailableActiveKeys);
          for (String key : localAvailableActiveKeys) {
              int versionForKey = BlobStoreUtils.getVersionForKey(key, data.getNimbusHostPortInfo(), data.getConf());
              clusterState.setup_blobstore(key, data.getNimbusHostPortInfo(), versionForKey);
          }
      }
docs: sets up blobstore state for all current keys