storm-core/src/jvm/org/apache/storm/blobstore/LocalFsBlobStore.java - storm - Git at Google

 /**
  * 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.blobstore;

 import org.apache.storm.Config;
 import org.apache.storm.generated.SettableBlobMeta;
 import org.apache.storm.generated.AuthorizationException;
 import org.apache.storm.generated.KeyAlreadyExistsException;
 import org.apache.storm.generated.KeyNotFoundException;
 import org.apache.storm.generated.ReadableBlobMeta;

 import org.apache.storm.nimbus.NimbusInfo;
 import org.apache.storm.utils.ConfigUtils;
 import org.apache.storm.utils.Utils;
 import org.apache.curator.framework.CuratorFramework;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import javax.security.auth.Subject;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
 import java.io.IOException;
 import java.io.FileNotFoundException;
 import java.io.InputStream;

 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;;

 import static org.apache.storm.blobstore.BlobStoreAclHandler.ADMIN;
 import static org.apache.storm.blobstore.BlobStoreAclHandler.READ;
 import static org.apache.storm.blobstore.BlobStoreAclHandler.WRITE;

 import com.google.common.annotations.VisibleForTesting;

 /**
  * Provides a local file system backed blob store implementation for Nimbus.
  *
  * For a local blob store the user and the supervisor use NimbusBlobStore Client API in order to talk to nimbus through thrift.
  * The authentication and authorization here is based on the subject.
  * We currently have NIMBUS_ADMINS and SUPERVISOR_ADMINS configuration. NIMBUS_ADMINS are given READ, WRITE and ADMIN
  * access whereas the SUPERVISOR_ADMINS are given READ access in order to read and download the blobs form the nimbus.
  *
  * The ACLs for the blob store are validated against whether the subject is a NIMBUS_ADMIN, SUPERVISOR_ADMIN or USER
  * who has read, write or admin privileges in order to perform respective operations on the blob.
  *
  * For local blob store
  * 1. The USER interacts with nimbus to upload and access blobs through NimbusBlobStore Client API.
  * 2. The USER sets the ACLs, and the blob access is validated against these ACLs.
  * 3. The SUPERVISOR interacts with nimbus through the NimbusBlobStore Client API to download the blobs.
  * The supervisors principal should match the set of users configured into SUPERVISOR_ADMINS.
  * Here, the PrincipalToLocalPlugin takes care of mapping the principal to user name before the ACL validation.
  */
 public class LocalFsBlobStore extends BlobStore {
     public static final Logger LOG = LoggerFactory.getLogger(LocalFsBlobStore.class);
     private static final String DATA_PREFIX = "data_";
     private static final String META_PREFIX = "meta_";
     protected BlobStoreAclHandler _aclHandler;
     private final String BLOBSTORE_SUBTREE = "/blobstore/";
     private NimbusInfo nimbusInfo;
     private FileBlobStoreImpl fbs;
     private final int allPermissions = READ | WRITE | ADMIN;
     private Map conf;
     private CuratorFramework zkClient;

     @Override
     public void prepare(Map conf, String overrideBase, NimbusInfo nimbusInfo) {
         this.conf = conf;
         this.nimbusInfo = nimbusInfo;
         zkClient = BlobStoreUtils.createZKClient(conf);
         if (overrideBase == null) {
             overrideBase = ConfigUtils.absoluteStormBlobStoreDir(conf);
         }
         File baseDir = new File(overrideBase, BASE_BLOBS_DIR_NAME);
         try {
             fbs = new FileBlobStoreImpl(baseDir, conf);
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
         _aclHandler = new BlobStoreAclHandler(conf);
     }

     @Override
     public AtomicOutputStream createBlob(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyAlreadyExistsException {
         LOG.debug("Creating Blob for key {}", key);
         validateKey(key);
         _aclHandler.normalizeSettableBlobMeta(key, meta, who, allPermissions);
         BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
         _aclHandler.hasPermissions(meta.get_acl(), allPermissions, who, key);
         if (fbs.exists(DATA_PREFIX+key)) {
             throw new KeyAlreadyExistsException(key);
         }
         BlobStoreFileOutputStream mOut = null;
         try {
             mOut = new BlobStoreFileOutputStream(fbs.write(META_PREFIX+key, true));
             mOut.write(Utils.thriftSerialize(meta));
             mOut.close();
             mOut = null;
             return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX+key, true));
         } catch (IOException e) {
             throw new RuntimeException(e);
         } finally {
             if (mOut != null) {
                 try {
                     mOut.cancel();
                 } catch (IOException e) {
                     //Ignored
                 }
             }
         }
     }

     @Override
     public AtomicOutputStream updateBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
         validateKey(key);
         checkPermission(key, who, WRITE);
         try {
             return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX+key, false));
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }

     private SettableBlobMeta getStoredBlobMeta(String key) throws KeyNotFoundException {
         InputStream in = null;
         try {
             LocalFsBlobStoreFile pf = fbs.read(META_PREFIX+key);
             try {
                 in = pf.getInputStream();
             } catch (FileNotFoundException fnf) {
                 throw new KeyNotFoundException(key);
             }
             ByteArrayOutputStream out = new ByteArrayOutputStream();
             byte [] buffer = new byte[2048];
             int len;
             while ((len = in.read(buffer)) > 0) {
                 out.write(buffer, 0, len);
             }
             in.close();
             in = null;
             return Utils.thriftDeserialize(SettableBlobMeta.class, out.toByteArray());
         } catch (IOException e) {
             throw new RuntimeException(e);
         } finally {
             if (in != null) {
                 try {
                     in.close();
                 } catch (IOException e) {
                     //Ignored
                 }
             }
         }
     }

     @Override
     public ReadableBlobMeta getBlobMeta(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
         validateKey(key);
         if(!checkForBlobOrDownload(key)) {
             checkForBlobUpdate(key);
         }
         SettableBlobMeta meta = getStoredBlobMeta(key);
         _aclHandler.validateUserCanReadMeta(meta.get_acl(), who, key);
         ReadableBlobMeta rbm = new ReadableBlobMeta();
         rbm.set_settable(meta);
         try {
             LocalFsBlobStoreFile pf = fbs.read(DATA_PREFIX+key);
             rbm.set_version(pf.getModTime());
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
         return rbm;
     }

     @Override
     public void setBlobMeta(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyNotFoundException {
         validateKey(key);
         checkForBlobOrDownload(key);
         _aclHandler.normalizeSettableBlobMeta(key, meta, who, ADMIN);
         BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
         SettableBlobMeta orig = getStoredBlobMeta(key);
         _aclHandler.hasPermissions(orig.get_acl(), ADMIN, who, key);
         BlobStoreFileOutputStream mOut = null;
         try {
             mOut = new BlobStoreFileOutputStream(fbs.write(META_PREFIX+key, false));
             mOut.write(Utils.thriftSerialize(meta));
             mOut.close();
             mOut = null;
         } catch (IOException e) {
             throw new RuntimeException(e);
         } finally {
             if (mOut != null) {
                 try {
                     mOut.cancel();
                 } catch (IOException e) {
                     //Ignored
                 }
             }
         }
     }

     @Override
     public void deleteBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
         validateKey(key);

         if (!_aclHandler.checkForValidUsers(who, WRITE)) {
             // need to get ACL from meta
             LOG.debug("Retrieving meta to get ACL info... key: {} subject: {}", key, who);

             try {
                 checkPermission(key, who, WRITE);
             } catch (KeyNotFoundException e) {
                 LOG.error("Error while retrieving meta from ZK or local... key: {} subject: {}", key, who);
                 throw e;
             }
         } else {
             // able to delete the blob without checking meta's ACL
             // skip checking everything and continue deleting local files
             LOG.debug("Given subject is eligible to delete key without checking ACL, skipping... key: {} subject: {}",
                     key, who);
         }

         try {
             deleteKeyIgnoringFileNotFound(DATA_PREFIX + key);
             deleteKeyIgnoringFileNotFound(META_PREFIX + key);
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }

     private void checkPermission(String key, Subject who, int mask) throws KeyNotFoundException, AuthorizationException {
         checkForBlobOrDownload(key);
         SettableBlobMeta meta = getStoredBlobMeta(key);
         _aclHandler.hasPermissions(meta.get_acl(), mask, who, key);
     }

     private void deleteKeyIgnoringFileNotFound(String key) throws IOException {
         try {
             fbs.deleteKey(key);
         } catch (IOException e) {
             if (e instanceof FileNotFoundException) {
                 LOG.debug("Ignoring FileNotFoundException since we're about to delete such key... key: {}", key);
             } else {
                 throw e;
             }
         }
     }

     @Override
     public InputStreamWithMeta getBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
         validateKey(key);
         if(!checkForBlobOrDownload(key)) {
             checkForBlobUpdate(key);
         }
         SettableBlobMeta meta = getStoredBlobMeta(key);
         _aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
         try {
             return new BlobStoreFileInputStream(fbs.read(DATA_PREFIX+key));
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }

     @Override
     public Iterator<String> listKeys() {
         try {
             return new KeyTranslationIterator(fbs.listKeys(), DATA_PREFIX);
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
     }

     @Override
     public void shutdown() {
         if (zkClient != null) {
             zkClient.close();
         }
     }

     @Override
     public int getBlobReplication(String key, Subject who) throws Exception {
         int replicationCount = 0;
         validateKey(key);
         SettableBlobMeta meta = getStoredBlobMeta(key);
         _aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
         if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) == null) {
             return 0;
         }
         replicationCount = zkClient.getChildren().forPath(BLOBSTORE_SUBTREE + key).size();
         return replicationCount;
     }

     @Override
     public int updateBlobReplication(String key, int replication, Subject who) throws AuthorizationException, KeyNotFoundException {
         throw new UnsupportedOperationException("For local file system blob store the update blobs function does not work. " +
                 "Please use HDFS blob store to make this feature available.");
     }

     //This additional check and download is for nimbus high availability in case you have more than one nimbus
     public synchronized boolean checkForBlobOrDownload(String key) throws KeyNotFoundException {
         boolean checkBlobDownload = false;
         try {
             List<String> keyList = BlobStoreUtils.getKeyListFromBlobStore(this);
             if (!keyList.contains(key)) {
                 if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) != null) {
                     Set<NimbusInfo> nimbusSet = BlobStoreUtils.getNimbodesWithLatestSequenceNumberOfBlob(zkClient, key);
                     if (BlobStoreUtils.downloadMissingBlob(conf, this, key, nimbusSet)) {
                         LOG.debug("Updating blobs state");
                         BlobStoreUtils.createStateInZookeeper(conf, key, nimbusInfo);
                         checkBlobDownload = true;
                     }
                 }
             }
         } catch (KeyNotFoundException e) {
             throw e;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
         return checkBlobDownload;
     }

     public synchronized void checkForBlobUpdate(String key) {
         BlobStoreUtils.updateKeyForBlobStore(conf, this, zkClient, key, nimbusInfo);
     }

     public void fullCleanup(long age) throws IOException {
         fbs.fullCleanup(age);
     }

     @VisibleForTesting
     File getKeyDataDir(String key) {
         return fbs.getKeyDir(DATA_PREFIX + key);
     }
 }
	/**
	* 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.blobstore;

	import org.apache.storm.Config;
	import org.apache.storm.generated.SettableBlobMeta;
	import org.apache.storm.generated.AuthorizationException;
	import org.apache.storm.generated.KeyAlreadyExistsException;
	import org.apache.storm.generated.KeyNotFoundException;
	import org.apache.storm.generated.ReadableBlobMeta;

	import org.apache.storm.nimbus.NimbusInfo;
	import org.apache.storm.utils.ConfigUtils;
	import org.apache.storm.utils.Utils;
	import org.apache.curator.framework.CuratorFramework;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import javax.security.auth.Subject;
	import java.io.ByteArrayOutputStream;
	import java.io.File;
	import java.io.IOException;
	import java.io.FileNotFoundException;
	import java.io.InputStream;

	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;;

	import static org.apache.storm.blobstore.BlobStoreAclHandler.ADMIN;
	import static org.apache.storm.blobstore.BlobStoreAclHandler.READ;
	import static org.apache.storm.blobstore.BlobStoreAclHandler.WRITE;

	import com.google.common.annotations.VisibleForTesting;

	/**
	* Provides a local file system backed blob store implementation for Nimbus.
	*
	* For a local blob store the user and the supervisor use NimbusBlobStore Client API in order to talk to nimbus through thrift.
	* The authentication and authorization here is based on the subject.
	* We currently have NIMBUS_ADMINS and SUPERVISOR_ADMINS configuration. NIMBUS_ADMINS are given READ, WRITE and ADMIN
	* access whereas the SUPERVISOR_ADMINS are given READ access in order to read and download the blobs form the nimbus.
	*
	* The ACLs for the blob store are validated against whether the subject is a NIMBUS_ADMIN, SUPERVISOR_ADMIN or USER
	* who has read, write or admin privileges in order to perform respective operations on the blob.
	*
	* For local blob store
	* 1. The USER interacts with nimbus to upload and access blobs through NimbusBlobStore Client API.
	* 2. The USER sets the ACLs, and the blob access is validated against these ACLs.
	* 3. The SUPERVISOR interacts with nimbus through the NimbusBlobStore Client API to download the blobs.
	* The supervisors principal should match the set of users configured into SUPERVISOR_ADMINS.
	* Here, the PrincipalToLocalPlugin takes care of mapping the principal to user name before the ACL validation.
	*/
	public class LocalFsBlobStore extends BlobStore {
	public static final Logger LOG = LoggerFactory.getLogger(LocalFsBlobStore.class);
	private static final String DATA_PREFIX = "data_";
	private static final String META_PREFIX = "meta_";
	protected BlobStoreAclHandler _aclHandler;
	private final String BLOBSTORE_SUBTREE = "/blobstore/";
	private NimbusInfo nimbusInfo;
	private FileBlobStoreImpl fbs;
	private final int allPermissions = READ \| WRITE \| ADMIN;
	private Map conf;
	private CuratorFramework zkClient;

	@Override
	public void prepare(Map conf, String overrideBase, NimbusInfo nimbusInfo) {
	this.conf = conf;
	this.nimbusInfo = nimbusInfo;
	zkClient = BlobStoreUtils.createZKClient(conf);
	if (overrideBase == null) {
	overrideBase = ConfigUtils.absoluteStormBlobStoreDir(conf);
	}
	File baseDir = new File(overrideBase, BASE_BLOBS_DIR_NAME);
	try {
	fbs = new FileBlobStoreImpl(baseDir, conf);
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	_aclHandler = new BlobStoreAclHandler(conf);
	}

	@Override
	public AtomicOutputStream createBlob(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyAlreadyExistsException {
	LOG.debug("Creating Blob for key {}", key);
	validateKey(key);
	_aclHandler.normalizeSettableBlobMeta(key, meta, who, allPermissions);
	BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
	_aclHandler.hasPermissions(meta.get_acl(), allPermissions, who, key);
	if (fbs.exists(DATA_PREFIX+key)) {
	throw new KeyAlreadyExistsException(key);
	}
	BlobStoreFileOutputStream mOut = null;
	try {
	mOut = new BlobStoreFileOutputStream(fbs.write(META_PREFIX+key, true));
	mOut.write(Utils.thriftSerialize(meta));
	mOut.close();
	mOut = null;
	return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX+key, true));
	} catch (IOException e) {
	throw new RuntimeException(e);
	} finally {
	if (mOut != null) {
	try {
	mOut.cancel();
	} catch (IOException e) {
	//Ignored
	}
	}
	}
	}

	@Override
	public AtomicOutputStream updateBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
	validateKey(key);
	checkPermission(key, who, WRITE);
	try {
	return new BlobStoreFileOutputStream(fbs.write(DATA_PREFIX+key, false));
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}

	private SettableBlobMeta getStoredBlobMeta(String key) throws KeyNotFoundException {
	InputStream in = null;
	try {
	LocalFsBlobStoreFile pf = fbs.read(META_PREFIX+key);
	try {
	in = pf.getInputStream();
	} catch (FileNotFoundException fnf) {
	throw new KeyNotFoundException(key);
	}
	ByteArrayOutputStream out = new ByteArrayOutputStream();
	byte [] buffer = new byte[2048];
	int len;
	while ((len = in.read(buffer)) > 0) {
	out.write(buffer, 0, len);
	}
	in.close();
	in = null;
	return Utils.thriftDeserialize(SettableBlobMeta.class, out.toByteArray());
	} catch (IOException e) {
	throw new RuntimeException(e);
	} finally {
	if (in != null) {
	try {
	in.close();
	} catch (IOException e) {
	//Ignored
	}
	}
	}
	}

	@Override
	public ReadableBlobMeta getBlobMeta(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
	validateKey(key);
	if(!checkForBlobOrDownload(key)) {
	checkForBlobUpdate(key);
	}
	SettableBlobMeta meta = getStoredBlobMeta(key);
	_aclHandler.validateUserCanReadMeta(meta.get_acl(), who, key);
	ReadableBlobMeta rbm = new ReadableBlobMeta();
	rbm.set_settable(meta);
	try {
	LocalFsBlobStoreFile pf = fbs.read(DATA_PREFIX+key);
	rbm.set_version(pf.getModTime());
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	return rbm;
	}

	@Override
	public void setBlobMeta(String key, SettableBlobMeta meta, Subject who) throws AuthorizationException, KeyNotFoundException {
	validateKey(key);
	checkForBlobOrDownload(key);
	_aclHandler.normalizeSettableBlobMeta(key, meta, who, ADMIN);
	BlobStoreAclHandler.validateSettableACLs(key, meta.get_acl());
	SettableBlobMeta orig = getStoredBlobMeta(key);
	_aclHandler.hasPermissions(orig.get_acl(), ADMIN, who, key);
	BlobStoreFileOutputStream mOut = null;
	try {
	mOut = new BlobStoreFileOutputStream(fbs.write(META_PREFIX+key, false));
	mOut.write(Utils.thriftSerialize(meta));
	mOut.close();
	mOut = null;
	} catch (IOException e) {
	throw new RuntimeException(e);
	} finally {
	if (mOut != null) {
	try {
	mOut.cancel();
	} catch (IOException e) {
	//Ignored
	}
	}
	}
	}

	@Override
	public void deleteBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
	validateKey(key);

	if (!_aclHandler.checkForValidUsers(who, WRITE)) {
	// need to get ACL from meta
	LOG.debug("Retrieving meta to get ACL info... key: {} subject: {}", key, who);

	try {
	checkPermission(key, who, WRITE);
	} catch (KeyNotFoundException e) {
	LOG.error("Error while retrieving meta from ZK or local... key: {} subject: {}", key, who);
	throw e;
	}
	} else {
	// able to delete the blob without checking meta's ACL
	// skip checking everything and continue deleting local files
	LOG.debug("Given subject is eligible to delete key without checking ACL, skipping... key: {} subject: {}",
	key, who);
	}

	try {
	deleteKeyIgnoringFileNotFound(DATA_PREFIX + key);
	deleteKeyIgnoringFileNotFound(META_PREFIX + key);
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}

	private void checkPermission(String key, Subject who, int mask) throws KeyNotFoundException, AuthorizationException {
	checkForBlobOrDownload(key);
	SettableBlobMeta meta = getStoredBlobMeta(key);
	_aclHandler.hasPermissions(meta.get_acl(), mask, who, key);
	}

	private void deleteKeyIgnoringFileNotFound(String key) throws IOException {
	try {
	fbs.deleteKey(key);
	} catch (IOException e) {
	if (e instanceof FileNotFoundException) {
	LOG.debug("Ignoring FileNotFoundException since we're about to delete such key... key: {}", key);
	} else {
	throw e;
	}
	}
	}

	@Override
	public InputStreamWithMeta getBlob(String key, Subject who) throws AuthorizationException, KeyNotFoundException {
	validateKey(key);
	if(!checkForBlobOrDownload(key)) {
	checkForBlobUpdate(key);
	}
	SettableBlobMeta meta = getStoredBlobMeta(key);
	_aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
	try {
	return new BlobStoreFileInputStream(fbs.read(DATA_PREFIX+key));
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}

	@Override
	public Iterator<String> listKeys() {
	try {
	return new KeyTranslationIterator(fbs.listKeys(), DATA_PREFIX);
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}

	@Override
	public void shutdown() {
	if (zkClient != null) {
	zkClient.close();
	}
	}

	@Override
	public int getBlobReplication(String key, Subject who) throws Exception {
	int replicationCount = 0;
	validateKey(key);
	SettableBlobMeta meta = getStoredBlobMeta(key);
	_aclHandler.hasPermissions(meta.get_acl(), READ, who, key);
	if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) == null) {
	return 0;
	}
	replicationCount = zkClient.getChildren().forPath(BLOBSTORE_SUBTREE + key).size();
	return replicationCount;
	}

	@Override
	public int updateBlobReplication(String key, int replication, Subject who) throws AuthorizationException, KeyNotFoundException {
	throw new UnsupportedOperationException("For local file system blob store the update blobs function does not work. " +
	"Please use HDFS blob store to make this feature available.");
	}

	//This additional check and download is for nimbus high availability in case you have more than one nimbus
	public synchronized boolean checkForBlobOrDownload(String key) throws KeyNotFoundException {
	boolean checkBlobDownload = false;
	try {
	List<String> keyList = BlobStoreUtils.getKeyListFromBlobStore(this);
	if (!keyList.contains(key)) {
	if (zkClient.checkExists().forPath(BLOBSTORE_SUBTREE + key) != null) {
	Set<NimbusInfo> nimbusSet = BlobStoreUtils.getNimbodesWithLatestSequenceNumberOfBlob(zkClient, key);
	if (BlobStoreUtils.downloadMissingBlob(conf, this, key, nimbusSet)) {
	LOG.debug("Updating blobs state");
	BlobStoreUtils.createStateInZookeeper(conf, key, nimbusInfo);
	checkBlobDownload = true;
	}
	}
	}
	} catch (KeyNotFoundException e) {
	throw e;
	} catch (Exception e) {
	throw new RuntimeException(e);
	}
	return checkBlobDownload;
	}

	public synchronized void checkForBlobUpdate(String key) {
	BlobStoreUtils.updateKeyForBlobStore(conf, this, zkClient, key, nimbusInfo);
	}

	public void fullCleanup(long age) throws IOException {
	fbs.fullCleanup(age);
	}

	@VisibleForTesting
	File getKeyDataDir(String key) {
	return fbs.getKeyDir(DATA_PREFIX + key);
	}
	}