diff --git a/Makefile b/Makefile
index abe07c71d..b69b5286f 100644
--- a/Makefile
+++ b/Makefile
@@ -166,6 +166,12 @@ else
 	CXXFLAGS += -fno-rtti
 endif
 
+ifdef ASSERT_STATUS_CHECKED
+ifeq ($(filter -DROCKSDB_ASSERT_STATUS_CHECKED,$(OPT)),)
+	OPT += -DROCKSDB_ASSERT_STATUS_CHECKED
+endif
+endif
+
 $(warning Warning: Compiling in debug mode. Don't use the resulting binary in production)
 endif
 
@@ -657,6 +663,14 @@ PARALLEL_TEST = \
 ifdef COMPILE_WITH_UBSAN
         TESTS := $(shell echo $(TESTS) | sed 's/\boptions_settable_test\b//g')
 endif
+ifdef ASSERT_STATUS_CHECKED
+	# This is a new check for which we will add support incrementally. The
+	# whitelist can be removed once support is fully added.
+        TESTS_WHITELIST = \
+		options_test
+        TESTS := $(filter $(TESTS_WHITELIST),$(TESTS))
+        PARALLEL_TEST := $(filter $(TESTS_WHITELIST),$(PARALLEL_TEST))
+endif
 SUBSET := $(TESTS)
 ifdef ROCKSDBTESTS_START
         SUBSET := $(shell echo $(SUBSET) | sed 's/^.*$(ROCKSDBTESTS_START)/$(ROCKSDBTESTS_START)/')
diff --git a/file/writable_file_writer.h b/file/writable_file_writer.h
index 6f1d8987b..4a422a6bb 100644
--- a/file/writable_file_writer.h
+++ b/file/writable_file_writer.h
@@ -46,6 +46,7 @@ class WritableFileWriter {
     for (auto& listener : listeners_) {
       listener->OnFileWriteFinish(info);
     }
+    info.status.PermitUncheckedError();
   }
 #endif  // ROCKSDB_LITE
 
@@ -126,7 +127,10 @@ class WritableFileWriter {
 
   WritableFileWriter& operator=(const WritableFileWriter&) = delete;
 
-  ~WritableFileWriter() { Close(); }
+  ~WritableFileWriter() {
+    auto s = Close();
+    s.PermitUncheckedError();
+  }
 
   std::string file_name() const { return file_name_; }
 
diff --git a/hdfs/env_hdfs.h b/hdfs/env_hdfs.h
index 3a28aacd5..07c00e41b 100644
--- a/hdfs/env_hdfs.h
+++ b/hdfs/env_hdfs.h
@@ -238,8 +238,6 @@ class HdfsEnv : public Env {
 
 namespace ROCKSDB_NAMESPACE {
 
-static const Status notsup;
-
 class HdfsEnv : public Env {
 
  public:
@@ -260,79 +258,81 @@ class HdfsEnv : public Env {
       const std::string& /*fname*/,
       std::unique_ptr<RandomAccessFile>* /*result*/,
       const EnvOptions& /*options*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status NewWritableFile(const std::string& /*fname*/,
                                  std::unique_ptr<WritableFile>* /*result*/,
                                  const EnvOptions& /*options*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status NewDirectory(const std::string& /*name*/,
                               std::unique_ptr<Directory>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status FileExists(const std::string& /*fname*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status GetChildren(const std::string& /*path*/,
                              std::vector<std::string>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status DeleteFile(const std::string& /*fname*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status CreateDir(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status CreateDirIfMissing(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status DeleteDir(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status GetFileSize(const std::string& /*fname*/,
                              uint64_t* /*size*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status GetFileModificationTime(const std::string& /*fname*/,
                                          uint64_t* /*time*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status RenameFile(const std::string& /*src*/,
                             const std::string& /*target*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status LinkFile(const std::string& /*src*/,
                           const std::string& /*target*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status LockFile(const std::string& /*fname*/,
                           FileLock** /*lock*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
-  virtual Status UnlockFile(FileLock* /*lock*/) override { return notsup; }
+  virtual Status UnlockFile(FileLock* /*lock*/) override {
+    return Status::NotSupported();
+  }
 
   virtual Status NewLogger(const std::string& /*fname*/,
                            std::shared_ptr<Logger>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   Status IsDirectory(const std::string& /*path*/, bool* /*is_dir*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual void Schedule(void (* /*function*/)(void* arg), void* /*arg*/,
@@ -352,7 +352,7 @@ class HdfsEnv : public Env {
   }
 
   virtual Status GetTestDirectory(std::string* /*path*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual uint64_t NowMicros() override { return 0; }
@@ -360,16 +360,16 @@ class HdfsEnv : public Env {
   virtual void SleepForMicroseconds(int /*micros*/) override {}
 
   virtual Status GetHostName(char* /*name*/, uint64_t /*len*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status GetCurrentTime(int64_t* /*unix_time*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual Status GetAbsolutePath(const std::string& /*db_path*/,
                                  std::string* /*outputpath*/) override {
-    return notsup;
+    return Status::NotSupported();
   }
 
   virtual void SetBackgroundThreads(int /*number*/,
diff --git a/include/rocksdb/io_status.h b/include/rocksdb/io_status.h
index 6dd7be11f..8846753b9 100644
--- a/include/rocksdb/io_status.h
+++ b/include/rocksdb/io_status.h
@@ -170,6 +170,9 @@ inline IOStatus::IOStatus(Code _code, SubCode _subcode, const Slice& msg,
 }
 
 inline IOStatus::IOStatus(const IOStatus& s) : Status(s.code_, s.subcode_) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   retryable_ = s.retryable_;
   data_loss_ = s.data_loss_;
   scope_ = s.scope_;
@@ -179,6 +182,10 @@ inline IOStatus& IOStatus::operator=(const IOStatus& s) {
   // The following condition catches both aliasing (when this == &s),
   // and the common case where both s and *this are ok.
   if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     code_ = s.code_;
     subcode_ = s.subcode_;
     retryable_ = s.retryable_;
@@ -204,6 +211,10 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
 #endif
 {
   if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     code_ = std::move(s.code_);
     s.code_ = kOk;
     subcode_ = std::move(s.subcode_);
@@ -211,7 +222,7 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
     retryable_ = s.retryable_;
     data_loss_ = s.data_loss_;
     scope_ = s.scope_;
-    scope_ = kIOErrorScopeFileSystem;
+    s.scope_ = kIOErrorScopeFileSystem;
     delete[] state_;
     state_ = nullptr;
     std::swap(state_, s.state_);
@@ -220,10 +231,18 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
 }
 
 inline bool IOStatus::operator==(const IOStatus& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   return (code_ == rhs.code_);
 }
 
 inline bool IOStatus::operator!=(const IOStatus& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   return !(*this == rhs);
 }
 
diff --git a/include/rocksdb/status.h b/include/rocksdb/status.h
index ddddee949..8b4ef1dc2 100644
--- a/include/rocksdb/status.h
+++ b/include/rocksdb/status.h
@@ -25,7 +25,12 @@ class Status {
  public:
   // Create a success status.
   Status() : code_(kOk), subcode_(kNone), sev_(kNoError), state_(nullptr) {}
-  ~Status() { delete[] state_; }
+  ~Status() {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    assert(checked_);
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    delete[] state_;
+  }
 
   // Copy the specified status.
   Status(const Status& s);
@@ -43,6 +48,15 @@ class Status {
   bool operator==(const Status& rhs) const;
   bool operator!=(const Status& rhs) const;
 
+  // In case of intentionally swallowing an error, user must explicitly call
+  // this function. That way we are easily able to search the code to find where
+  // error swallowing occurs.
+  void PermitUncheckedError() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+  }
+
   enum Code : unsigned char {
     kOk = 0,
     kNotFound = 1,
@@ -63,7 +77,12 @@ class Status {
     kMaxCode
   };
 
-  Code code() const { return code_; }
+  Code code() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code_;
+  }
 
   enum SubCode : unsigned char {
     kNone = 0,
@@ -83,7 +102,12 @@ class Status {
     kMaxSubCode
   };
 
-  SubCode subcode() const { return subcode_; }
+  SubCode subcode() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return subcode_;
+  }
 
   enum Severity : unsigned char {
     kNoError = 0,
@@ -95,10 +119,20 @@ class Status {
   };
 
   Status(const Status& s, Severity sev);
-  Severity severity() const { return sev_; }
+  Severity severity() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return sev_;
+  }
 
   // Returns a C style string indicating the message of the Status
-  const char* getState() const { return state_; }
+  const char* getState() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return state_;
+  }
 
   // Return a success status.
   static Status OK() { return Status(); }
@@ -233,65 +267,156 @@ class Status {
   }
 
   // Returns true iff the status indicates success.
-  bool ok() const { return code() == kOk; }
+  bool ok() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kOk;
+  }
 
   // Returns true iff the status indicates success *with* something
   // overwritten
   bool IsOkOverwritten() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return code() == kOk && subcode() == kOverwritten;
   }
 
   // Returns true iff the status indicates a NotFound error.
-  bool IsNotFound() const { return code() == kNotFound; }
+  bool IsNotFound() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kNotFound;
+  }
 
   // Returns true iff the status indicates a Corruption error.
-  bool IsCorruption() const { return code() == kCorruption; }
+  bool IsCorruption() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kCorruption;
+  }
 
   // Returns true iff the status indicates a NotSupported error.
-  bool IsNotSupported() const { return code() == kNotSupported; }
+  bool IsNotSupported() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kNotSupported;
+  }
 
   // Returns true iff the status indicates an InvalidArgument error.
-  bool IsInvalidArgument() const { return code() == kInvalidArgument; }
+  bool IsInvalidArgument() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kInvalidArgument;
+  }
 
   // Returns true iff the status indicates an IOError.
-  bool IsIOError() const { return code() == kIOError; }
+  bool IsIOError() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kIOError;
+  }
 
   // Returns true iff the status indicates an MergeInProgress.
-  bool IsMergeInProgress() const { return code() == kMergeInProgress; }
+  bool IsMergeInProgress() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kMergeInProgress;
+  }
 
   // Returns true iff the status indicates Incomplete
-  bool IsIncomplete() const { return code() == kIncomplete; }
+  bool IsIncomplete() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kIncomplete;
+  }
 
   // Returns true iff the status indicates Shutdown In progress
-  bool IsShutdownInProgress() const { return code() == kShutdownInProgress; }
+  bool IsShutdownInProgress() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kShutdownInProgress;
+  }
 
-  bool IsTimedOut() const { return code() == kTimedOut; }
+  bool IsTimedOut() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kTimedOut;
+  }
 
-  bool IsAborted() const { return code() == kAborted; }
+  bool IsAborted() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kAborted;
+  }
 
   bool IsLockLimit() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return code() == kAborted && subcode() == kLockLimit;
   }
 
   // Returns true iff the status indicates that a resource is Busy and
   // temporarily could not be acquired.
-  bool IsBusy() const { return code() == kBusy; }
+  bool IsBusy() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kBusy;
+  }
 
-  bool IsDeadlock() const { return code() == kBusy && subcode() == kDeadlock; }
+  bool IsDeadlock() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kBusy && subcode() == kDeadlock;
+  }
 
   // Returns true iff the status indicated that the operation has Expired.
-  bool IsExpired() const { return code() == kExpired; }
+  bool IsExpired() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kExpired;
+  }
 
   // Returns true iff the status indicates a TryAgain error.
   // This usually means that the operation failed, but may succeed if
   // re-attempted.
-  bool IsTryAgain() const { return code() == kTryAgain; }
+  bool IsTryAgain() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kTryAgain;
+  }
 
   // Returns true iff the status indicates the proposed compaction is too large
-  bool IsCompactionTooLarge() const { return code() == kCompactionTooLarge; }
+  bool IsCompactionTooLarge() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kCompactionTooLarge;
+  }
 
   // Returns true iff the status indicates Column Family Dropped
-  bool IsColumnFamilyDropped() const { return code() == kColumnFamilyDropped; }
+  bool IsColumnFamilyDropped() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kColumnFamilyDropped;
+  }
 
   // Returns true iff the status indicates a NoSpace error
   // This is caused by an I/O error returning the specific "out of space"
@@ -299,6 +424,9 @@ class Status {
   // with a specific subcode, enabling users to take the appropriate action
   // if needed
   bool IsNoSpace() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return (code() == kIOError) && (subcode() == kNoSpace);
   }
 
@@ -306,6 +434,9 @@ class Status {
   // cases where we limit the memory used in certain operations (eg. the size
   // of a write batch) in order to avoid out of memory exceptions.
   bool IsMemoryLimit() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return (code() == kAborted) && (subcode() == kMemoryLimit);
   }
 
@@ -314,17 +445,26 @@ class Status {
   // directory" error condition. A PathNotFound error is an I/O error with
   // a specific subcode, enabling users to take appropriate action if necessary
   bool IsPathNotFound() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return (code() == kIOError) && (subcode() == kPathNotFound);
   }
 
   // Returns true iff the status indicates manual compaction paused. This
   // is caused by a call to PauseManualCompaction
   bool IsManualCompactionPaused() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return (code() == kIncomplete) && (subcode() == kManualCompactionPaused);
   }
 
   // Returns true iff the status indicates a TxnNotPrepared error.
   bool IsTxnNotPrepared() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     return (code() == kInvalidArgument) && (subcode() == kTxnNotPrepared);
   }
 
@@ -342,6 +482,9 @@ class Status {
   SubCode subcode_;
   Severity sev_;
   const char* state_;
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  mutable bool checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
 
   explicit Status(Code _code, SubCode _subcode = kNone)
       : code_(_code), subcode_(_subcode), sev_(kNoError), state_(nullptr) {}
@@ -355,16 +498,26 @@ class Status {
 
 inline Status::Status(const Status& s)
     : code_(s.code_), subcode_(s.subcode_), sev_(s.sev_) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);
 }
 inline Status::Status(const Status& s, Severity sev)
     : code_(s.code_), subcode_(s.subcode_), sev_(sev) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);
 }
 inline Status& Status::operator=(const Status& s) {
   // The following condition catches both aliasing (when this == &s),
   // and the common case where both s and *this are ok.
   if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     code_ = s.code_;
     subcode_ = s.subcode_;
     sev_ = s.sev_;
@@ -379,6 +532,9 @@ inline Status::Status(Status&& s)
     noexcept
 #endif
     : Status() {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   *this = std::move(s);
 }
 
@@ -388,6 +544,10 @@ inline Status& Status::operator=(Status&& s)
 #endif
 {
   if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
     code_ = std::move(s.code_);
     s.code_ = kOk;
     subcode_ = std::move(s.subcode_);
@@ -402,10 +562,18 @@ inline Status& Status::operator=(Status&& s)
 }
 
 inline bool Status::operator==(const Status& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   return (code_ == rhs.code_);
 }
 
 inline bool Status::operator!=(const Status& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   return !(*this == rhs);
 }
 
diff --git a/options/cf_options.cc b/options/cf_options.cc
index 3466b33e5..a3352c07b 100644
--- a/options/cf_options.cc
+++ b/options/cf_options.cc
@@ -459,8 +459,9 @@ std::unordered_map<std::string, OptionTypeInfo>
           [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
              const std::string& value, char* addr) {
             auto mop = reinterpret_cast<std::shared_ptr<MergeOperator>*>(addr);
-            ObjectRegistry::NewInstance()->NewSharedObject<MergeOperator>(value,
-                                                                          mop);
+            ObjectRegistry::NewInstance()
+                ->NewSharedObject<MergeOperator>(value, mop)
+                .PermitUncheckedError();
             return Status::OK();
           }}},
         {"compaction_style",
diff --git a/options/options_helper.cc b/options/options_helper.cc
index 33ff268a1..cecfd4d2b 100644
--- a/options/options_helper.cc
+++ b/options/options_helper.cc
@@ -1135,7 +1135,7 @@ Status GetTableFactoryFromMap(
       return s;
     }
     table_factory->reset(new BlockBasedTableFactory(bbt_opt));
-    return Status::OK();
+    return s;
   } else if (factory_name == PlainTableFactory().Name()) {
     PlainTableOptions pt_opt;
     s = GetPlainTableOptionsFromMap(config_options, PlainTableOptions(),
@@ -1144,12 +1144,12 @@ Status GetTableFactoryFromMap(
       return s;
     }
     table_factory->reset(new PlainTableFactory(pt_opt));
-    return Status::OK();
+    return s;
   }
   // Return OK for not supported table factories as TableFactory
   // Deserialization is optional.
   table_factory->reset();
-  return Status::OK();
+  return s;
 }
 
 std::unordered_map<std::string, EncodingType>
@@ -1306,9 +1306,9 @@ Status OptionTypeInfo::SerializeOption(const ConfigOptions& config_options,
   // we skip it in the serialization.
   Status s;
   if (opt_addr == nullptr || IsDeprecated()) {
-    return Status::OK();
+    s = Status::OK();
   } else if (string_func != nullptr) {
-    return string_func(config_options, opt_name, opt_addr, opt_value);
+    s = string_func(config_options, opt_name, opt_addr, opt_value);
   } else if (SerializeSingleOptionHelper(opt_addr, type, opt_value)) {
     s = Status::OK();
   } else {
diff --git a/options/options_parser.cc b/options/options_parser.cc
index 901a8bf9d..82dedda79 100644
--- a/options/options_parser.cc
+++ b/options/options_parser.cc
@@ -75,55 +75,68 @@ Status PersistRocksDBOptions(const ConfigOptions& config_options_in,
 
   std::string options_file_content;
 
-  writable->Append(option_file_header + "[" +
-                   opt_section_titles[kOptionSectionVersion] +
-                   "]\n"
-                   "  rocksdb_version=" +
-                   ToString(ROCKSDB_MAJOR) + "." + ToString(ROCKSDB_MINOR) +
-                   "." + ToString(ROCKSDB_PATCH) + "\n");
-  writable->Append("  options_file_version=" +
-                   ToString(ROCKSDB_OPTION_FILE_MAJOR) + "." +
-                   ToString(ROCKSDB_OPTION_FILE_MINOR) + "\n");
-  writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
-                   "]\n  ");
-
-  s = GetStringFromDBOptions(config_options, db_opt, &options_file_content);
-  if (!s.ok()) {
-    writable->Close();
-    return s;
+  s = writable->Append(option_file_header + "[" +
+                       opt_section_titles[kOptionSectionVersion] +
+                       "]\n"
+                       "  rocksdb_version=" +
+                       ToString(ROCKSDB_MAJOR) + "." + ToString(ROCKSDB_MINOR) +
+                       "." + ToString(ROCKSDB_PATCH) + "\n");
+  if (s.ok()) {
+    s = writable->Append(
+        "  options_file_version=" + ToString(ROCKSDB_OPTION_FILE_MAJOR) + "." +
+        ToString(ROCKSDB_OPTION_FILE_MINOR) + "\n");
+  }
+  if (s.ok()) {
+    s = writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
+                         "]\n  ");
   }
-  writable->Append(options_file_content + "\n");
 
-  for (size_t i = 0; i < cf_opts.size(); ++i) {
+  if (s.ok()) {
+    s = GetStringFromDBOptions(config_options, db_opt, &options_file_content);
+  }
+  if (s.ok()) {
+    s = writable->Append(options_file_content + "\n");
+  }
+
+  for (size_t i = 0; s.ok() && i < cf_opts.size(); ++i) {
     // CFOptions section
-    writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
-                     " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
-    s = GetStringFromColumnFamilyOptions(config_options, cf_opts[i],
-                                         &options_file_content);
-    if (!s.ok()) {
-      writable->Close();
-      return s;
+    s = writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
+                         " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
+    if (s.ok()) {
+      s = GetStringFromColumnFamilyOptions(config_options, cf_opts[i],
+                                           &options_file_content);
+    }
+    if (s.ok()) {
+      s = writable->Append(options_file_content + "\n");
     }
-    writable->Append(options_file_content + "\n");
     // TableOptions section
     auto* tf = cf_opts[i].table_factory.get();
     if (tf != nullptr) {
-      writable->Append("[" + opt_section_titles[kOptionSectionTableOptions] +
-                       tf->Name() + " \"" + EscapeOptionString(cf_names[i]) +
-                       "\"]\n  ");
-      options_file_content.clear();
-      s = tf->GetOptionString(config_options, &options_file_content);
-      if (!s.ok()) {
-        return s;
+      if (s.ok()) {
+        s = writable->Append(
+            "[" + opt_section_titles[kOptionSectionTableOptions] + tf->Name() +
+            " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
+      }
+      if (s.ok()) {
+        options_file_content.clear();
+        s = tf->GetOptionString(config_options, &options_file_content);
+      }
+      if (s.ok()) {
+        s = writable->Append(options_file_content + "\n");
       }
-      writable->Append(options_file_content + "\n");
     }
   }
-  writable->Sync(true /* use_fsync */);
-  writable->Close();
-
-  return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
-      config_options, db_opt, cf_names, cf_opts, file_name, fs);
+  if (s.ok()) {
+    s = writable->Sync(true /* use_fsync */);
+  }
+  if (s.ok()) {
+    s = writable->Close();
+  }
+  if (s.ok()) {
+    return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+        config_options, db_opt, cf_names, cf_opts, file_name, fs);
+  }
+  return s;
 }
 
 RocksDBOptionsParser::RocksDBOptionsParser() { Reset(); }
@@ -464,7 +477,7 @@ Status RocksDBOptionsParser::EndSection(
       }
     }
   }
-  return Status::OK();
+  return s;
 }
 
 Status RocksDBOptionsParser::ValidityCheck() {
@@ -609,15 +622,35 @@ Status RocksDBOptionsParser::VerifyDBOptions(
         char buffer[kBufferSize];
         std::string base_value;
         std::string file_value;
-        opt_info.SerializeOption(config_options, pair.first, base_addr,
-                                 &base_value);
-        opt_info.SerializeOption(config_options, pair.first, file_addr,
-                                 &file_value);
+        int offset =
+            snprintf(buffer, sizeof(buffer),
+                     "[RocksDBOptionsParser]: "
+                     "failed the verification on ColumnFamilyOptions::%s",
+                     pair.first.c_str());
+        Status s = opt_info.SerializeOption(config_options, pair.first,
+                                            base_addr, &base_value);
+        if (s.ok()) {
+          s = opt_info.SerializeOption(config_options, pair.first, file_addr,
+                                       &file_value);
+        }
         snprintf(buffer, sizeof(buffer),
                  "[RocksDBOptionsParser]: "
                  "failed the verification on DBOptions::%s --- "
                  "The specified one is %s while the persisted one is %s.\n",
                  pair.first.c_str(), base_value.c_str(), file_value.c_str());
+        assert(offset >= 0);
+        assert(static_cast<size_t>(offset) < sizeof(buffer));
+        if (s.ok()) {
+          snprintf(
+              buffer + offset, sizeof(buffer) - static_cast<size_t>(offset),
+              "--- The specified one is %s while the persisted one is %s.\n",
+              base_value.c_str(), file_value.c_str());
+        } else {
+          snprintf(buffer + offset,
+                   sizeof(buffer) - static_cast<size_t>(offset),
+                   "--- Unable to re-serialize an option: %s.\n",
+                   s.ToString().c_str());
+        }
         return Status::InvalidArgument(Slice(buffer, strlen(buffer)));
       }
     }
@@ -659,15 +692,30 @@ Status RocksDBOptionsParser::VerifyCFOptions(
         char buffer[kBufferSize];
         std::string base_value;
         std::string file_value;
-        opt_info.SerializeOption(config_options, pair.first, base_addr,
-                                 &base_value);
-        opt_info.SerializeOption(config_options, pair.first, file_addr,
-                                 &file_value);
-        snprintf(buffer, sizeof(buffer),
-                 "[RocksDBOptionsParser]: "
-                 "failed the verification on ColumnFamilyOptions::%s --- "
-                 "The specified one is %s while the persisted one is %s.\n",
-                 pair.first.c_str(), base_value.c_str(), file_value.c_str());
+        Status s = opt_info.SerializeOption(config_options, pair.first,
+                                            base_addr, &base_value);
+        if (s.ok()) {
+          s = opt_info.SerializeOption(config_options, pair.first, file_addr,
+                                       &file_value);
+        }
+        int offset =
+            snprintf(buffer, sizeof(buffer),
+                     "[RocksDBOptionsParser]: "
+                     "failed the verification on ColumnFamilyOptions::%s",
+                     pair.first.c_str());
+        assert(offset >= 0);
+        assert(static_cast<size_t>(offset) < sizeof(buffer));
+        if (s.ok()) {
+          snprintf(
+              buffer + offset, sizeof(buffer) - static_cast<size_t>(offset),
+              "--- The specified one is %s while the persisted one is %s.\n",
+              base_value.c_str(), file_value.c_str());
+        } else {
+          snprintf(buffer + offset,
+                   sizeof(buffer) - static_cast<size_t>(offset),
+                   "--- Unable to re-serialize an option: %s.\n",
+                   s.ToString().c_str());
+        }
         return Status::InvalidArgument(Slice(buffer, sizeof(buffer)));
       }  // if (! matches)
     }    // CheckSanityLevel
diff --git a/options/options_test.cc b/options/options_test.cc
index 4002f4630..c34fb8666 100644
--- a/options/options_test.cc
+++ b/options/options_test.cc
@@ -2101,7 +2101,7 @@ TEST_F(OptionsParserTest, Comment) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_OK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2132,7 +2132,7 @@ TEST_F(OptionsParserTest, ExtraSpace) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_OK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2150,7 +2150,7 @@ TEST_F(OptionsParserTest, MissingDBOptions) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_NOK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2180,7 +2180,7 @@ TEST_F(OptionsParserTest, DoubleDBOptions) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_NOK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2208,7 +2208,7 @@ TEST_F(OptionsParserTest, NoDefaultCFOptions) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_NOK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2238,7 +2238,7 @@ TEST_F(OptionsParserTest, DefaultCFOptionsMustBeTheFirst) {
       "  # if a section is blank, we will use the default\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_NOK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2267,7 +2267,7 @@ TEST_F(OptionsParserTest, DuplicateCFOptions) {
       "[CFOptions \"something_else\"]\n";
 
   const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
   RocksDBOptionsParser parser;
   ASSERT_NOK(
       parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@@ -2335,8 +2335,12 @@ TEST_F(OptionsParserTest, IgnoreUnknownOptions) {
         "  # if a section is blank, we will use the default\n";
 
     const std::string kTestFileName = "test-rocksdb-options.ini";
-    env_->DeleteFile(kTestFileName);
-    env_->WriteToNewFile(kTestFileName, options_file_content);
+    auto s = env_->FileExists(kTestFileName);
+    ASSERT_TRUE(s.ok() || s.IsNotFound());
+    if (s.ok()) {
+      ASSERT_OK(env_->DeleteFile(kTestFileName));
+    }
+    ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
     RocksDBOptionsParser parser;
     ASSERT_NOK(parser.Parse(kTestFileName, fs_.get(), false,
                             4096 /* readahead_size */));
@@ -2384,7 +2388,7 @@ TEST_F(OptionsParserTest, ParseVersion) {
     snprintf(buffer, kLength - 1, file_template.c_str(), iv.c_str());
 
     parser.Reset();
-    env_->WriteToNewFile(iv, buffer);
+    ASSERT_OK(env_->WriteToNewFile(iv, buffer));
     ASSERT_NOK(parser.Parse(iv, fs_.get(), false, 0 /* readahead_size */));
   }
 
@@ -2393,7 +2397,7 @@ TEST_F(OptionsParserTest, ParseVersion) {
   for (auto vv : valid_versions) {
     snprintf(buffer, kLength - 1, file_template.c_str(), vv.c_str());
     parser.Reset();
-    env_->WriteToNewFile(vv, buffer);
+    ASSERT_OK(env_->WriteToNewFile(vv, buffer));
     ASSERT_OK(parser.Parse(vv, fs_.get(), false, 0 /* readahead_size */));
   }
 }
diff --git a/test_util/testutil.h b/test_util/testutil.h
index 83779b538..8cb62a2fb 100644
--- a/test_util/testutil.h
+++ b/test_util/testutil.h
@@ -594,14 +594,17 @@ inline std::string EncodeInt(uint64_t x) {
                                 const std::string& content) {
       std::unique_ptr<WritableFile> r;
       auto s = NewWritableFile(file_name, &r, EnvOptions());
-      if (!s.ok()) {
-        return s;
+      if (s.ok()) {
+        s = r->Append(content);
       }
-      r->Append(content);
-      r->Flush();
-      r->Close();
-      assert(files_[file_name] == content);
-      return Status::OK();
+      if (s.ok()) {
+        s = r->Flush();
+      }
+      if (s.ok()) {
+        s = r->Close();
+      }
+      assert(!s.ok() || files_[file_name] == content);
+      return s;
     }
 
     // The following text is boilerplate that forwards all methods to target()
diff --git a/util/status.cc b/util/status.cc
index 8adb01d0c..ace4f1411 100644
--- a/util/status.cc
+++ b/util/status.cc
@@ -75,6 +75,9 @@ Status::Status(Code _code, SubCode _subcode, const Slice& msg,
 }
 
 std::string Status::ToString() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
   char tmp[30];
   const char* type;
   switch (code_) {