diff --git a/utilities/persistent_cache/block_cache_tier.cc b/utilities/persistent_cache/block_cache_tier.cc
index 49c41c38b..768346444 100644
--- a/utilities/persistent_cache/block_cache_tier.cc
+++ b/utilities/persistent_cache/block_cache_tier.cc
@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "util/stop_watch.h"
+#include "util/sync_point.h"
 #include "utilities/persistent_cache/block_cache_tier_file.h"
 
 namespace rocksdb {
@@ -54,8 +55,15 @@ Status BlockCacheTier::Open() {
     }
   }
 
+  // create a new file
   assert(!cache_file_);
-  NewCacheFile();
+  status = NewCacheFile();
+  if (!status.ok()) {
+    Error(opt_.log, "Error creating new file %s. %s", opt_.path.c_str(),
+          status.ToString().c_str());
+    return status;
+  }
+
   assert(cache_file_);
 
   if (opt_.pipeline_writes) {
@@ -231,7 +239,10 @@ Status BlockCacheTier::InsertImpl(const Slice& key, const Slice& data) {
     }
 
     assert(cache_file_->Eof());
-    NewCacheFile();
+    Status status = NewCacheFile();
+    if (!status.ok()) {
+      return status;
+    }
   }
 
   // Insert into lookup index
@@ -280,7 +291,6 @@ Status BlockCacheTier::Lookup(const Slice& key, unique_ptr<char[]>* val,
 
   status = file->Read(lba, &blk_key, &blk_val, scratch.get());
   --file->refs_;
-  assert(status);
   if (!status) {
     stats_.cache_misses_++;
     stats_.cache_errors_++;
@@ -309,25 +319,36 @@ bool BlockCacheTier::Erase(const Slice& key) {
   return true;
 }
 
-void BlockCacheTier::NewCacheFile() {
+Status BlockCacheTier::NewCacheFile() {
   lock_.AssertHeld();
 
-  Info(opt_.log, "Creating cache file %d", writer_cache_id_);
+  TEST_SYNC_POINT_CALLBACK("BlockCacheTier::NewCacheFile:DeleteDir",
+                           (void*)(GetCachePath().c_str()));
+
+  std::unique_ptr<WriteableCacheFile> f(
+    new WriteableCacheFile(opt_.env, &buffer_allocator_, &writer_,
+                           GetCachePath(), writer_cache_id_,
+                           opt_.cache_file_size, opt_.log));
+
+  bool status = f->Create(opt_.enable_direct_writes, opt_.enable_direct_reads);
+  if (!status) {
+    return Status::IOError("Error creating file");
+  }
+
+  Info(opt_.log, "Created cache file %d", writer_cache_id_);
 
   writer_cache_id_++;
-
-  cache_file_ = new WriteableCacheFile(opt_.env, &buffer_allocator_, &writer_,
-                                       GetCachePath(), writer_cache_id_,
-                                       opt_.cache_file_size, opt_.log);
-  bool status;
-  status =
-      cache_file_->Create(opt_.enable_direct_writes, opt_.enable_direct_reads);
-  assert(status);
+  cache_file_ = f.release();
 
   // insert to cache files tree
   status = metadata_.Insert(cache_file_);
-  (void)status;
   assert(status);
+  if (!status) {
+    Error(opt_.log, "Error inserting to metadata");
+    return Status::IOError("Error inserting to metadata");
+  }
+
+  return Status::OK();
 }
 
 bool BlockCacheTier::Reserve(const size_t size) {
diff --git a/utilities/persistent_cache/block_cache_tier.h b/utilities/persistent_cache/block_cache_tier.h
index 257024760..a33dc369a 100644
--- a/utilities/persistent_cache/block_cache_tier.h
+++ b/utilities/persistent_cache/block_cache_tier.h
@@ -103,7 +103,7 @@ class BlockCacheTier : public PersistentCacheTier {
   // insert implementation
   Status InsertImpl(const Slice& key, const Slice& data);
   // Create a new cache file
-  void NewCacheFile();
+  Status NewCacheFile();
   // Get cache directory path
   std::string GetCachePath() const { return opt_.path + "/cache"; }
   // Cleanup folder
diff --git a/utilities/persistent_cache/block_cache_tier_file.cc b/utilities/persistent_cache/block_cache_tier_file.cc
index 0ea0345ad..a24321bd2 100644
--- a/utilities/persistent_cache/block_cache_tier_file.cc
+++ b/utilities/persistent_cache/block_cache_tier_file.cc
@@ -216,7 +216,10 @@ bool RandomAccessCacheFile::Read(const LBA& lba, Slice* key, Slice* val,
   ReadLock _(&rwlock_);
 
   assert(lba.cache_id_ == cache_id_);
-  assert(file_);
+
+  if (!file_) {
+    return false;
+  }
 
   Slice result;
   Status s = file_->Read(lba.off_, lba.size_, &result, scratch);
@@ -259,11 +262,12 @@ WriteableCacheFile::~WriteableCacheFile() {
     // This file never flushed. We give priority to shutdown since this is a
     // cache
     // TODO(krad): Figure a way to flush the pending data
-    assert(file_);
-
-    assert(refs_ == 1);
-    --refs_;
+    if (file_) {
+      assert(refs_ == 1);
+      --refs_;
+    }
   }
+  assert(!refs_);
   ClearBuffers();
 }
 
diff --git a/utilities/persistent_cache/persistent_cache_test.cc b/utilities/persistent_cache/persistent_cache_test.cc
index 42ef6bde1..2908a2c5a 100644
--- a/utilities/persistent_cache/persistent_cache_test.cc
+++ b/utilities/persistent_cache/persistent_cache_test.cc
@@ -38,6 +38,32 @@ static void OnOpenForWrite(void* arg) {
 }
 #endif
 
+static void RemoveDirectory(const std::string& folder) {
+  std::vector<std::string> files;
+  Status status = Env::Default()->GetChildren(folder, &files);
+  if (!status.ok()) {
+    // we assume the directory does not exist
+    return;
+  }
+
+  // cleanup files with the patter :digi:.rc
+  for (auto file : files) {
+    if (file == "." || file == "..") {
+      continue;
+    }
+    status = Env::Default()->DeleteFile(folder + "/" + file);
+    assert(status.ok());
+  }
+
+  status = Env::Default()->DeleteDir(folder);
+  assert(status.ok());
+}
+
+static void OnDeleteDir(void* arg) {
+  char* dir = static_cast<char*>(arg);
+  RemoveDirectory(std::string(dir));
+}
+
 //
 // Simple logger that prints message on stdout
 //
@@ -116,6 +142,21 @@ PersistentCacheTierTest::PersistentCacheTierTest()
 #endif
 }
 
+// Block cache tests
+TEST_F(PersistentCacheTierTest, BlockCacheInsertWithFileCreateError) {
+  cache_ = NewBlockCache(Env::Default(), path_,
+                         /*size=*/std::numeric_limits<uint64_t>::max(),
+                         /*direct_writes=*/ false);
+  rocksdb::SyncPoint::GetInstance()->SetCallBack( 
+    "BlockCacheTier::NewCacheFile:DeleteDir", OnDeleteDir);
+
+  RunNegativeInsertTest(/*nthreads=*/ 1,
+                        /*max_keys*/
+                          static_cast<size_t>(10 * 1024 * kStressFactor));
+
+  rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 #ifdef TRAVIS
 // Travis is unable to handle the normal version of the tests running out of
 // fds, out of space and timeouts. This is an easier version of the test
diff --git a/utilities/persistent_cache/persistent_cache_test.h b/utilities/persistent_cache/persistent_cache_test.h
index 184fbf1ba..9991c0761 100644
--- a/utilities/persistent_cache/persistent_cache_test.h
+++ b/utilities/persistent_cache/persistent_cache_test.h
@@ -132,7 +132,12 @@ class PersistentCacheTierTest : public testing::Test {
       memset(data, '0' + (i % 10), sizeof(data));
       auto k = prefix + PaddedNumber(i, /*count=*/8);
       Slice key(k);
-      while (!cache_->Insert(key, data, sizeof(data)).ok()) {
+      while (true) {
+        Status status = cache_->Insert(key, data, sizeof(data));
+        if (status.ok()) {
+          break;
+        }
+        ASSERT_TRUE(status.IsTryAgain());
         Env::Default()->SleepForMicroseconds(1 * 1000 * 1000);
       }
     }
@@ -180,6 +185,17 @@ class PersistentCacheTierTest : public testing::Test {
     cache_.reset();
   }
 
+  // template for negative insert test
+  void RunNegativeInsertTest(const size_t nthreads, const size_t max_keys) {
+    Insert(nthreads, max_keys);
+    Verify(nthreads, /*eviction_enabled=*/true);
+    ASSERT_LT(stats_verify_hits_, max_keys);
+    ASSERT_GT(stats_verify_missed_, 0);
+
+    cache_->Close();
+    cache_.reset();
+  }
+
   // template for insert with eviction test
   void RunInsertTestWithEviction(const size_t nthreads, const size_t max_keys) {
     Insert(nthreads, max_keys);