diff --git a/bolt/common/caching/AsyncDataCache.cpp b/bolt/common/caching/AsyncDataCache.cpp
index afa4d7002..72460faef 100644
--- a/bolt/common/caching/AsyncDataCache.cpp
+++ b/bolt/common/caching/AsyncDataCache.cpp
@@ -682,6 +682,9 @@ void AsyncDataCache::shutdown() {
   for (auto& shard : shards_) {
     shard->shutdown();
   }
+  if (ssdCache_) {
+    ssdCache_->shutdown();
+  }
 }
 
 void CacheShard::shutdown() {
diff --git a/bolt/common/caching/SsdCache.cpp b/bolt/common/caching/SsdCache.cpp
index 3dcd461e5..376246415 100644
--- a/bolt/common/caching/SsdCache.cpp
+++ b/bolt/common/caching/SsdCache.cpp
@@ -61,7 +61,7 @@ SsdCache::SsdCache(
       "Ssd path '{}' does not start with '/' that points to local file system.",
       filePrefix_);
   filesystems::getFileSystem(filePrefix_, nullptr)
-      ->mkdir(std::filesystem::path(filePrefix).parent_path().string());
+      ->mkdir(std::filesystem::path(filePrefix_).parent_path().string());
 
   files_.reserve(numShards_);
   // Cache size must be a multiple of this so that each shard has the same max
@@ -74,7 +74,8 @@ SsdCache::SsdCache(
         i,
         fileMaxRegions,
         checkpointIntervalBytes / numShards,
-        disableFileCow));
+        disableFileCow,
+        executor_));
   }
 }
 
@@ -84,20 +85,19 @@ SsdFile& SsdCache::file(uint64_t fileId) {
 }
 
 bool SsdCache::startWrite() {
-  if (isShutdown_) {
-    return false;
-  }
-  if (writesInProgress_.fetch_add(numShards_) == 0) {
+  std::lock_guard<std::mutex> l(mutex_);
+  checkNotShutdownLocked();
+  if (writesInProgress_ == 0) {
     // No write was pending, so now all shards are counted as writing.
+    writesInProgress_ += numShards_;
     return true;
   }
-  // There were writes in progress, so compensate for the increment.
-  writesInProgress_.fetch_sub(numShards_);
+  BOLT_CHECK_GE(writesInProgress_, 0);
   return false;
 }
 
 void SsdCache::write(std::vector<CachePin> pins) {
-  BOLT_CHECK_LE(numShards_, writesInProgress_);
+  BOLT_CHECK_EQ(numShards_, writesInProgress_);
 
   BOLT_TEST_ADJUST("bytedance::bolt::cache::SsdCache::write", this);
 
@@ -163,12 +163,11 @@ bool SsdCache::removeFileEntries(
       success &= files_[i]->removeFileEntries(filesToRemove, filesRetained);
     } catch (const std::exception& e) {
       BOLT_SSD_CACHE_LOG(ERROR) << "Error removing file entries from SSD shard "
-                                << files_[i]->shardId() << ": " << e.what();
+                               << files_[i]->shardId() << ": " << e.what();
       success = false;
     }
     --writesInProgress_;
   }
-
   return success;
 }
 
@@ -180,12 +179,6 @@ SsdCacheStats SsdCache::stats() const {
   return stats;
 }
 
-void SsdCache::clear() {
-  for (auto& file : files_) {
-    file->clear();
-  }
-}
-
 std::string SsdCache::toString() const {
   auto data = stats();
   uint64_t capacity = maxBytes();
@@ -198,20 +191,44 @@ std::string SsdCache::toString() const {
   return out.str();
 }
 
-void SsdCache::testingDeleteFiles() {
-  for (auto& file : files_) {
-    file->deleteFile();
+void SsdCache::shutdown() {
+  {
+    std::lock_guard<std::mutex> l(mutex_);
+    if (shutdown_) {
+      BOLT_SSD_CACHE_LOG(INFO) << "SSD cache has already been shutdown";
+    }
+    shutdown_ = true;
   }
-}
 
-void SsdCache::shutdown() {
-  isShutdown_ = true;
+  BOLT_SSD_CACHE_LOG(INFO) << "SSD cache is shutting down";
   while (writesInProgress_) {
     std::this_thread::sleep_for(std::chrono::milliseconds(100)); // NOLINT
   }
   for (auto& file : files_) {
     file->checkpoint(true);
   }
+  BOLT_SSD_CACHE_LOG(INFO) << "SSD cache has been shutdown";
+}
+
+void SsdCache::testingClear() {
+  for (auto& file : files_) {
+    file->testingClear();
+  }
+}
+
+void SsdCache::testingDeleteFiles() {
+  for (auto& file : files_) {
+    file->testingDeleteFile();
+  }
+}
+
+uint64_t SsdCache::testingTotalLogEvictionFilesSize() {
+  uint64_t size = 0;
+  for (auto& file : files_) {
+    std::filesystem::path p{file->getEvictLogFilePath()};
+    size += std::filesystem::file_size(p);
+  }
+  return size;
 }
 
 } // namespace bytedance::bolt::cache
diff --git a/bolt/common/caching/SsdCache.h b/bolt/common/caching/SsdCache.h
index ce5d27103..9484703e9 100644
--- a/bolt/common/caching/SsdCache.h
+++ b/bolt/common/caching/SsdCache.h
@@ -106,29 +106,43 @@ class SsdCache {
   /// Drops all entries. Outstanding pins become invalid but reading them will
   /// mostly succeed since the files will not be rewritten until new content is
   /// stored.
-  void clear();
+  void testingClear();
 
   /// Deletes backing files. Used in testing.
   void testingDeleteFiles();
 
+  /// Returns the total size of eviction log files. Used in testing.
+  uint64_t testingTotalLogEvictionFilesSize();
+
   /// Stops writing to the cache files and waits for pending writes to finish.
   /// If checkpointing is on, makes a checkpoint.
   void shutdown();
 
   std::string toString() const;
 
+  const std::string& filePrefix() const {
+    return filePrefix_;
+  }
+
  private:
+  void checkNotShutdownLocked() {
+    BOLT_CHECK(
+        !shutdown_, "Unexpected write after SSD cache has been shutdown");
+  }
+
   const std::string filePrefix_;
   const int32_t numShards_;
+
+  // Stats for selecting entries to save from AsyncDataCache.
+  const std::unique_ptr<FileGroupStats> groupStats_;
+  folly::Executor* const executor_;
+  mutable std::mutex mutex_;
+
   std::vector<std::unique_ptr<SsdFile>> files_;
 
   // Count of shards with unfinished writes.
   std::atomic<int32_t> writesInProgress_{0};
-
-  // Stats for selecting entries to save from AsyncDataCache.
-  std::unique_ptr<FileGroupStats> groupStats_;
-  folly::Executor* executor_;
-  std::atomic<bool> isShutdown_{false};
+  bool shutdown_{false};
 };
 
 } // namespace bytedance::bolt::cache
diff --git a/bolt/common/caching/SsdFile.cpp b/bolt/common/caching/SsdFile.cpp
index 0a9b74008..814201352 100644
--- a/bolt/common/caching/SsdFile.cpp
+++ b/bolt/common/caching/SsdFile.cpp
@@ -153,7 +153,7 @@ SsdFile::SsdFile(
   BOLT_CHECK_GE(
       fd_,
       0,
-      "Cannot open or create {}. Error: {}",
+      "Could not open or create {}. Error: {}",
       filename,
       folly::errnoStr(errno));
 
@@ -175,10 +175,10 @@ SsdFile::SsdFile(
   writableRegions_.resize(numRegions_);
   std::iota(writableRegions_.begin(), writableRegions_.end(), 0);
   tracker_.resize(maxRegions_);
-  regionSizes_.resize(maxRegions_);
-  erasedRegionSizes_.resize(maxRegions_);
-  regionPins_.resize(maxRegions_);
-  if (checkpointIntervalBytes_) {
+  regionSizes_.resize(maxRegions_, 0);
+  erasedRegionSizes_.resize(maxRegions_, 0);
+  regionPins_.resize(maxRegions_, 0);
+  if (checkpointEnabled()) {
     initializeCheckpoint();
   }
 }
@@ -353,7 +353,8 @@ bool SsdFile::growOrEvictLocked() {
 
 void SsdFile::clearRegionEntriesLocked(const std::vector<int32_t>& regions) {
   std::unordered_set<int32_t> regionSet{regions.begin(), regions.end()};
-  // Remove all 'entries_' where the dependent points one of 'regionIndices'.
+  // Remove all 'entries_' that reference data in regions described by
+  // 'regionIndices'.
   auto it = entries_.begin();
   while (it != entries_.end()) {
     const auto region = regionIndex(it->second.offset());
@@ -439,8 +440,7 @@ void SsdFile::write(std::vector<CachePin>& pins) {
     storeIndex += numWritten;
   }
 
-  if ((checkpointIntervalBytes_ > 0) &&
-      (bytesAfterCheckpoint_ >= checkpointIntervalBytes_)) {
+  if (checkpointEnabled()) {
     checkpoint();
   }
 }
@@ -460,9 +460,9 @@ void SsdFile::verifyWrite(AsyncDataCacheEntry& entry, SsdRun ssdRun) {
   auto testData = std::make_unique<char[]>(entry.size());
   const auto rc = ::pread(fd_, testData.get(), entry.size(), ssdRun.offset());
   BOLT_CHECK_EQ(rc, entry.size());
-  if (entry.tinyData() != 0) {
+  if (entry.tinyData() != nullptr) {
     if (::memcmp(testData.get(), entry.tinyData(), entry.size()) != 0) {
-      BOLT_FAIL("bad read back");
+      BOLT_FAIL("Bad read back");
     }
   } else {
     const auto& data = entry.data();
@@ -476,8 +476,8 @@ void SsdFile::verifyWrite(AsyncDataCacheEntry& entry, SsdRun ssdRun) {
       if (badIndex != -1) {
         BOLT_FAIL("Bad read back");
       }
-      bytesLeft -= run.numBytes();
-      offset += run.numBytes();
+      bytesLeft -= compareSize;
+      offset += compareSize;
       if (bytesLeft <= 0) {
         break;
       };
@@ -515,7 +515,7 @@ void SsdFile::updateStats(SsdCacheStats& stats) const {
   stats.readCheckpointErrors += stats_.readCheckpointErrors;
 }
 
-void SsdFile::clear() {
+void SsdFile::testingClear() {
   std::lock_guard<std::shared_mutex> l(mutex_);
   entries_.clear();
   std::fill(regionSizes_.begin(), regionSizes_.end(), 0);
@@ -524,7 +524,8 @@ void SsdFile::clear() {
   std::iota(writableRegions_.begin(), writableRegions_.end(), 0);
 }
 
-void SsdFile::deleteFile() {
+void SsdFile::testingDeleteFile() {
+  process::TraceContext trace("SsdFile::testingDeleteFile");
   if (fd_) {
     close(fd_);
     fd_ = 0;
@@ -554,7 +555,7 @@ bool SsdFile::removeFileEntries(
     const FileCacheKey& cacheKey = it->first;
     const SsdRun& ssdRun = it->second;
 
-    if (!cacheKey.fileNum.hasValue()) {
+    if (!cacheKey.fileNum.has_value()) {
       ++it;
       continue;
     }
@@ -603,7 +604,7 @@ bool SsdFile::removeFileEntries(
 }
 
 void SsdFile::logEviction(const std::vector<int32_t>& regions) {
-  if (checkpointIntervalBytes_ > 0) {
+  if (checkpointEnabled()) {
     const int32_t rc = ::write(
         evictLogFd_, regions.data(), regions.size() * sizeof(regions[0]));
     if (rc != regions.size() * sizeof(regions[0])) {
@@ -628,12 +629,12 @@ void SsdFile::deleteCheckpoint(bool keepLog) {
   }
 
   checkpointDeleted_ = true;
-  const auto logPath = fileName_ + kLogExtension;
+  const auto logPath = getEvictLogFilePath();
   int32_t logRc = 0;
   if (!keepLog) {
     logRc = ::unlink(logPath.c_str());
   }
-  const auto checkpointPath = fileName_ + kCheckpointExtension;
+  const auto checkpointPath = getCheckpointFilePath();
   const auto checkpointRc = ::unlink(checkpointPath.c_str());
   if ((logRc != 0) || (checkpointRc != 0)) {
     ++stats_.deleteCheckpointErrors;
@@ -665,7 +666,7 @@ inline const char* asChar(const T* ptr) {
 
 void SsdFile::checkpoint(bool force) {
   std::lock_guard<std::shared_mutex> l(mutex_);
-  if (!force && (bytesAfterCheckpoint_ < checkpointIntervalBytes_)) {
+  if (!needCheckpoint(force)) {
     return;
   }
 
@@ -695,75 +696,83 @@ void SsdFile::checkpoint(bool force) {
     };
 
     std::ofstream state;
-    auto checkpointPath = fileName_ + kCheckpointExtension;
-    state.exceptions(std::ofstream::failbit);
-    state.open(checkpointPath, std::ios_base::out | std::ios_base::trunc);
-    // The checkpoint state file contains:
-    // int32_t The 4 bytes of kCheckpointMagic,
-    // int32_t maxRegions,
-    // int32_t numRegions,
-    // regionScores from the 'tracker_',
-    // {fileId, fileName} pairs,
-    // kMapMarker,
-    // {fileId, offset, SSdRun} triples,
-    // kEndMarker.
-    state.write(kCheckpointMagic, sizeof(int32_t));
-    state.write(asChar(&maxRegions_), sizeof(maxRegions_));
-    state.write(asChar(&numRegions_), sizeof(numRegions_));
-
-    // Copy the region scores before writing out for tsan.
-    const auto scoresCopy = tracker_.copyScores();
-    state.write(asChar(scoresCopy.data()), maxRegions_ * sizeof(uint64_t));
-    std::unordered_set<uint64_t> fileNums;
-    for (const auto& entry : entries_) {
-      const auto fileNum = entry.first.fileNum.id();
-      if (fileNums.insert(fileNum).second) {
-        state.write(asChar(&fileNum), sizeof(fileNum));
-        const auto name = fileIds().string(fileNum);
-        const int32_t length = name.size();
-        state.write(asChar(&length), sizeof(length));
-        state.write(name.data(), length);
+    const auto checkpointPath = getCheckpointFilePath();
+    try {
+      state.exceptions(std::ofstream::failbit);
+      state.open(checkpointPath, std::ios_base::out | std::ios_base::trunc);
+      // The checkpoint state file contains:
+      // int32_t The 4 bytes of kCheckpointMagic,
+      // int32_t maxRegions,
+      // int32_t numRegions,
+      // regionScores from the 'tracker_',
+      // {fileId, fileName} pairs,
+      // kMapMarker,
+      // {fileId, offset, SSdRun} triples,
+      // kEndMarker.
+      state.write(kCheckpointMagic, sizeof(int32_t));
+      state.write(asChar(&maxRegions_), sizeof(maxRegions_));
+      state.write(asChar(&numRegions_), sizeof(numRegions_));
+
+      // Copy the region scores before writing out for tsan.
+      const auto scoresCopy = tracker_.copyScores();
+      state.write(asChar(scoresCopy.data()), maxRegions_ * sizeof(uint64_t));
+      std::unordered_set<uint64_t> fileNums;
+      for (const auto& entry : entries_) {
+        const auto fileNum = entry.first.fileNum.id();
+        if (fileNums.insert(fileNum).second) {
+          state.write(asChar(&fileNum), sizeof(fileNum));
+          const auto name = fileIds().string(fileNum);
+          const int32_t length = name.size();
+          state.write(asChar(&length), sizeof(length));
+          state.write(name.data(), length);
+        }
       }
-    }
 
-    const auto mapMarker = kCheckpointMapMarker;
-    state.write(asChar(&mapMarker), sizeof(mapMarker));
-    for (auto& pair : entries_) {
-      auto id = pair.first.fileNum.id();
-      state.write(asChar(&id), sizeof(id));
-      state.write(asChar(&pair.first.offset), sizeof(pair.first.offset));
-      auto offsetAndSize = pair.second.bits();
-      state.write(asChar(&offsetAndSize), sizeof(offsetAndSize));
-    }
+      const auto mapMarker = kCheckpointMapMarker;
+      state.write(asChar(&mapMarker), sizeof(mapMarker));
+      for (auto& pair : entries_) {
+        auto id = pair.first.fileNum.id();
+        state.write(asChar(&id), sizeof(id));
+        state.write(asChar(&pair.first.offset), sizeof(pair.first.offset));
+        auto offsetAndSize = pair.second.bits();
+        state.write(asChar(&offsetAndSize), sizeof(offsetAndSize));
+      }
 
-    // NOTE: we need to ensure cache file data sync update completes before
-    // updating checkpoint file.
-    const auto fileSyncRc = fileSync->move();
-    checkRc(*fileSyncRc, "Sync of cache data file");
+      // NOTE: we need to ensure cache file data sync update completes before
+      // updating checkpoint file.
+      const auto fileSyncRc = fileSync->move();
+      checkRc(*fileSyncRc, "Sync of cache data file");
 
-    const auto endMarker = kCheckpointEndMarker;
-    state.write(asChar(&endMarker), sizeof(endMarker));
+      const auto endMarker = kCheckpointEndMarker;
+      state.write(asChar(&endMarker), sizeof(endMarker));
 
-    if (state.bad()) {
-      ++stats_.writeCheckpointErrors;
-      checkRc(-1, "Write of checkpoint file");
+      if (state.bad()) {
+        ++stats_.writeCheckpointErrors;
+        checkRc(-1, "Write of checkpoint file");
+      }
+      state.close();
+
+      // Sync checkpoint data file. ofstream does not have a sync method, so open
+      // as fd and sync that.
+      const auto checkpointFd = checkRc(
+          ::open(checkpointPath.c_str(), O_WRONLY),
+          "Open of checkpoint file for sync");
+      BOLT_CHECK_GE(checkpointFd, 0);
+      checkRc(::fsync(checkpointFd), "Sync of checkpoint file");
+      ::close(checkpointFd);
+
+      // NOTE: we shall truncate eviction log after checkpoint file sync
+      // completes so that we never recover from an old checkpoint file without
+      // log evictions. The latter may lead to data consistent issue.
+      checkRc(::ftruncate(evictLogFd_, 0), "Truncate of event log");
+      checkRc(::fsync(evictLogFd_), "Sync of evict log");
+    } catch (const std::exception& e) {
+      try {
+        checkpointError(-1, e.what());
+      } catch (const std::exception&) {
+      }
+      // Ignore nested exception.
     }
-    state.close();
-
-    // Sync checkpoint data file. ofstream does not have a sync method, so open
-    // as fd and sync that.
-    const auto checkpointFd = checkRc(
-        ::open(checkpointPath.c_str(), O_WRONLY),
-        "Open of checkpoint file for sync");
-    BOLT_CHECK_GE(checkpointFd, 0);
-    checkRc(::fsync(checkpointFd), "Sync of checkpoint file");
-    ::close(checkpointFd);
-
-    // NOTE: we shall truncate eviction log after checkpoint file sync
-    // completes so that we never recover from an old checkpoint file without
-    // log evictions. The latter might lead to data consistent issue.
-    checkRc(::ftruncate(evictLogFd_, 0), "Truncate of event log");
-    checkRc(::fsync(evictLogFd_), "Sync of evict log");
   } catch (const std::exception& e) {
     try {
       checkpointError(-1, e.what());
@@ -774,24 +783,24 @@ void SsdFile::checkpoint(bool force) {
 }
 
 void SsdFile::initializeCheckpoint() {
-  if (checkpointIntervalBytes_ == 0) {
+  if (!checkpointEnabled()) {
     return;
   }
   bool hasCheckpoint = true;
-  std::ifstream state(fileName_ + kCheckpointExtension);
+  std::ifstream state(getCheckpointFilePath());
   if (!state.is_open()) {
     hasCheckpoint = false;
     ++stats_.openCheckpointErrors;
     BOLT_SSD_CACHE_LOG(INFO)
         << "Starting shard " << shardId_ << " without checkpoint";
   }
-  const auto logPath = fileName_ + kLogExtension;
+  const auto logPath = getEvictLogFilePath();
   evictLogFd_ = ::open(logPath.c_str(), O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
   if (evictLogFd_ < 0) {
     ++stats_.openLogErrors;
     // Failure to open the log at startup is a process terminating error.
     BOLT_FAIL(
-        "Could not open evict log {}, rc {}: {}",
+        "Could not open evict log {}, rc {} :{}",
         logPath,
         evictLogFd_,
         folly::errnoStr(errno));
@@ -806,7 +815,7 @@ void SsdFile::initializeCheckpoint() {
     ++stats_.readCheckpointErrors;
     try {
       BOLT_SSD_CACHE_LOG(ERROR) << "Error recovering from checkpoint "
-                                << e.what() << ": Starting without checkpoint";
+                               << e.what() << ": Starting without checkpoint";
       entries_.clear();
       deleteCheckpoint(true);
     } catch (const std::exception&) {
diff --git a/bolt/common/caching/SsdFile.h b/bolt/common/caching/SsdFile.h
index 178e1bbf0..3620b916a 100644
--- a/bolt/common/caching/SsdFile.h
+++ b/bolt/common/caching/SsdFile.h
@@ -202,11 +202,11 @@ class SsdFile {
       const std::string& filename,
       int32_t shardId,
       int32_t maxRegions,
-      int64_t checkpointInternalBytes = 0,
+      int64_t checkpointIntervalBytes = 0,
       bool disableFileCow = false,
       folly::Executor* executor = nullptr);
 
-  // Adds entries of  'pins'  to this file. 'pins' must be in read mode and
+  // Adds entries of  'pins' to this file. 'pins' must be in read mode and
   // those pins that are successfully added to SSD are marked as being on SSD.
   // The file of the entries must be a file that is backed by 'this'.
   void write(std::vector<CachePin>& pins);
@@ -258,12 +258,6 @@ class SsdFile {
   // Adds 'stats_' to 'stats'.
   void updateStats(SsdCacheStats& stats) const;
 
-  // Resets this' to a post-construction empty state. See SsdCache::clear().
-  void clear();
-
-  // Deletes the backing file. Used in testing.
-  void deleteFile();
-
   /// Remove cached entries of files in the fileNum set 'filesToRemove'. If
   /// successful, return true, and 'filesRetained' contains entries that should
   /// not be removed, ex., from pinned regions. Otherwise, return false and
@@ -278,9 +272,33 @@ class SsdFile {
   // written since last checkpoint and silently returns if not.
   void checkpoint(bool force = false);
 
+  /// Returns the SSD file path.
+  const std::string& fileName() const {
+    return fileName_;
+  }
+
+  /// Returns the eviction log file path.
+  std::string getEvictLogFilePath() const {
+    return fileName_ + kLogExtension;
+  }
+
+  /// Deletes the backing file. Used in testing.
+  void testingDeleteFile();
+
+  /// Resets this' to a post-construction empty state. See SsdCache::clear().
+  void testingClear();
+
   /// Returns true if copy on write is disabled for this file. Used in testing.
   bool testingIsCowDisabled() const;
 
+  std::vector<double> testingCopyScores() {
+    return tracker_.copyScores();
+  }
+
+  int32_t testingNumWritableRegions() const {
+    return writableRegions_.size();
+  }
+
  private:
   // 4 first bytes of a checkpoint file. Allows distinguishing between format
   // versions.
@@ -341,10 +359,28 @@ class SsdFile {
   // the files for making new checkpoints.
   void initializeCheckpoint();
 
-  // Synchronously logs that 'regions' are no longer valid in a possibly xisting
+  // Synchronously logs that 'regions' are no longer valid in a possibly existing
   // checkpoint.
   void logEviction(const std::vector<int32_t>& regions);
 
+  // Returns true if checkpoint has been enabled.
+  bool checkpointEnabled() const {
+    return checkpointIntervalBytes_ > 0;
+  }
+
+  // Returns true if checkpoint is needed.
+  bool needCheckpoint(bool force) const {
+    if (!checkpointEnabled()) {
+      return false;
+    }
+    return force || (bytesAfterCheckpoint_ >= checkpointIntervalBytes_);
+  }
+
+  // Returns the checkpoint file path.
+  std::string getCheckpointFilePath() const {
+    return fileName_ + kCheckpointExtension;
+  }
+
   static constexpr const char* kLogExtension = ".log";
   static constexpr const char* kCheckpointExtension = ".cpt";
 
diff --git a/bolt/common/caching/tests/AsyncDataCacheTest.cpp b/bolt/common/caching/tests/AsyncDataCacheTest.cpp
index 4246454c6..ee63918ab 100644
--- a/bolt/common/caching/tests/AsyncDataCacheTest.cpp
+++ b/bolt/common/caching/tests/AsyncDataCacheTest.cpp
@@ -92,19 +92,22 @@ class AsyncDataCacheTest : public testing::Test {
   }
 
   void TearDown() override {
-    if (executor_) {
-      executor_->join();
-    }
     if (cache_) {
-      auto ssdCache = cache_->ssdCache();
+      cache_->shutdown();
+      auto* ssdCache = cache_->ssdCache();
       if (ssdCache) {
         ssdCache->testingDeleteFiles();
       }
-      cache_->shutdown();
+    }
+    if (executor_) {
+      executor_->join();
     }
   }
 
-  void initializeCache(uint64_t maxBytes, int64_t ssdBytes = 0) {
+  void initializeCache(
+      uint64_t maxBytes,
+      int64_t ssdBytes = 0,
+      int64_t checkpointIntervalBytes = 0) {
     if (cache_ != nullptr) {
       cache_->shutdown();
     }
@@ -125,7 +128,8 @@ class AsyncDataCacheTest : public testing::Test {
           ssdBytes,
           4,
           executor(),
-          ssdBytes / 20);
+          checkpointIntervalBytes > 0 ? checkpointIntervalBytes
+                                      : ssdBytes / 20);
     }
 
     memory::MemoryManager::Options options;
@@ -1162,4 +1166,45 @@ DEBUG_ONLY_TEST_F(AsyncDataCacheTest, ttl) {
 }
 #endif
 
+TEST_F(AsyncDataCacheTest, shutdown) {
+  constexpr uint64_t kRamBytes = 32 << 20;
+  constexpr uint64_t kSsdBytes = 64UL << 20;
+
+  // Initialize cache with a big checkpointIntervalBytes, giving eviction log
+  // chance to survive.
+  initializeCache(kRamBytes, kSsdBytes, kSsdBytes * 10);
+  ASSERT_EQ(cache_->ssdCache()->stats().openCheckpointErrors, 4);
+
+  // Write large amount of data, making sure eviction is triggered and the log
+  // file is not zero.
+  loadLoop(0, 16 * kSsdBytes);
+  ASSERT_GT(cache_->ssdCache()->stats().regionsEvicted, 0);
+  ASSERT_GT(cache_->ssdCache()->testingTotalLogEvictionFilesSize(), 0);
+
+  // Shutdown cache.
+  const auto bytesWrittenBeforeShutdown =
+      cache_->ssdCache()->stats().bytesWritten;
+  cache_->ssdCache()->shutdown();
+  const auto bytesWrittenAfterShutdown =
+      cache_->ssdCache()->stats().bytesWritten;
+  cache_->ssdCache()->shutdown();
+
+  // No new data has been written after shutdown.
+  ASSERT_EQ(bytesWrittenBeforeShutdown, bytesWrittenAfterShutdown);
+  // Eviction log files have been truncated.
+  ASSERT_EQ(cache_->ssdCache()->testingTotalLogEvictionFilesSize(), 0);
+
+  // New cache write attempt is blocked and triggers exception.
+  VELOX_ASSERT_THROW(
+      cache_->ssdCache()->startWrite(),
+      "Unexpected write after SSD cache has been shutdown");
+
+  // Re-initialize cache.
+  cache_->ssdCache()->testingClear();
+  initializeCache(kRamBytes, kSsdBytes, kSsdBytes * 10);
+  // Checkpoint files are intact and readable.
+  ASSERT_EQ(cache_->ssdCache()->stats().openCheckpointErrors, 0);
+  ASSERT_EQ(cache_->ssdCache()->stats().readCheckpointErrors, 0);
+}
+
 // TODO: add concurrent fuzzer test.
diff --git a/bolt/common/caching/tests/SsdFileTest.cpp b/bolt/common/caching/tests/SsdFileTest.cpp
index 9a3dcf577..eccf77bc7 100644
--- a/bolt/common/caching/tests/SsdFileTest.cpp
+++ b/bolt/common/caching/tests/SsdFileTest.cpp
@@ -62,7 +62,7 @@ class SsdFileTest : public testing::Test {
 
   void TearDown() override {
     if (ssdFile_) {
-      ssdFile_->deleteFile();
+      ssdFile_->testingDeleteFile();
     }
     if (cache_) {
       cache_->shutdown();
diff --git a/bolt/dwio/dwrf/test/CacheInputTest.cpp b/bolt/dwio/dwrf/test/CacheInputTest.cpp
index 0e3295602..a4e8d2537 100644
--- a/bolt/dwio/dwrf/test/CacheInputTest.cpp
+++ b/bolt/dwio/dwrf/test/CacheInputTest.cpp
@@ -75,14 +75,14 @@ class CacheTest : public testing::Test {
   }
 
   void TearDown() override {
+    if (cache_) {
+      cache_->shutdown();
+    }
     executor_->join();
-    auto ssdCache = cache_->ssdCache();
+    auto* ssdCache = cache_->ssdCache();
     if (ssdCache) {
       ssdCache->testingDeleteFiles();
     }
-    if (cache_) {
-      cache_->shutdown();
-    }
   }
 
   void initializeCache(uint64_t maxBytes, uint64_t ssdBytes = 0) {