diff --git a/HISTORY.md b/HISTORY.md
index 16a64832b..c87f9932d 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -3,6 +3,7 @@
 ### Bug Fixes
 ### New Features
 ### Public API change
+* Made SystemClock extend the Customizable class and added a CreateFromString method.  Implementations need to be registered with the ObjectRegistry and to implement a Name() method in order to be created via this method.
 
 ## 6.25.0 (2021-09-20)
 ### Bug Fixes
diff --git a/db/blob/blob_file_builder_test.cc b/db/blob/blob_file_builder_test.cc
index 1b85d05e8..4b30fcfbf 100644
--- a/db/blob/blob_file_builder_test.cc
+++ b/db/blob/blob_file_builder_test.cc
@@ -39,9 +39,10 @@ class TestFileNumberGenerator {
 
 class BlobFileBuilderTest : public testing::Test {
  protected:
-  BlobFileBuilderTest() : mock_env_(Env::Default()) {
-    fs_ = mock_env_.GetFileSystem().get();
-    clock_ = mock_env_.GetSystemClock().get();
+  BlobFileBuilderTest() {
+    mock_env_.reset(MockEnv::Create(Env::Default()));
+    fs_ = mock_env_->GetFileSystem().get();
+    clock_ = mock_env_->GetSystemClock().get();
   }
 
   void VerifyBlobFile(uint64_t blob_file_number,
@@ -108,7 +109,7 @@ class BlobFileBuilderTest : public testing::Test {
     ASSERT_EQ(footer.expiration_range, ExpirationRange());
   }
 
-  MockEnv mock_env_;
+  std::unique_ptr<Env> mock_env_;
   FileSystem* fs_;
   SystemClock* clock_;
   FileOptions file_options_;
@@ -123,11 +124,11 @@ TEST_F(BlobFileBuilderTest, BuildAndCheckOneFile) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileBuilderTest_BuildAndCheckOneFile"),
       0);
   options.enable_blob_files = true;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
@@ -206,12 +207,12 @@ TEST_F(BlobFileBuilderTest, BuildAndCheckMultipleFiles) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileBuilderTest_BuildAndCheckMultipleFiles"),
       0);
   options.enable_blob_files = true;
   options.blob_file_size = value_size;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
@@ -293,11 +294,12 @@ TEST_F(BlobFileBuilderTest, InlinedValues) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_InlinedValues"),
+      test::PerThreadDBPath(mock_env_.get(),
+                            "BlobFileBuilderTest_InlinedValues"),
       0);
   options.enable_blob_files = true;
   options.min_blob_size = 1024;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
@@ -347,10 +349,11 @@ TEST_F(BlobFileBuilderTest, Compression) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_Compression"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileBuilderTest_Compression"),
+      0);
   options.enable_blob_files = true;
   options.blob_compression_type = kSnappyCompression;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
@@ -429,11 +432,12 @@ TEST_F(BlobFileBuilderTest, CompressionError) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_CompressionError"),
+      test::PerThreadDBPath(mock_env_.get(),
+                            "BlobFileBuilderTest_CompressionError"),
       0);
   options.enable_blob_files = true;
   options.blob_compression_type = kSnappyCompression;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
 
@@ -506,11 +510,12 @@ TEST_F(BlobFileBuilderTest, Checksum) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_Checksum"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileBuilderTest_Checksum"),
+      0);
   options.enable_blob_files = true;
   options.file_checksum_gen_factory =
       std::make_shared<DummyFileChecksumGenFactory>();
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
@@ -575,12 +580,12 @@ class BlobFileBuilderIOErrorTest
     : public testing::Test,
       public testing::WithParamInterface<std::string> {
  protected:
-  BlobFileBuilderIOErrorTest()
-      : mock_env_(Env::Default()),
-        fs_(mock_env_.GetFileSystem().get()),
-        sync_point_(GetParam()) {}
+  BlobFileBuilderIOErrorTest() : sync_point_(GetParam()) {
+    mock_env_.reset(MockEnv::Create(Env::Default()));
+    fs_ = mock_env_->GetFileSystem().get();
+  }
 
-  MockEnv mock_env_;
+  std::unique_ptr<Env> mock_env_;
   FileSystem* fs_;
   FileOptions file_options_;
   std::string sync_point_;
@@ -602,11 +607,12 @@ TEST_P(BlobFileBuilderIOErrorTest, IOError) {
 
   Options options;
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileBuilderIOErrorTest_IOError"),
+      test::PerThreadDBPath(mock_env_.get(),
+                            "BlobFileBuilderIOErrorTest_IOError"),
       0);
   options.enable_blob_files = true;
   options.blob_file_size = value_size;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
 
   ImmutableOptions immutable_options(options);
   MutableCFOptions mutable_cf_options(options);
diff --git a/db/blob/blob_file_cache_test.cc b/db/blob/blob_file_cache_test.cc
index bef2d6202..b0ec22a67 100644
--- a/db/blob/blob_file_cache_test.cc
+++ b/db/blob/blob_file_cache_test.cc
@@ -84,17 +84,18 @@ void WriteBlobFile(uint32_t column_family_id,
 
 class BlobFileCacheTest : public testing::Test {
  protected:
-  BlobFileCacheTest() : mock_env_(Env::Default()) {}
+  BlobFileCacheTest() { mock_env_.reset(MockEnv::Create(Env::Default())); }
 
-  MockEnv mock_env_;
+  std::unique_ptr<Env> mock_env_;
 };
 
 TEST_F(BlobFileCacheTest, GetBlobFileReader) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.statistics = CreateDBStatistics();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileCacheTest_GetBlobFileReader"),
+      test::PerThreadDBPath(mock_env_.get(),
+                            "BlobFileCacheTest_GetBlobFileReader"),
       0);
   options.enable_blob_files = true;
 
@@ -135,10 +136,10 @@ TEST_F(BlobFileCacheTest, GetBlobFileReader) {
 
 TEST_F(BlobFileCacheTest, GetBlobFileReader_Race) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.statistics = CreateDBStatistics();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileCacheTest_GetBlobFileReader_Race"),
       0);
   options.enable_blob_files = true;
@@ -187,10 +188,10 @@ TEST_F(BlobFileCacheTest, GetBlobFileReader_Race) {
 
 TEST_F(BlobFileCacheTest, GetBlobFileReader_IOError) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.statistics = CreateDBStatistics();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileCacheTest_GetBlobFileReader_IOError"),
       0);
   options.enable_blob_files = true;
@@ -221,10 +222,10 @@ TEST_F(BlobFileCacheTest, GetBlobFileReader_IOError) {
 
 TEST_F(BlobFileCacheTest, GetBlobFileReader_CacheFull) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.statistics = CreateDBStatistics();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileCacheTest_GetBlobFileReader_CacheFull"),
       0);
   options.enable_blob_files = true;
diff --git a/db/blob/blob_file_reader_test.cc b/db/blob/blob_file_reader_test.cc
index 8544b53d4..ebafef834 100644
--- a/db/blob/blob_file_reader_test.cc
+++ b/db/blob/blob_file_reader_test.cc
@@ -134,16 +134,15 @@ void WriteBlobFile(const ImmutableOptions& immutable_options,
 
 class BlobFileReaderTest : public testing::Test {
  protected:
-  BlobFileReaderTest() : mock_env_(Env::Default()) {}
-
-  MockEnv mock_env_;
+  BlobFileReaderTest() { mock_env_.reset(MockEnv::Create(Env::Default())); }
+  std::unique_ptr<Env> mock_env_;
 };
 
 TEST_F(BlobFileReaderTest, CreateReaderAndGetBlob) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderTest_CreateReaderAndGetBlob"),
       0);
   options.enable_blob_files = true;
@@ -400,9 +399,10 @@ TEST_F(BlobFileReaderTest, Malformed) {
   // detect the error when we open it for reading
 
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileReaderTest_Malformed"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileReaderTest_Malformed"),
+      0);
   options.enable_blob_files = true;
 
   ImmutableOptions immutable_options(options);
@@ -452,9 +452,9 @@ TEST_F(BlobFileReaderTest, Malformed) {
 
 TEST_F(BlobFileReaderTest, TTL) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileReaderTest_TTL"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileReaderTest_TTL"), 0);
   options.enable_blob_files = true;
 
   ImmutableOptions immutable_options(options);
@@ -486,9 +486,9 @@ TEST_F(BlobFileReaderTest, TTL) {
 
 TEST_F(BlobFileReaderTest, ExpirationRangeInHeader) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderTest_ExpirationRangeInHeader"),
       0);
   options.enable_blob_files = true;
@@ -525,9 +525,9 @@ TEST_F(BlobFileReaderTest, ExpirationRangeInHeader) {
 
 TEST_F(BlobFileReaderTest, ExpirationRangeInFooter) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderTest_ExpirationRangeInFooter"),
       0);
   options.enable_blob_files = true;
@@ -564,9 +564,9 @@ TEST_F(BlobFileReaderTest, ExpirationRangeInFooter) {
 
 TEST_F(BlobFileReaderTest, IncorrectColumnFamily) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderTest_IncorrectColumnFamily"),
       0);
   options.enable_blob_files = true;
@@ -602,9 +602,10 @@ TEST_F(BlobFileReaderTest, IncorrectColumnFamily) {
 
 TEST_F(BlobFileReaderTest, BlobCRCError) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileReaderTest_BlobCRCError"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileReaderTest_BlobCRCError"),
+      0);
   options.enable_blob_files = true;
 
   ImmutableOptions immutable_options(options);
@@ -660,9 +661,10 @@ TEST_F(BlobFileReaderTest, Compression) {
   }
 
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_, "BlobFileReaderTest_Compression"), 0);
+      test::PerThreadDBPath(mock_env_.get(), "BlobFileReaderTest_Compression"),
+      0);
   options.enable_blob_files = true;
 
   ImmutableOptions immutable_options(options);
@@ -726,9 +728,9 @@ TEST_F(BlobFileReaderTest, UncompressionError) {
   }
 
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderTest_UncompressionError"),
       0);
   options.enable_blob_files = true;
@@ -785,13 +787,13 @@ class BlobFileReaderIOErrorTest
     : public testing::Test,
       public testing::WithParamInterface<std::string> {
  protected:
-  BlobFileReaderIOErrorTest()
-      : mock_env_(Env::Default()),
-        fault_injection_env_(&mock_env_),
-        sync_point_(GetParam()) {}
+  BlobFileReaderIOErrorTest() : sync_point_(GetParam()) {
+    mock_env_.reset(MockEnv::Create(Env::Default()));
+    fault_injection_env_.reset(new FaultInjectionTestEnv(mock_env_.get()));
+  }
 
-  MockEnv mock_env_;
-  FaultInjectionTestEnv fault_injection_env_;
+  std::unique_ptr<Env> mock_env_;
+  std::unique_ptr<FaultInjectionTestEnv> fault_injection_env_;
   std::string sync_point_;
 };
 
@@ -807,9 +809,9 @@ TEST_P(BlobFileReaderIOErrorTest, IOError) {
   // Simulates an I/O error during the specified step
 
   Options options;
-  options.env = &fault_injection_env_;
+  options.env = fault_injection_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&fault_injection_env_,
+      test::PerThreadDBPath(fault_injection_env_.get(),
                             "BlobFileReaderIOErrorTest_IOError"),
       0);
   options.enable_blob_files = true;
@@ -831,8 +833,8 @@ TEST_P(BlobFileReaderIOErrorTest, IOError) {
                 &blob_offset, &blob_size);
 
   SyncPoint::GetInstance()->SetCallBack(sync_point_, [this](void* /* arg */) {
-    fault_injection_env_.SetFilesystemActive(false,
-                                             Status::IOError(sync_point_));
+    fault_injection_env_->SetFilesystemActive(false,
+                                              Status::IOError(sync_point_));
   });
   SyncPoint::GetInstance()->EnableProcessing();
 
@@ -870,10 +872,11 @@ class BlobFileReaderDecodingErrorTest
     : public testing::Test,
       public testing::WithParamInterface<std::string> {
  protected:
-  BlobFileReaderDecodingErrorTest()
-      : mock_env_(Env::Default()), sync_point_(GetParam()) {}
+  BlobFileReaderDecodingErrorTest() : sync_point_(GetParam()) {
+    mock_env_.reset(MockEnv::Create(Env::Default()));
+  }
 
-  MockEnv mock_env_;
+  std::unique_ptr<Env> mock_env_;
   std::string sync_point_;
 };
 
@@ -885,9 +888,9 @@ INSTANTIATE_TEST_CASE_P(BlobFileReaderTest, BlobFileReaderDecodingErrorTest,
 
 TEST_P(BlobFileReaderDecodingErrorTest, DecodingError) {
   Options options;
-  options.env = &mock_env_;
+  options.env = mock_env_.get();
   options.cf_paths.emplace_back(
-      test::PerThreadDBPath(&mock_env_,
+      test::PerThreadDBPath(mock_env_.get(),
                             "BlobFileReaderDecodingErrorTest_DecodingError"),
       0);
   options.enable_blob_files = true;
diff --git a/db/db_compaction_test.cc b/db/db_compaction_test.cc
index 5eac7a929..326344411 100644
--- a/db/db_compaction_test.cc
+++ b/db/db_compaction_test.cc
@@ -4979,7 +4979,7 @@ TEST_P(DBCompactionDirectIOTest, DirectIO) {
   options.create_if_missing = true;
   options.disable_auto_compactions = true;
   options.use_direct_io_for_flush_and_compaction = GetParam();
-  options.env = new MockEnv(Env::Default());
+  options.env = MockEnv::Create(Env::Default());
   Reopen(options);
   bool readahead = false;
   SyncPoint::GetInstance()->SetCallBack(
diff --git a/db/db_dynamic_level_test.cc b/db/db_dynamic_level_test.cc
index 18221b4f4..7e9546498 100644
--- a/db/db_dynamic_level_test.cc
+++ b/db/db_dynamic_level_test.cc
@@ -13,9 +13,9 @@
 #if !defined(ROCKSDB_LITE)
 
 #include "db/db_test_util.h"
-#include "env/mock_env.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
+#include "rocksdb/env.h"
 #include "util/random.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -30,7 +30,7 @@ TEST_F(DBTestDynamicLevel, DynamicLevelMaxBytesBase) {
     return;
   }
   // Use InMemoryEnv, or it would be too slow.
-  std::unique_ptr<Env> env(new MockEnv(env_));
+  std::unique_ptr<Env> env(NewMemEnv(env_));
 
   const int kNKeys = 1000;
   int keys[kNKeys];
diff --git a/db/db_flush_test.cc b/db/db_flush_test.cc
index 73cdd813b..a5bda71c8 100644
--- a/db/db_flush_test.cc
+++ b/db/db_flush_test.cc
@@ -1381,7 +1381,7 @@ TEST_P(DBFlushDirectIOTest, DirectIO) {
   options.disable_auto_compactions = true;
   options.max_background_flushes = 2;
   options.use_direct_io_for_flush_and_compaction = GetParam();
-  options.env = new MockEnv(Env::Default());
+  options.env = MockEnv::Create(Env::Default());
   SyncPoint::GetInstance()->SetCallBack(
       "BuildTable:create_file", [&](void* arg) {
         bool* use_direct_writes = static_cast<bool*>(arg);
diff --git a/db/db_test.cc b/db/db_test.cc
index b99615caa..7729dea52 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -95,7 +95,7 @@ class DBTestWithParam
 };
 
 TEST_F(DBTest, MockEnvTest) {
-  std::unique_ptr<MockEnv> env{new MockEnv(Env::Default())};
+  std::unique_ptr<MockEnv> env{MockEnv::Create(Env::Default())};
   Options options;
   options.create_if_missing = true;
   options.env = env.get();
@@ -4845,7 +4845,7 @@ TEST_F(DBTest, DynamicLevelCompressionPerLevel2) {
   options.level0_stop_writes_trigger = 2;
   options.soft_pending_compaction_bytes_limit = 1024 * 1024;
   options.target_file_size_base = 20;
-
+  options.env = env_;
   options.level_compaction_dynamic_level_bytes = true;
   options.max_bytes_for_level_base = 200;
   options.max_bytes_for_level_multiplier = 8;
diff --git a/db/db_test_util.cc b/db/db_test_util.cc
index 393e16346..ea07516b8 100644
--- a/db/db_test_util.cc
+++ b/db/db_test_util.cc
@@ -63,7 +63,7 @@ DBTestBase::DBTestBase(const std::string path, bool env_do_fsync)
   EXPECT_OK(test::CreateEnvFromSystem(config_options, &base_env, &env_guard_));
   EXPECT_NE(nullptr, base_env);
   if (getenv("MEM_ENV")) {
-    mem_env_ = new MockEnv(base_env);
+    mem_env_ = MockEnv::Create(base_env, base_env->GetSystemClock());
   }
 #ifndef ROCKSDB_LITE
   if (getenv("ENCRYPTED_ENV")) {
diff --git a/db/db_universal_compaction_test.cc b/db/db_universal_compaction_test.cc
index d3414e2e7..610341bed 100644
--- a/db/db_universal_compaction_test.cc
+++ b/db/db_universal_compaction_test.cc
@@ -737,6 +737,7 @@ TEST_P(DBTestUniversalCompactionParallel, UniversalCompactionParallel) {
   Options options = CurrentOptions();
   options.compaction_style = kCompactionStyleUniversal;
   options.num_levels = num_levels_;
+  options.env = env_;
   options.write_buffer_size = 1 << 10;  // 1KB
   options.level0_file_num_compaction_trigger = 3;
   options.max_background_compactions = 3;
diff --git a/db/external_sst_file_test.cc b/db/external_sst_file_test.cc
index 633a0ddac..660f1f5ff 100644
--- a/db/external_sst_file_test.cc
+++ b/db/external_sst_file_test.cc
@@ -1991,7 +1991,8 @@ TEST_F(ExternalSSTFileTest, CompactionDeadlock) {
     if (running_threads.load() == 0) {
       break;
     }
-    env_->SleepForMicroseconds(500000);
+    // Make sure we do a "real sleep", not a mock one.
+    SystemClock::Default()->SleepForMicroseconds(500000);
   }
 
   ASSERT_EQ(running_threads.load(), 0);
diff --git a/db/fault_injection_test.cc b/db/fault_injection_test.cc
index 1a3715e32..af9329fbf 100644
--- a/db/fault_injection_test.cc
+++ b/db/fault_injection_test.cc
@@ -94,7 +94,7 @@ class FaultInjectionTest
       return false;
     } else {
       if (option_config_ == kMultiLevels) {
-        base_env_.reset(new MockEnv(Env::Default()));
+        base_env_.reset(MockEnv::Create(Env::Default()));
       }
       return true;
     }
diff --git a/db/wal_manager_test.cc b/db/wal_manager_test.cc
index 580379a6c..a579c4dad 100644
--- a/db/wal_manager_test.cc
+++ b/db/wal_manager_test.cc
@@ -32,13 +32,12 @@ namespace ROCKSDB_NAMESPACE {
 class WalManagerTest : public testing::Test {
  public:
   WalManagerTest()
-      : env_(new MockEnv(Env::Default())),
-        dbname_(test::PerThreadDBPath("wal_manager_test")),
+      : dbname_(test::PerThreadDBPath("wal_manager_test")),
         db_options_(),
         table_cache_(NewLRUCache(50000, 16)),
         write_buffer_manager_(db_options_.db_write_buffer_size),
         current_log_number_(0) {
-    DestroyDB(dbname_, Options());
+    env_.reset(MockEnv::Create(Env::Default())), DestroyDB(dbname_, Options());
   }
 
   void Init() {
@@ -247,7 +246,7 @@ TEST_F(WalManagerTest, WALArchivalSizeLimit) {
   ASSERT_TRUE(archive_size <= db_options_.WAL_size_limit_MB * 1024 * 1024);
 
   db_options_.WAL_ttl_seconds = 1;
-  env_->FakeSleepForMicroseconds(2 * 1000 * 1000);
+  env_->SleepForMicroseconds(2 * 1000 * 1000);
   Reopen();
   wal_manager_->PurgeObsoleteWALFiles();
 
@@ -273,7 +272,7 @@ TEST_F(WalManagerTest, WALArchivalTtl) {
   ASSERT_GT(log_files.size(), 0U);
 
   db_options_.WAL_ttl_seconds = 1;
-  env_->FakeSleepForMicroseconds(3 * 1000 * 1000);
+  env_->SleepForMicroseconds(3 * 1000 * 1000);
   Reopen();
   wal_manager_->PurgeObsoleteWALFiles();
 
diff --git a/env/emulated_clock.h b/env/emulated_clock.h
new file mode 100644
index 000000000..622737635
--- /dev/null
+++ b/env/emulated_clock.h
@@ -0,0 +1,114 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+//
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+#pragma once
+
+#include <atomic>
+#include <string>
+
+#include "rocksdb/status.h"
+#include "rocksdb/system_clock.h"
+
+namespace ROCKSDB_NAMESPACE {
+// A SystemClock that can "mock" sleep and counts its operations.
+class EmulatedSystemClock : public SystemClockWrapper {
+ private:
+  // Something to return when mocking current time
+  const int64_t maybe_starting_time_;
+  std::atomic<int> sleep_counter_{0};
+  std::atomic<int> cpu_counter_{0};
+  std::atomic<int64_t> addon_microseconds_{0};
+  // Do not modify in the env of a running DB (could cause deadlock)
+  std::atomic<bool> time_elapse_only_sleep_;
+  bool no_slowdown_;
+
+ public:
+  explicit EmulatedSystemClock(const std::shared_ptr<SystemClock>& base,
+                               bool time_elapse_only_sleep = false);
+
+  static const char* kClassName() { return "TimeEmulatedSystemClock"; }
+  const char* Name() const override { return kClassName(); }
+
+  virtual void SleepForMicroseconds(int micros) override {
+    sleep_counter_++;
+    if (no_slowdown_ || time_elapse_only_sleep_) {
+      addon_microseconds_.fetch_add(micros);
+    }
+    if (!no_slowdown_) {
+      SystemClockWrapper::SleepForMicroseconds(micros);
+    }
+  }
+
+  void MockSleepForMicroseconds(int64_t micros) {
+    sleep_counter_++;
+    assert(no_slowdown_);
+    addon_microseconds_.fetch_add(micros);
+  }
+
+  void MockSleepForSeconds(int64_t seconds) {
+    sleep_counter_++;
+    assert(no_slowdown_);
+    addon_microseconds_.fetch_add(seconds * 1000000);
+  }
+
+  void SetTimeElapseOnlySleep(bool enabled) {
+    // We cannot set these before destroying the last DB because they might
+    // cause a deadlock or similar without the appropriate options set in
+    // the DB.
+    time_elapse_only_sleep_ = enabled;
+    no_slowdown_ = enabled;
+  }
+
+  bool IsTimeElapseOnlySleep() const { return time_elapse_only_sleep_.load(); }
+  void SetMockSleep(bool enabled = true) { no_slowdown_ = enabled; }
+  bool IsMockSleepEnabled() const { return no_slowdown_; }
+
+  int GetSleepCounter() const { return sleep_counter_.load(); }
+
+  virtual Status GetCurrentTime(int64_t* unix_time) override {
+    Status s;
+    if (time_elapse_only_sleep_) {
+      *unix_time = maybe_starting_time_;
+    } else {
+      s = SystemClockWrapper::GetCurrentTime(unix_time);
+    }
+    if (s.ok()) {
+      // mock microseconds elapsed to seconds of time
+      *unix_time += addon_microseconds_.load() / 1000000;
+    }
+    return s;
+  }
+
+  virtual uint64_t CPUNanos() override {
+    cpu_counter_++;
+    return SystemClockWrapper::CPUNanos();
+  }
+
+  virtual uint64_t CPUMicros() override {
+    cpu_counter_++;
+    return SystemClockWrapper::CPUMicros();
+  }
+
+  virtual uint64_t NowNanos() override {
+    return (time_elapse_only_sleep_ ? 0 : SystemClockWrapper::NowNanos()) +
+           addon_microseconds_.load() * 1000;
+  }
+
+  virtual uint64_t NowMicros() override {
+    return (time_elapse_only_sleep_ ? 0 : SystemClockWrapper::NowMicros()) +
+           addon_microseconds_.load();
+  }
+
+  int GetCpuCounter() const { return cpu_counter_.load(); }
+
+  void ResetCounters() {
+    cpu_counter_.store(0);
+    sleep_counter_.store(0);
+  }
+};
+}  // namespace ROCKSDB_NAMESPACE
diff --git a/env/env.cc b/env/env.cc
index 5aa25e523..c940b36c9 100644
--- a/env/env.cc
+++ b/env/env.cc
@@ -12,6 +12,7 @@
 #include <thread>
 
 #include "env/composite_env_wrapper.h"
+#include "env/emulated_clock.h"
 #include "env/unique_id.h"
 #include "logging/env_logger.h"
 #include "memory/arena.h"
@@ -20,7 +21,9 @@
 #include "rocksdb/convenience.h"
 #include "rocksdb/options.h"
 #include "rocksdb/system_clock.h"
+#include "rocksdb/utilities/customizable_util.h"
 #include "rocksdb/utilities/object_registry.h"
+#include "rocksdb/utilities/options_type.h"
 #include "util/autovector.h"
 #include "util/string_util.h"
 
@@ -1128,4 +1131,81 @@ const std::shared_ptr<FileSystem>& Env::GetFileSystem() const {
 const std::shared_ptr<SystemClock>& Env::GetSystemClock() const {
   return system_clock_;
 }
+namespace {
+static std::unordered_map<std::string, OptionTypeInfo> sc_wrapper_type_info = {
+#ifndef ROCKSDB_LITE
+    {"target",
+     OptionTypeInfo::AsCustomSharedPtr<SystemClock>(
+         0, OptionVerificationType::kByName, OptionTypeFlags::kDontSerialize)},
+#endif  // ROCKSDB_LITE
+};
+
+}  // namespace
+SystemClockWrapper::SystemClockWrapper(const std::shared_ptr<SystemClock>& t)
+    : target_(t) {
+  RegisterOptions("", &target_, &sc_wrapper_type_info);
+}
+
+Status SystemClockWrapper::PrepareOptions(const ConfigOptions& options) {
+  if (target_ == nullptr) {
+    target_ = SystemClock::Default();
+  }
+  return SystemClock::PrepareOptions(options);
+}
+
+#ifndef ROCKSDB_LITE
+std::string SystemClockWrapper::SerializeOptions(
+    const ConfigOptions& config_options, const std::string& header) const {
+  auto parent = SystemClock::SerializeOptions(config_options, "");
+  if (config_options.IsShallow() || target_ == nullptr ||
+      target_->IsInstanceOf(SystemClock::kDefaultName())) {
+    return parent;
+  } else {
+    std::string result = header;
+    if (!StartsWith(parent, OptionTypeInfo::kIdPropName())) {
+      result.append(OptionTypeInfo::kIdPropName()).append("=");
+    }
+    result.append(parent);
+    if (!EndsWith(result, config_options.delimiter)) {
+      result.append(config_options.delimiter);
+    }
+    result.append("target=").append(target_->ToString(config_options));
+    return result;
+  }
+}
+#endif  // ROCKSDB_LITE
+
+#ifndef ROCKSDB_LITE
+static int RegisterBuiltinSystemClocks(ObjectLibrary& library,
+                                       const std::string& /*arg*/) {
+  library.Register<SystemClock>(
+      EmulatedSystemClock::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<SystemClock>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new EmulatedSystemClock(SystemClock::Default()));
+        return guard->get();
+      });
+  size_t num_types;
+  return static_cast<int>(library.GetFactoryCount(&num_types));
+}
+#endif  // ROCKSDB_LITE
+
+Status SystemClock::CreateFromString(const ConfigOptions& config_options,
+                                     const std::string& value,
+                                     std::shared_ptr<SystemClock>* result) {
+  auto clock = SystemClock::Default();
+  if (clock->IsInstanceOf(value)) {
+    *result = clock;
+    return Status::OK();
+  } else {
+#ifndef ROCKSDB_LITE
+    static std::once_flag once;
+    std::call_once(once, [&]() {
+      RegisterBuiltinSystemClocks(*(ObjectLibrary::Default().get()), "");
+    });
+#endif  // ROCKSDB_LITE
+    return LoadSharedObject<SystemClock>(config_options, value, nullptr,
+                                         result);
+  }
+}
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/env/env_basic_test.cc b/env/env_basic_test.cc
index 9cd8a11e7..40413b569 100644
--- a/env/env_basic_test.cc
+++ b/env/env_basic_test.cc
@@ -29,7 +29,7 @@ using CreateEnvFunc = Env*();
 static Env* GetDefaultEnv() { return Env::Default(); }
 
 static Env* GetMockEnv() {
-  static std::unique_ptr<Env> mock_env(new MockEnv(Env::Default()));
+  static std::unique_ptr<Env> mock_env(MockEnv::Create(Env::Default()));
   return mock_env.get();
 }
 #ifndef ROCKSDB_LITE
diff --git a/env/env_posix.cc b/env/env_posix.cc
index e6d657764..712e75d74 100644
--- a/env/env_posix.cc
+++ b/env/env_posix.cc
@@ -127,7 +127,10 @@ class PosixDynamicLibrary : public DynamicLibrary {
 
 class PosixClock : public SystemClock {
  public:
-  const char* Name() const override { return "PosixClock"; }
+  static const char* kClassName() { return "PosixClock"; }
+  const char* Name() const override { return kClassName(); }
+  const char* NickName() const override { return kDefaultName(); }
+
   uint64_t NowMicros() override {
     struct timeval tv;
     gettimeofday(&tv, nullptr);
diff --git a/env/env_test.cc b/env/env_test.cc
index dea6bbf87..3657e595c 100644
--- a/env/env_test.cc
+++ b/env/env_test.cc
@@ -36,6 +36,7 @@
 #endif
 
 #include "db/db_impl/db_impl.h"
+#include "env/emulated_clock.h"
 #include "env/env_chroot.h"
 #include "env/env_encryption_ctr.h"
 #include "env/unique_id.h"
@@ -47,6 +48,8 @@
 #include "rocksdb/env_encryption.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/system_clock.h"
+#include "rocksdb/utilities/object_registry.h"
+#include "test_util/mock_time_env.h"
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
@@ -2393,33 +2396,37 @@ TEST_F(EnvTest, EnvWriteVerificationTest) {
   ASSERT_OK(s);
 }
 
-#ifndef ROCKSDB_LITE
-class EncryptionProviderTest : public testing::Test {
+class CreateEnvTest : public testing::Test {
  public:
+  CreateEnvTest() {
+    config_options_.ignore_unknown_options = false;
+    config_options_.ignore_unsupported_options = false;
+  }
+  ConfigOptions config_options_;
 };
 
-TEST_F(EncryptionProviderTest, LoadCTRProvider) {
-  ConfigOptions config_options;
-  config_options.invoke_prepare_options = false;
+#ifndef ROCKSDB_LITE
+TEST_F(CreateEnvTest, LoadCTRProvider) {
+  config_options_.invoke_prepare_options = false;
   std::string CTR = CTREncryptionProvider::kClassName();
   std::shared_ptr<EncryptionProvider> provider;
   // Test a provider with no cipher
   ASSERT_OK(
-      EncryptionProvider::CreateFromString(config_options, CTR, &provider));
+      EncryptionProvider::CreateFromString(config_options_, CTR, &provider));
   ASSERT_NE(provider, nullptr);
   ASSERT_EQ(provider->Name(), CTR);
-  ASSERT_NOK(provider->PrepareOptions(config_options));
+  ASSERT_NOK(provider->PrepareOptions(config_options_));
   ASSERT_NOK(provider->ValidateOptions(DBOptions(), ColumnFamilyOptions()));
   auto cipher = provider->GetOptions<std::shared_ptr<BlockCipher>>("Cipher");
   ASSERT_NE(cipher, nullptr);
   ASSERT_EQ(cipher->get(), nullptr);
   provider.reset();
 
-  ASSERT_OK(EncryptionProvider::CreateFromString(config_options,
+  ASSERT_OK(EncryptionProvider::CreateFromString(config_options_,
                                                  CTR + "://test", &provider));
   ASSERT_NE(provider, nullptr);
   ASSERT_EQ(provider->Name(), CTR);
-  ASSERT_OK(provider->PrepareOptions(config_options));
+  ASSERT_OK(provider->PrepareOptions(config_options_));
   ASSERT_OK(provider->ValidateOptions(DBOptions(), ColumnFamilyOptions()));
   cipher = provider->GetOptions<std::shared_ptr<BlockCipher>>("Cipher");
   ASSERT_NE(cipher, nullptr);
@@ -2427,11 +2434,11 @@ TEST_F(EncryptionProviderTest, LoadCTRProvider) {
   ASSERT_STREQ(cipher->get()->Name(), "ROT13");
   provider.reset();
 
-  ASSERT_OK(EncryptionProvider::CreateFromString(config_options, "1://test",
+  ASSERT_OK(EncryptionProvider::CreateFromString(config_options_, "1://test",
                                                  &provider));
   ASSERT_NE(provider, nullptr);
   ASSERT_EQ(provider->Name(), CTR);
-  ASSERT_OK(provider->PrepareOptions(config_options));
+  ASSERT_OK(provider->PrepareOptions(config_options_));
   ASSERT_OK(provider->ValidateOptions(DBOptions(), ColumnFamilyOptions()));
   cipher = provider->GetOptions<std::shared_ptr<BlockCipher>>("Cipher");
   ASSERT_NE(cipher, nullptr);
@@ -2440,7 +2447,7 @@ TEST_F(EncryptionProviderTest, LoadCTRProvider) {
   provider.reset();
 
   ASSERT_OK(EncryptionProvider::CreateFromString(
-      config_options, "id=" + CTR + "; cipher=ROT13", &provider));
+      config_options_, "id=" + CTR + "; cipher=ROT13", &provider));
   ASSERT_NE(provider, nullptr);
   ASSERT_EQ(provider->Name(), CTR);
   cipher = provider->GetOptions<std::shared_ptr<BlockCipher>>("Cipher");
@@ -2450,15 +2457,66 @@ TEST_F(EncryptionProviderTest, LoadCTRProvider) {
   provider.reset();
 }
 
-TEST_F(EncryptionProviderTest, LoadROT13Cipher) {
-  ConfigOptions config_options;
+TEST_F(CreateEnvTest, LoadROT13Cipher) {
   std::shared_ptr<BlockCipher> cipher;
   // Test a provider with no cipher
-  ASSERT_OK(BlockCipher::CreateFromString(config_options, "ROT13", &cipher));
+  ASSERT_OK(BlockCipher::CreateFromString(config_options_, "ROT13", &cipher));
   ASSERT_NE(cipher, nullptr);
   ASSERT_STREQ(cipher->Name(), "ROT13");
 }
+#endif  // ROCKSDB_LITE
+
+TEST_F(CreateEnvTest, CreateDefaultSystemClock) {
+  std::shared_ptr<SystemClock> clock, copy;
+  ASSERT_OK(SystemClock::CreateFromString(config_options_,
+                                          SystemClock::kDefaultName(), &clock));
+  ASSERT_NE(clock, nullptr);
+  ASSERT_EQ(clock, SystemClock::Default());
+#ifndef ROCKSDB_LITE
+  std::string opts_str = clock->ToString(config_options_);
+  std::string mismatch;
+  ASSERT_OK(SystemClock::CreateFromString(config_options_, opts_str, &copy));
+  ASSERT_TRUE(clock->AreEquivalent(config_options_, copy.get(), &mismatch));
+#endif  // ROCKSDB_LITE
+}
+
+#ifndef ROCKSDB_LITE
+TEST_F(CreateEnvTest, CreateMockSystemClock) {
+  std::shared_ptr<SystemClock> mock, copy;
+
+  config_options_.registry->AddLibrary("test")->Register<SystemClock>(
+      MockSystemClock::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<SystemClock>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new MockSystemClock(nullptr));
+        return guard->get();
+      });
 
+  ASSERT_OK(SystemClock::CreateFromString(
+      config_options_, EmulatedSystemClock::kClassName(), &mock));
+  ASSERT_NE(mock, nullptr);
+  ASSERT_STREQ(mock->Name(), EmulatedSystemClock::kClassName());
+  ASSERT_EQ(mock->Inner(), SystemClock::Default().get());
+  std::string opts_str = mock->ToString(config_options_);
+  std::string mismatch;
+  ASSERT_OK(SystemClock::CreateFromString(config_options_, opts_str, &copy));
+  ASSERT_TRUE(mock->AreEquivalent(config_options_, copy.get(), &mismatch));
+
+  std::string id = std::string("id=") + EmulatedSystemClock::kClassName() +
+                   ";target=" + MockSystemClock::kClassName();
+
+  ASSERT_OK(SystemClock::CreateFromString(config_options_, id, &mock));
+  ASSERT_NE(mock, nullptr);
+  ASSERT_STREQ(mock->Name(), EmulatedSystemClock::kClassName());
+  ASSERT_NE(mock->Inner(), nullptr);
+  ASSERT_STREQ(mock->Inner()->Name(), MockSystemClock::kClassName());
+  ASSERT_EQ(mock->Inner()->Inner(), SystemClock::Default().get());
+  opts_str = mock->ToString(config_options_);
+  ASSERT_OK(SystemClock::CreateFromString(config_options_, opts_str, &copy));
+  ASSERT_TRUE(mock->AreEquivalent(config_options_, copy.get(), &mismatch));
+  ASSERT_OK(SystemClock::CreateFromString(
+      config_options_, EmulatedSystemClock::kClassName(), &mock));
+}
 #endif  // ROCKSDB_LITE
 
 namespace {
diff --git a/env/mock_env.cc b/env/mock_env.cc
index 3733371fc..e028f062e 100644
--- a/env/mock_env.cc
+++ b/env/mock_env.cc
@@ -12,21 +12,83 @@
 #include <algorithm>
 #include <chrono>
 
+#include "env/emulated_clock.h"
 #include "file/filename.h"
 #include "port/sys_time.h"
 #include "rocksdb/file_system.h"
+#include "rocksdb/utilities/options_type.h"
 #include "test_util/sync_point.h"
 #include "util/cast_util.h"
 #include "util/hash.h"
 #include "util/random.h"
 #include "util/rate_limiter.h"
+#include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
+namespace {
+int64_t MaybeCurrentTime(const std::shared_ptr<SystemClock>& clock) {
+  int64_t time = 1337346000;  // arbitrary fallback default
+  clock->GetCurrentTime(&time).PermitUncheckedError();
+  return time;
+}
+
+static std::unordered_map<std::string, OptionTypeInfo> time_elapse_type_info = {
+#ifndef ROCKSDB_LITE
+    {"time_elapse_only_sleep",
+     {0, OptionType::kBoolean, OptionVerificationType::kNormal,
+      OptionTypeFlags::kCompareNever,
+      [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+         const std::string& value, void* addr) {
+        auto clock = static_cast<EmulatedSystemClock*>(addr);
+        clock->SetTimeElapseOnlySleep(ParseBoolean("", value));
+        return Status::OK();
+      },
+      [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+         const void* addr, std::string* value) {
+        const auto clock = static_cast<const EmulatedSystemClock*>(addr);
+        *value = clock->IsTimeElapseOnlySleep() ? "true" : "false";
+        return Status::OK();
+      },
+      nullptr}},
+#endif  // ROCKSDB_LITE
+};
+static std::unordered_map<std::string, OptionTypeInfo> mock_sleep_type_info = {
+#ifndef ROCKSDB_LITE
+    {"mock_sleep",
+     {0, OptionType::kBoolean, OptionVerificationType::kNormal,
+      OptionTypeFlags::kCompareNever,
+      [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+         const std::string& value, void* addr) {
+        auto clock = static_cast<EmulatedSystemClock*>(addr);
+        clock->SetMockSleep(ParseBoolean("", value));
+        return Status::OK();
+      },
+      [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
+         const void* addr, std::string* value) {
+        const auto clock = static_cast<const EmulatedSystemClock*>(addr);
+        *value = clock->IsMockSleepEnabled() ? "true" : "false";
+        return Status::OK();
+      },
+      nullptr}},
+#endif  // ROCKSDB_LITE
+};
+}  // namespace
+
+EmulatedSystemClock::EmulatedSystemClock(
+    const std::shared_ptr<SystemClock>& base, bool time_elapse_only_sleep)
+    : SystemClockWrapper(base),
+      maybe_starting_time_(MaybeCurrentTime(base)),
+      time_elapse_only_sleep_(time_elapse_only_sleep),
+      no_slowdown_(time_elapse_only_sleep) {
+  RegisterOptions("", this, &time_elapse_type_info);
+  RegisterOptions("", this, &mock_sleep_type_info);
+}
 
 class MemFile {
  public:
-  explicit MemFile(Env* env, const std::string& fn, bool _is_lock_file = false)
-      : env_(env),
+  explicit MemFile(SystemClock* clock, const std::string& fn,
+                   bool _is_lock_file = false)
+      : clock_(clock),
         fn_(fn),
         refs_(0),
         is_lock_file_(_is_lock_file),
@@ -166,7 +228,7 @@ class MemFile {
  private:
   uint64_t Now() {
     int64_t unix_time = 0;
-    auto s = env_->GetCurrentTime(&unix_time);
+    auto s = clock_->GetCurrentTime(&unix_time);
     assert(s.ok());
     return static_cast<uint64_t>(unix_time);
   }
@@ -174,7 +236,7 @@ class MemFile {
   // Private since only Unref() should be used to delete it.
   ~MemFile() { assert(refs_ == 0); }
 
-  Env* env_;
+  SystemClock* clock_;
   const std::string fn_;
   mutable port::Mutex mutex_;
   int refs_;
@@ -403,20 +465,20 @@ class TestMemLogger : public Logger {
   std::atomic_size_t log_size_;
   static const uint64_t flush_every_seconds_ = 5;
   std::atomic_uint_fast64_t last_flush_micros_;
-  Env* env_;
+  SystemClock* clock_;
   IOOptions options_;
   IODebugContext* dbg_;
   std::atomic<bool> flush_pending_;
 
  public:
-  TestMemLogger(std::unique_ptr<FSWritableFile> f, Env* env,
+  TestMemLogger(std::unique_ptr<FSWritableFile> f, SystemClock* clock,
                 const IOOptions& options, IODebugContext* dbg,
                 const InfoLogLevel log_level = InfoLogLevel::ERROR_LEVEL)
       : Logger(log_level),
         file_(std::move(f)),
         log_size_(0),
         last_flush_micros_(0),
-        env_(env),
+        clock_(clock),
         options_(options),
         dbg_(dbg),
         flush_pending_(false) {}
@@ -426,7 +488,7 @@ class TestMemLogger : public Logger {
     if (flush_pending_) {
       flush_pending_ = false;
     }
-    last_flush_micros_ = env_->NowMicros();
+    last_flush_micros_ = clock_->NowMicros();
   }
 
   using Logger::Logv;
@@ -506,8 +568,11 @@ class TestMemLogger : public Logger {
 
 class MockFileSystem : public FileSystem {
  public:
-  explicit MockFileSystem(Env* env, bool supports_direct_io = true)
-      : env_(env), supports_direct_io_(supports_direct_io) {}
+  explicit MockFileSystem(const std::shared_ptr<SystemClock>& clock,
+                          bool supports_direct_io = true)
+      : system_clock_(clock), supports_direct_io_(supports_direct_io) {
+    clock_ = system_clock_.get();
+  }
 
   ~MockFileSystem() override {
     for (auto i = file_map_.begin(); i != file_map_.end(); ++i) {
@@ -620,12 +685,13 @@ class MockFileSystem : public FileSystem {
   // Map from filenames to MemFile objects, representing a simple file system.
   port::Mutex mutex_;
   std::map<std::string, MemFile*> file_map_;  // Protected by mutex_.
-  Env* env_;
+  std::shared_ptr<SystemClock> system_clock_;
+  SystemClock* clock_;
   bool supports_direct_io_;
 };
 
 }  // Anonymous namespace
-// Partial implementation of the Env interface.
+// Partial implementation of the FileSystem interface.
 IOStatus MockFileSystem::NewSequentialFile(
     const std::string& fname, const FileOptions& file_opts,
     std::unique_ptr<FSSequentialFile>* result, IODebugContext* /*dbg*/) {
@@ -705,7 +771,7 @@ IOStatus MockFileSystem::NewWritableFile(
   if (file_map_.find(fn) != file_map_.end()) {
     DeleteFileInternal(fn);
   }
-  MemFile* file = new MemFile(env_, fn, false);
+  MemFile* file = new MemFile(clock_, fn, false);
   file->Ref();
   file_map_[fn] = file;
   if (file_opts.use_direct_writes && !supports_direct_io_) {
@@ -723,7 +789,7 @@ IOStatus MockFileSystem::ReopenWritableFile(
   MutexLock lock(&mutex_);
   MemFile* file = nullptr;
   if (file_map_.find(fn) == file_map_.end()) {
-    file = new MemFile(env_, fn, false);
+    file = new MemFile(clock_, fn, false);
     // Only take a reference when we create the file objectt
     file->Ref();
     file_map_[fn] = file;
@@ -842,7 +908,7 @@ IOStatus MockFileSystem::CreateDir(const std::string& dirname,
   auto dn = NormalizeMockPath(dirname);
   MutexLock lock(&mutex_);
   if (file_map_.find(dn) == file_map_.end()) {
-    MemFile* file = new MemFile(env_, dn, false);
+    MemFile* file = new MemFile(clock_, dn, false);
     file->Ref();
     file_map_[dn] = file;
   } else {
@@ -965,14 +1031,14 @@ IOStatus MockFileSystem::NewLogger(const std::string& fname,
   auto iter = file_map_.find(fn);
   MemFile* file = nullptr;
   if (iter == file_map_.end()) {
-    file = new MemFile(env_, fn, false);
+    file = new MemFile(clock_, fn, false);
     file->Ref();
     file_map_[fn] = file;
   } else {
     file = iter->second;
   }
   std::unique_ptr<FSWritableFile> f(new MockWritableFile(file, FileOptions()));
-  result->reset(new TestMemLogger(std::move(f), env_, io_opts, dbg));
+  result->reset(new TestMemLogger(std::move(f), clock_, io_opts, dbg));
   return IOStatus::OK();
 }
 
@@ -990,7 +1056,7 @@ IOStatus MockFileSystem::LockFile(const std::string& fname,
         return IOStatus::IOError(fn, "lock is already held.");
       }
     } else {
-      auto* file = new MemFile(env_, fn, true);
+      auto* file = new MemFile(clock_, fn, true);
       file->Ref();
       file->Lock();
       file_map_[fn] = file;
@@ -1034,57 +1100,30 @@ Status MockFileSystem::CorruptBuffer(const std::string& fname) {
   iter->second->CorruptBuffer();
   return Status::OK();
 }
-namespace {
-class MockSystemClock : public SystemClockWrapper {
- public:
-  explicit MockSystemClock(const std::shared_ptr<SystemClock>& c)
-      : SystemClockWrapper(c), fake_sleep_micros_(0) {}
-
-  void FakeSleepForMicroseconds(int64_t micros) {
-    fake_sleep_micros_.fetch_add(micros);
-  }
 
-  const char* Name() const override { return "MockSystemClock"; }
+MockEnv::MockEnv(Env* env, const std::shared_ptr<FileSystem>& fs,
+                 const std::shared_ptr<SystemClock>& clock)
+    : CompositeEnvWrapper(env, fs, clock) {}
 
-  Status GetCurrentTime(int64_t* unix_time) override {
-    auto s = SystemClockWrapper::GetCurrentTime(unix_time);
-    if (s.ok()) {
-      auto fake_time = fake_sleep_micros_.load() / (1000 * 1000);
-      *unix_time += fake_time;
-    }
-    return s;
-  }
-
-  uint64_t NowMicros() override {
-    return SystemClockWrapper::NowMicros() + fake_sleep_micros_.load();
-  }
-
-  uint64_t NowNanos() override {
-    return SystemClockWrapper::NowNanos() + fake_sleep_micros_.load() * 1000;
-  }
+MockEnv* MockEnv::Create(Env* env) {
+  auto clock =
+      std::make_shared<EmulatedSystemClock>(env->GetSystemClock(), true);
+  return MockEnv::Create(env, clock);
+}
 
- private:
-  std::atomic<int64_t> fake_sleep_micros_;
-};
-}  // namespace
-MockEnv::MockEnv(Env* base_env)
-    : CompositeEnvWrapper(
-          base_env, std::make_shared<MockFileSystem>(this),
-          std::make_shared<MockSystemClock>(base_env->GetSystemClock())) {}
+MockEnv* MockEnv::Create(Env* env, const std::shared_ptr<SystemClock>& clock) {
+  auto fs = std::make_shared<MockFileSystem>(clock);
+  return new MockEnv(env, fs, clock);
+}
 
 Status MockEnv::CorruptBuffer(const std::string& fname) {
   auto mock = static_cast_with_check<MockFileSystem>(GetFileSystem().get());
   return mock->CorruptBuffer(fname);
 }
 
-void MockEnv::FakeSleepForMicroseconds(int64_t micros) {
-  auto mock = static_cast_with_check<MockSystemClock>(GetSystemClock().get());
-  mock->FakeSleepForMicroseconds(micros);
-}
-
 #ifndef ROCKSDB_LITE
 // This is to maintain the behavior before swithcing from InMemoryEnv to MockEnv
-Env* NewMemEnv(Env* base_env) { return new MockEnv(base_env); }
+Env* NewMemEnv(Env* base_env) { return MockEnv::Create(base_env); }
 
 #else  // ROCKSDB_LITE
 
diff --git a/env/mock_env.h b/env/mock_env.h
index 5e7faf55b..6c4b12f96 100644
--- a/env/mock_env.h
+++ b/env/mock_env.h
@@ -16,20 +16,18 @@
 #include "env/composite_env_wrapper.h"
 #include "rocksdb/env.h"
 #include "rocksdb/status.h"
+#include "rocksdb/system_clock.h"
 
 namespace ROCKSDB_NAMESPACE {
-
 class MockEnv : public CompositeEnvWrapper {
  public:
-  explicit MockEnv(Env* base_env);
+  static MockEnv* Create(Env* base);
+  static MockEnv* Create(Env* base, const std::shared_ptr<SystemClock>& clock);
 
   Status CorruptBuffer(const std::string& fname);
-
-  // Doesn't really sleep, just affects output of GetCurrentTime(), NowMicros()
-  // and NowNanos()
-  void FakeSleepForMicroseconds(int64_t micros);
-
  private:
+  MockEnv(Env* env, const std::shared_ptr<FileSystem>& fs,
+          const std::shared_ptr<SystemClock>& clock);
 };
 
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/env/mock_env_test.cc b/env/mock_env_test.cc
index 7f339540d..8e0d8ea8c 100644
--- a/env/mock_env_test.cc
+++ b/env/mock_env_test.cc
@@ -19,9 +19,7 @@ class MockEnvTest : public testing::Test {
   MockEnv* env_;
   const EnvOptions soptions_;
 
-  MockEnvTest()
-      : env_(new MockEnv(Env::Default())) {
-  }
+  MockEnvTest() : env_(MockEnv::Create(Env::Default())) {}
   ~MockEnvTest() override { delete env_; }
 };
 
@@ -68,7 +66,7 @@ TEST_F(MockEnvTest, FakeSleeping) {
   int64_t now = 0;
   auto s = env_->GetCurrentTime(&now);
   ASSERT_OK(s);
-  env_->FakeSleepForMicroseconds(3 * 1000 * 1000);
+  env_->SleepForMicroseconds(3 * 1000 * 1000);
   int64_t after_sleep = 0;
   s = env_->GetCurrentTime(&after_sleep);
   ASSERT_OK(s);
diff --git a/include/rocksdb/system_clock.h b/include/rocksdb/system_clock.h
index 3a52b42e2..456847a19 100644
--- a/include/rocksdb/system_clock.h
+++ b/include/rocksdb/system_clock.h
@@ -11,6 +11,7 @@
 
 #include <memory>
 
+#include "rocksdb/customizable.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "rocksdb/status.h"
 
@@ -24,15 +25,21 @@ struct ConfigOptions;
 
 // A SystemClock is an interface used by the rocksdb implementation to access
 // operating system time-related functionality.
-class SystemClock {
+class SystemClock : public Customizable {
  public:
   virtual ~SystemClock() {}
 
   static const char* Type() { return "SystemClock"; }
-
+  static Status CreateFromString(const ConfigOptions& options,
+                                 const std::string& value,
+                                 std::shared_ptr<SystemClock>* result);
   // The name of this system clock
   virtual const char* Name() const = 0;
 
+  // The name/nickname for the Default SystemClock.  This name can be used
+  // to determine if the clock is the default one.
+  static const char* kDefaultName() { return "DefaultClock"; }
+
   // Return a default SystemClock suitable for the current operating
   // system.
   static const std::shared_ptr<SystemClock>& Default();
@@ -73,8 +80,7 @@ class SystemClock {
 // of the SystemClock interface to the target/wrapped class.
 class SystemClockWrapper : public SystemClock {
  public:
-  explicit SystemClockWrapper(const std::shared_ptr<SystemClock>& t)
-      : target_(t) {}
+  explicit SystemClockWrapper(const std::shared_ptr<SystemClock>& t);
 
   uint64_t NowMicros() override { return target_->NowMicros(); }
 
@@ -96,6 +102,13 @@ class SystemClockWrapper : public SystemClock {
     return target_->TimeToString(time);
   }
 
+  Status PrepareOptions(const ConfigOptions& options) override;
+#ifndef ROCKSDB_LITE
+  std::string SerializeOptions(const ConfigOptions& config_options,
+                               const std::string& header) const override;
+#endif  // ROCKSDB_LITE
+  const Customizable* Inner() const override { return target_.get(); }
+
  protected:
   std::shared_ptr<SystemClock> target_;
 };
diff --git a/logging/auto_roll_logger_test.cc b/logging/auto_roll_logger_test.cc
index b9e8ed55a..19e7ea43f 100644
--- a/logging/auto_roll_logger_test.cc
+++ b/logging/auto_roll_logger_test.cc
@@ -20,6 +20,7 @@
 #include <vector>
 
 #include "db/db_test_util.h"
+#include "env/emulated_clock.h"
 #include "logging/logging.h"
 #include "port/port.h"
 #include "rocksdb/db.h"
@@ -30,25 +31,6 @@
 #include "test_util/testutil.h"
 
 namespace ROCKSDB_NAMESPACE {
-namespace {
-class NoSleepClock : public SystemClockWrapper {
- public:
-  NoSleepClock(
-      const std::shared_ptr<SystemClock>& base = SystemClock::Default())
-      : SystemClockWrapper(base) {}
-  const char* Name() const override { return "NoSleepClock"; }
-  void SleepForMicroseconds(int micros) override {
-    fake_time_ += static_cast<uint64_t>(micros);
-  }
-
-  uint64_t NowNanos() override { return fake_time_ * 1000; }
-
-  uint64_t NowMicros() override { return fake_time_; }
-
- private:
-  uint64_t fake_time_ = 6666666666;
-};
-}  // namespace
 
 // In this test we only want to Log some simple log message with
 // no format. LogMessage() provides such a simple interface and
@@ -219,7 +201,8 @@ TEST_F(AutoRollLoggerTest, RollLogFileBySize) {
 }
 
 TEST_F(AutoRollLoggerTest, RollLogFileByTime) {
-  auto nsc = std::make_shared<NoSleepClock>();
+  auto nsc =
+      std::make_shared<EmulatedSystemClock>(SystemClock::Default(), true);
 
   size_t time = 2;
   size_t log_size = 1024 * 5;
@@ -288,7 +271,8 @@ TEST_F(AutoRollLoggerTest, CompositeRollByTimeAndSizeLogger) {
 
   InitTestDb();
 
-  auto nsc = std::make_shared<NoSleepClock>();
+  auto nsc =
+      std::make_shared<EmulatedSystemClock>(SystemClock::Default(), true);
   AutoRollLogger logger(FileSystem::Default(), nsc, kTestDir, "", log_max_size,
                         time, keep_log_file_num);
 
@@ -306,7 +290,8 @@ TEST_F(AutoRollLoggerTest, CompositeRollByTimeAndSizeLogger) {
 // port
 TEST_F(AutoRollLoggerTest, CreateLoggerFromOptions) {
   DBOptions options;
-  auto nsc = std::make_shared<NoSleepClock>();
+  auto nsc =
+      std::make_shared<EmulatedSystemClock>(SystemClock::Default(), true);
   std::unique_ptr<Env> nse(new CompositeEnvWrapper(Env::Default(), nsc));
 
   std::shared_ptr<Logger> logger;
diff --git a/logging/env_logger_test.cc b/logging/env_logger_test.cc
index 375e2cf5b..b06e78588 100644
--- a/logging/env_logger_test.cc
+++ b/logging/env_logger_test.cc
@@ -5,7 +5,6 @@
 //
 
 #include "logging/env_logger.h"
-#include "env/mock_env.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 
diff --git a/monitoring/histogram_test.cc b/monitoring/histogram_test.cc
index fd7c00437..a37289365 100644
--- a/monitoring/histogram_test.cc
+++ b/monitoring/histogram_test.cc
@@ -31,11 +31,11 @@ void PopulateHistogram(Histogram& histogram,
     for (uint64_t i = low; i <= high; i++) {
       histogram.Add(i);
       // sleep a random microseconds [0-10)
-      clock->MockSleepForMicroseconds(rnd.Uniform(10));
+      clock->SleepForMicroseconds(rnd.Uniform(10));
     }
   }
   // make sure each data population at least take some time
-  clock->MockSleepForMicroseconds(1);
+  clock->SleepForMicroseconds(1);
 }
 
 void BasicOperation(Histogram& histogram) {
@@ -143,21 +143,21 @@ TEST_F(HistogramTest, HistogramWindowingExpire) {
       histogramWindowing(num_windows, micros_per_window, min_num_per_window);
   histogramWindowing.TEST_UpdateClock(clock);
   PopulateHistogram(histogramWindowing, 1, 1, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 100);
   ASSERT_EQ(histogramWindowing.min(), 1);
   ASSERT_EQ(histogramWindowing.max(), 1);
   ASSERT_EQ(histogramWindowing.Average(), 1);
 
   PopulateHistogram(histogramWindowing, 2, 2, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 200);
   ASSERT_EQ(histogramWindowing.min(), 1);
   ASSERT_EQ(histogramWindowing.max(), 2);
   ASSERT_EQ(histogramWindowing.Average(), 1.5);
 
   PopulateHistogram(histogramWindowing, 3, 3, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 300);
   ASSERT_EQ(histogramWindowing.min(), 1);
   ASSERT_EQ(histogramWindowing.max(), 3);
@@ -165,7 +165,7 @@ TEST_F(HistogramTest, HistogramWindowingExpire) {
 
   // dropping oldest window with value 1, remaining 2 ~ 4
   PopulateHistogram(histogramWindowing, 4, 4, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 300);
   ASSERT_EQ(histogramWindowing.min(), 2);
   ASSERT_EQ(histogramWindowing.max(), 4);
@@ -173,7 +173,7 @@ TEST_F(HistogramTest, HistogramWindowingExpire) {
 
   // dropping oldest window with value 2, remaining 3 ~ 5
   PopulateHistogram(histogramWindowing, 5, 5, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 300);
   ASSERT_EQ(histogramWindowing.min(), 3);
   ASSERT_EQ(histogramWindowing.max(), 5);
@@ -194,15 +194,15 @@ TEST_F(HistogramTest, HistogramWindowingMerge) {
 
   PopulateHistogram(histogramWindowing, 1, 1, 100);
   PopulateHistogram(otherWindowing, 1, 1, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
 
   PopulateHistogram(histogramWindowing, 2, 2, 100);
   PopulateHistogram(otherWindowing, 2, 2, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
 
   PopulateHistogram(histogramWindowing, 3, 3, 100);
   PopulateHistogram(otherWindowing, 3, 3, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
 
   histogramWindowing.Merge(otherWindowing);
   ASSERT_EQ(histogramWindowing.num(), 600);
@@ -212,14 +212,14 @@ TEST_F(HistogramTest, HistogramWindowingMerge) {
 
   // dropping oldest window with value 1, remaining 2 ~ 4
   PopulateHistogram(histogramWindowing, 4, 4, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 500);
   ASSERT_EQ(histogramWindowing.min(), 2);
   ASSERT_EQ(histogramWindowing.max(), 4);
 
   // dropping oldest window with value 2, remaining 3 ~ 5
   PopulateHistogram(histogramWindowing, 5, 5, 100);
-  clock->MockSleepForMicroseconds(micros_per_window);
+  clock->SleepForMicroseconds(micros_per_window);
   ASSERT_EQ(histogramWindowing.num(), 400);
   ASSERT_EQ(histogramWindowing.min(), 3);
   ASSERT_EQ(histogramWindowing.max(), 5);
diff --git a/options/customizable_test.cc b/options/customizable_test.cc
index 442edffcf..80b509c9d 100644
--- a/options/customizable_test.cc
+++ b/options/customizable_test.cc
@@ -28,6 +28,7 @@
 #include "rocksdb/utilities/options_type.h"
 #include "table/block_based/flush_block_policy.h"
 #include "table/mock_table.h"
+#include "test_util/mock_time_env.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
 #include "util/string_util.h"
@@ -1627,6 +1628,22 @@ TEST_F(LoadCustomizableTest, LoadEncryptionCipherTest) {
 }
 #endif  // !ROCKSDB_LITE
 
+TEST_F(LoadCustomizableTest, LoadSystemClockTest) {
+  std::shared_ptr<SystemClock> result;
+  ASSERT_NOK(SystemClock::CreateFromString(
+      config_options_, MockSystemClock::kClassName(), &result));
+  ASSERT_OK(SystemClock::CreateFromString(
+      config_options_, SystemClock::kDefaultName(), &result));
+  ASSERT_NE(result, nullptr);
+  ASSERT_TRUE(result->IsInstanceOf(SystemClock::kDefaultName()));
+  if (RegisterTests("Test")) {
+    ASSERT_OK(SystemClock::CreateFromString(
+        config_options_, MockSystemClock::kClassName(), &result));
+    ASSERT_NE(result, nullptr);
+    ASSERT_STREQ(result->Name(), MockSystemClock::kClassName());
+  }
+}
+
 TEST_F(LoadCustomizableTest, LoadFlushBlockPolicyFactoryTest) {
   std::shared_ptr<TableFactory> table;
   std::shared_ptr<FlushBlockPolicyFactory> result;
diff --git a/port/win/env_win.h b/port/win/env_win.h
index 82902d73c..e86515979 100644
--- a/port/win/env_win.h
+++ b/port/win/env_win.h
@@ -79,7 +79,9 @@ class WinClock : public SystemClock {
   WinClock();
   virtual ~WinClock() {}
 
-  const char* Name() const override { return "WindowsClock"; }
+  static const char* kClassName() { return "WindowsClock"; }
+  const char* Name() const override { return kClassName(); }
+  const char* NickName() const override { return kDefaultName(); }
 
   uint64_t NowMicros() override;
 
diff --git a/test_util/mock_time_env.h b/test_util/mock_time_env.h
index 61dc4e443..873b352d5 100644
--- a/test_util/mock_time_env.h
+++ b/test_util/mock_time_env.h
@@ -20,7 +20,8 @@ class MockSystemClock : public SystemClockWrapper {
   explicit MockSystemClock(const std::shared_ptr<SystemClock>& base)
       : SystemClockWrapper(base) {}
 
-  const char* Name() const override { return "MockSystemClock"; }
+  static const char* kClassName() { return "MockSystemClock"; }
+  const char* Name() const override { return kClassName(); }
   virtual Status GetCurrentTime(int64_t* time_sec) override {
     assert(time_sec != nullptr);
     *time_sec = static_cast<int64_t>(current_time_us_ / kMicrosInSecond);
@@ -50,7 +51,7 @@ class MockSystemClock : public SystemClockWrapper {
   // It's also similar to `set_current_time()`, which takes an absolute time in
   // seconds, vs. this one takes the sleep in microseconds.
   // Note: Not thread safe.
-  void MockSleepForMicroseconds(int micros) {
+  void SleepForMicroseconds(int micros) override {
     assert(micros >= 0);
     assert(current_time_us_ + static_cast<uint64_t>(micros) >=
            current_time_us_);
@@ -59,9 +60,8 @@ class MockSystemClock : public SystemClockWrapper {
 
   void MockSleepForSeconds(int seconds) {
     assert(seconds >= 0);
-    uint64_t micros = static_cast<uint64_t>(seconds) * kMicrosInSecond;
-    assert(current_time_us_ + micros >= current_time_us_);
-    current_time_us_.fetch_add(micros);
+    int micros = seconds * kMicrosInSecond;
+    SleepForMicroseconds(micros);
   }
 
   // TODO: this is a workaround for the different behavior on different platform
diff --git a/test_util/testutil.cc b/test_util/testutil.cc
index add1920e0..0950326d6 100644
--- a/test_util/testutil.cc
+++ b/test_util/testutil.cc
@@ -26,6 +26,7 @@
 #include "rocksdb/convenience.h"
 #include "rocksdb/system_clock.h"
 #include "rocksdb/utilities/object_registry.h"
+#include "test_util/mock_time_env.h"
 #include "test_util/sync_point.h"
 #include "util/random.h"
 
@@ -747,7 +748,13 @@ int RegisterTestObjects(ObjectLibrary& library, const std::string& /*arg*/) {
         guard->reset(new test::ChanglingCompactionFilterFactory(uri));
         return guard->get();
       });
-
+  library.Register<SystemClock>(
+      MockSystemClock::kClassName(),
+      [](const std::string& /*uri*/, std::unique_ptr<SystemClock>* guard,
+         std::string* /* errmsg */) {
+        guard->reset(new MockSystemClock(SystemClock::Default()));
+        return guard->get();
+      });
   return static_cast<int>(library.GetFactoryCount(&num_types));
 }
 
diff --git a/util/file_reader_writer_test.cc b/util/file_reader_writer_test.cc
index b7e1c2d41..1ab0246c2 100644
--- a/util/file_reader_writer_test.cc
+++ b/util/file_reader_writer_test.cc
@@ -691,7 +691,7 @@ std::string GenerateLine(int n) {
 TEST(LineFileReaderTest, LineFileReaderTest) {
   const int nlines = 1000;
 
-  std::unique_ptr<MockEnv> mem_env(new MockEnv(Env::Default()));
+  std::unique_ptr<Env> mem_env(MockEnv::Create(Env::Default()));
   std::shared_ptr<FileSystem> fs = mem_env->GetFileSystem();
   // Create an input file
   {
diff --git a/util/timer_test.cc b/util/timer_test.cc
index 3407fe9ee..a845e8ed7 100644
--- a/util/timer_test.cc
+++ b/util/timer_test.cc
@@ -36,7 +36,7 @@ TEST_F(TimerTest, SingleScheduleOnce) {
   ASSERT_EQ(0, count);
   // Wait for execution to finish
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
   ASSERT_EQ(1, count);
 
   ASSERT_TRUE(timer.Shutdown());
@@ -58,13 +58,13 @@ TEST_F(TimerTest, MultipleScheduleOnce) {
   ASSERT_EQ(0, count2);
 
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelay1Us); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelay1Us); });
 
   ASSERT_EQ(1, count1);
   ASSERT_EQ(0, count2);
 
   timer.TEST_WaitForRun([&] {
-    mock_clock_->MockSleepForMicroseconds(kInitDelay2Us - kInitDelay1Us);
+    mock_clock_->SleepForMicroseconds(kInitDelay2Us - kInitDelay1Us);
   });
 
   ASSERT_EQ(1, count1);
@@ -86,14 +86,14 @@ TEST_F(TimerTest, SingleScheduleRepeatedly) {
   ASSERT_EQ(0, count);
 
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
 
   ASSERT_EQ(1, count);
 
   // Wait for execution to finish
   for (int i = 1; i < kIterations; i++) {
     timer.TEST_WaitForRun(
-        [&] { mock_clock_->MockSleepForMicroseconds(kRepeatUs); });
+        [&] { mock_clock_->SleepForMicroseconds(kRepeatUs); });
   }
   ASSERT_EQ(kIterations, count);
 
@@ -126,7 +126,7 @@ TEST_F(TimerTest, MultipleScheduleRepeatedly) {
   // Wait for execution to finish
   for (int i = 1; i < kIterations * (kRepeatUs / kUsPerSec); i++) {
     timer.TEST_WaitForRun(
-        [&] { mock_clock_->MockSleepForMicroseconds(1 * kUsPerSec); });
+        [&] { mock_clock_->SleepForMicroseconds(1 * kUsPerSec); });
     ASSERT_EQ((i + 2) / (kRepeatUs / kUsPerSec), count1);
     ASSERT_EQ((i + 1) / (kRepeatUs / kUsPerSec), count2);
 
@@ -138,7 +138,7 @@ TEST_F(TimerTest, MultipleScheduleRepeatedly) {
 
   // Wait for execution to finish
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(1 * kUsPerSec); });
+      [&] { mock_clock_->SleepForMicroseconds(1 * kUsPerSec); });
   ASSERT_EQ(kIterations, count1);
   ASSERT_EQ(kIterations, count2);
   ASSERT_EQ(1, count3);
@@ -150,7 +150,7 @@ TEST_F(TimerTest, MultipleScheduleRepeatedly) {
 
   // execute the long interval one
   timer.TEST_WaitForRun([&] {
-    mock_clock_->MockSleepForMicroseconds(
+    mock_clock_->SleepForMicroseconds(
         kLargeRepeatUs - static_cast<int>(mock_clock_->NowMicros()));
   });
   ASSERT_EQ(2, count3);
@@ -178,12 +178,12 @@ TEST_F(TimerTest, AddAfterStartTest) {
   ASSERT_EQ(0, count);
   // Wait for execution to finish
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
   ASSERT_EQ(1, count);
 
   for (int i = 1; i < kIterations; i++) {
     timer.TEST_WaitForRun(
-        [&] { mock_clock_->MockSleepForMicroseconds(kRepeatUs); });
+        [&] { mock_clock_->SleepForMicroseconds(kRepeatUs); });
   }
   ASSERT_EQ(kIterations, count);
 
@@ -220,7 +220,7 @@ TEST_F(TimerTest, CancelRunningTask) {
     delete value;
     value = nullptr;
   });
-  mock_clock_->MockSleepForMicroseconds(kRepeatUs);
+  mock_clock_->SleepForMicroseconds(kRepeatUs);
   control_thr.join();
   ASSERT_TRUE(timer.Shutdown());
 }
@@ -258,7 +258,7 @@ TEST_F(TimerTest, ShutdownRunningTask) {
     TEST_SYNC_POINT("TimerTest::ShutdownRunningTest:BeforeShutdown");
     timer.Shutdown();
   });
-  mock_clock_->MockSleepForMicroseconds(kRepeatUs);
+  mock_clock_->SleepForMicroseconds(kRepeatUs);
   control_thr.join();
   delete value;
 }
@@ -288,14 +288,13 @@ TEST_F(TimerTest, AddSameFuncName) {
   ASSERT_EQ(0, func_counter2);
 
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
 
   ASSERT_EQ(0, func_counter1);
   ASSERT_EQ(1, func2_counter);
   ASSERT_EQ(1, func_counter2);
 
-  timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kRepeat1Us); });
+  timer.TEST_WaitForRun([&] { mock_clock_->SleepForMicroseconds(kRepeat1Us); });
 
   ASSERT_EQ(0, func_counter1);
   ASSERT_EQ(2, func2_counter);
@@ -315,14 +314,14 @@ TEST_F(TimerTest, RepeatIntervalWithFuncRunningTime) {
   int func_counter = 0;
   timer.Add(
       [&] {
-        mock_clock_->MockSleepForMicroseconds(kFuncRunningTimeUs);
+        mock_clock_->SleepForMicroseconds(kFuncRunningTimeUs);
         func_counter++;
       },
       "func", kInitDelayUs, kRepeatUs);
 
   ASSERT_EQ(0, func_counter);
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
   ASSERT_EQ(1, func_counter);
   ASSERT_EQ(kInitDelayUs + kFuncRunningTimeUs, mock_clock_->NowMicros());
 
@@ -338,7 +337,7 @@ TEST_F(TimerTest, RepeatIntervalWithFuncRunningTime) {
 
   // After the function running time, it's executed again
   timer.TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kFuncRunningTimeUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kFuncRunningTimeUs); });
   ASSERT_EQ(2, func_counter);
 
   ASSERT_TRUE(timer.Shutdown());
@@ -355,7 +354,7 @@ TEST_F(TimerTest, DestroyRunningTimer) {
   ASSERT_TRUE(timer_ptr->Start());
 
   timer_ptr->TEST_WaitForRun(
-      [&] { mock_clock_->MockSleepForMicroseconds(kInitDelayUs); });
+      [&] { mock_clock_->SleepForMicroseconds(kInitDelayUs); });
 
   // delete a running timer should not cause any exception
   delete timer_ptr;
@@ -389,7 +388,7 @@ TEST_F(TimerTest, DestroyTimerWithRunningFunc) {
     TEST_SYNC_POINT("TimerTest::DestroyTimerWithRunningFunc:BeforeDelete");
     delete timer_ptr;
   });
-  mock_clock_->MockSleepForMicroseconds(kRepeatUs);
+  mock_clock_->SleepForMicroseconds(kRepeatUs);
   control_thr.join();
 }