diff --git a/db/blob/blob_file_reader.cc b/db/blob/blob_file_reader.cc
index 0d05b5e571..447f090b50 100644
--- a/db/blob/blob_file_reader.cc
+++ b/db/blob/blob_file_reader.cc
@@ -250,7 +250,8 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
   Status s;
 
   IOOptions io_options;
-  s = file_reader->PrepareIOOptions(read_options, io_options);
+  IODebugContext dbg;
+  s = file_reader->PrepareIOOptions(read_options, io_options, &dbg);
   if (!s.ok()) {
     return s;
   }
@@ -259,13 +260,13 @@ Status BlobFileReader::ReadFromFile(const RandomAccessFileReader* file_reader,
     constexpr char* scratch = nullptr;
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, scratch,
-                          aligned_buf);
+                          aligned_buf, &dbg);
   } else {
     buf->reset(new char[read_size]);
     constexpr AlignedBuf* aligned_scratch = nullptr;
 
     s = file_reader->Read(io_options, read_offset, read_size, slice, buf->get(),
-                          aligned_scratch);
+                          aligned_scratch, &dbg);
   }
 
   if (!s.ok()) {
@@ -334,7 +335,8 @@ Status BlobFileReader::GetBlob(
     constexpr bool for_compaction = true;
 
     IOOptions io_options;
-    s = file_reader_->PrepareIOOptions(read_options, io_options);
+    IODebugContext dbg;
+    s = file_reader_->PrepareIOOptions(read_options, io_options, &dbg);
     if (!s.ok()) {
       return s;
     }
@@ -463,10 +465,11 @@ void BlobFileReader::MultiGetBlob(
   PERF_COUNTER_ADD(blob_read_count, num_blobs);
   PERF_COUNTER_ADD(blob_read_byte, total_len);
   IOOptions opts;
-  s = file_reader_->PrepareIOOptions(read_options, opts);
+  IODebugContext dbg;
+  s = file_reader_->PrepareIOOptions(read_options, opts, &dbg);
   if (s.ok()) {
     s = file_reader_->MultiRead(opts, read_reqs.data(), read_reqs.size(),
-                                direct_io ? &aligned_buf : nullptr);
+                                direct_io ? &aligned_buf : nullptr, &dbg);
   }
   if (!s.ok()) {
     for (auto& req : read_reqs) {
diff --git a/db/db_test.cc b/db/db_test.cc
index 81b4c2ed1b..cda3517d7d 100644
--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -144,6 +144,103 @@ TEST_F(DBTest, MockEnvTest) {
   delete db;
 }
 
+TEST_F(DBTest, RequestIdPlumbingTest) {
+  // test that request_id is passed to the filesystem, from
+  // ReadOptions to IODebugContext
+  Options options = CurrentOptions();
+  options.env = env_;
+
+  // Create a mock environment to capture IODebugContext during reads
+  const std::string* captured_request_id_dbg;
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "RandomAccessFileReader::Read:IODebugContext", [&](void* arg) {
+        IODebugContext* dbg = static_cast<IODebugContext*>(arg);
+        if (dbg == nullptr) {
+          captured_request_id_dbg = nullptr;
+        } else {
+          captured_request_id_dbg = dbg->request_id;
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(Put("k1", "v1"));
+  ASSERT_OK(Flush());
+
+  // test request_id plumbing during a get
+  {
+    const std::string test_request_id = "test_request_id_123";
+    ReadOptions read_opts;
+    read_opts.request_id = &test_request_id;
+    std::string value;
+    ASSERT_OK(db_->Get(read_opts, "k1", &value));
+
+    // Verify the request_id was propagated to the file system
+    ASSERT_NE(captured_request_id_dbg, nullptr);
+    ASSERT_EQ(*captured_request_id_dbg, test_request_id);
+  }
+
+  captured_request_id_dbg = nullptr;
+
+  // test request_id plumbing during iterator seek
+  ASSERT_OK(Put("k2", "v2"));
+  ASSERT_OK(Flush());
+  {
+    ReadOptions read_opts;
+    const std::string request_id = "test_request_id_456";
+    read_opts.request_id = &request_id;
+
+    std::unique_ptr<Iterator> iter(db_->NewIterator(read_opts));
+    iter->Seek("k2");
+    ASSERT_TRUE(iter->Valid());
+
+    // Verify the request_id was propagated to the file system
+    ASSERT_NE(captured_request_id_dbg, nullptr);
+    ASSERT_EQ(*captured_request_id_dbg, request_id);
+  }
+
+  // test request_id plumbing during multiget
+  captured_request_id_dbg = nullptr;
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "RandomAccessFileReader::MultiRead:IODebugContext", [&](void* arg) {
+        IODebugContext* dbg = static_cast<IODebugContext*>(arg);
+        if (dbg == nullptr) {
+          captured_request_id_dbg = nullptr;
+        } else {
+          captured_request_id_dbg = dbg->request_id;
+        }
+      });
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  ASSERT_OK(Put("k3", "v3"));
+  ASSERT_OK(Put("k4", "v4"));
+  ASSERT_OK(Flush());
+
+  {
+    ReadOptions read_opts;
+    const std::string multiget_request_id = "test_request_id_789";
+    read_opts.request_id = &multiget_request_id;
+
+    std::vector<std::string> values;
+    std::vector<Slice> keys = {Slice("k3"), Slice("k4")};
+
+    values.resize(keys.size());
+
+    std::vector<ColumnFamilyHandle*> cfhs(keys.size(),
+                                          db_->DefaultColumnFamily());
+    db_->MultiGet(read_opts, cfhs, keys, &values);
+
+    ASSERT_NE(captured_request_id_dbg, nullptr);
+    ASSERT_EQ(*captured_request_id_dbg, multiget_request_id);
+  }
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+}
+
 TEST_F(DBTest, MemEnvTest) {
   std::unique_ptr<Env> env{NewMemEnv(Env::Default())};
   Options options;
diff --git a/file/file_util.cc b/file/file_util.cc
index b3f6128aae..6e06ea0d95 100644
--- a/file/file_util.cc
+++ b/file/file_util.cc
@@ -247,15 +247,16 @@ IOStatus GenerateOneFileChecksum(
   Slice slice;
   uint64_t offset = 0;
   IOOptions opts;
-  io_s = reader->PrepareIOOptions(read_options, opts);
+  IODebugContext dbg;
+  io_s = reader->PrepareIOOptions(read_options, opts, &dbg);
   if (!io_s.ok()) {
     return io_s;
   }
   while (size > 0) {
     size_t bytes_to_read =
         static_cast<size_t>(std::min(uint64_t{readahead_size}, size));
-    io_s =
-        reader->Read(opts, offset, bytes_to_read, &slice, buf.get(), nullptr);
+    io_s = reader->Read(opts, offset, bytes_to_read, &slice, buf.get(), nullptr,
+                        &dbg);
     if (!io_s.ok()) {
       return IOStatus::Corruption("file read failed with error: " +
                                   io_s.ToString());
diff --git a/file/file_util.h b/file/file_util.h
index a8f20c8689..d19a4de6cd 100644
--- a/file/file_util.h
+++ b/file/file_util.h
@@ -86,7 +86,14 @@ IOStatus GenerateOneFileChecksum(
     const ReadOptions& read_options, Statistics* stats, SystemClock* clock);
 
 inline IOStatus PrepareIOFromReadOptions(const ReadOptions& ro,
-                                         SystemClock* clock, IOOptions& opts) {
+                                         SystemClock* clock, IOOptions& opts,
+                                         IODebugContext* dbg = nullptr) {
+  if (ro.request_id != nullptr) {
+    if (dbg != nullptr && dbg->request_id == nullptr) {
+      dbg->SetRequestId(ro.request_id);
+    }
+  }
+
   if (ro.deadline.count()) {
     std::chrono::microseconds now =
         std::chrono::microseconds(clock->NowMicros());
diff --git a/file/random_access_file_reader.cc b/file/random_access_file_reader.cc
index 46f5d1c262..c8edc86360 100644
--- a/file/random_access_file_reader.cc
+++ b/file/random_access_file_reader.cc
@@ -106,11 +106,14 @@ IOStatus RandomAccessFileReader::Create(
 
 IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
                                       size_t n, Slice* result, char* scratch,
-                                      AlignedBuf* aligned_buf) const {
+                                      AlignedBuf* aligned_buf,
+                                      IODebugContext* dbg) const {
   (void)aligned_buf;
   const Env::IOPriority rate_limiter_priority = opts.rate_limiter_priority;
 
   TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read", nullptr);
+  TEST_SYNC_POINT_CALLBACK("RandomAccessFileReader::Read:IODebugContext",
+                           const_cast<void*>(static_cast<void*>(dbg)));
 
   // To be paranoid: modify scratch a little bit, so in case underlying
   // FileSystem doesn't fill the buffer but return success and `scratch` returns
@@ -175,7 +178,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
           // the opts.timeout before calling file_->Read
           assert(!opts.timeout.count() || allowed == read_size);
           io_s = file_->Read(aligned_offset + buf.CurrentSize(), allowed, opts,
-                             &tmp, buf.Destination(), nullptr);
+                             &tmp, buf.Destination(), dbg);
         }
         if (ShouldNotifyListeners()) {
           auto finish_ts = FileOperationInfo::FinishNow();
@@ -237,7 +240,7 @@ IOStatus RandomAccessFileReader::Read(const IOOptions& opts, uint64_t offset,
           // the opts.timeout before calling file_->Read
           assert(!opts.timeout.count() || allowed == n);
           io_s = file_->Read(offset + pos, allowed, opts, &tmp_result,
-                             scratch + pos, nullptr);
+                             scratch + pos, dbg);
         }
         if (ShouldNotifyListeners()) {
           auto finish_ts = FileOperationInfo::FinishNow();
@@ -311,7 +314,8 @@ bool TryMerge(FSReadRequest* dest, const FSReadRequest& src) {
 IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
                                            FSReadRequest* read_reqs,
                                            size_t num_reqs,
-                                           AlignedBuf* aligned_buf) const {
+                                           AlignedBuf* aligned_buf,
+                                           IODebugContext* dbg) const {
   (void)aligned_buf;  // suppress warning of unused variable in LITE mode
   assert(num_reqs > 0);
 
@@ -420,8 +424,10 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
           remaining_bytes -= request_bytes;
         }
       }
-      io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts,
-                              /*IODebugContext*=*/nullptr);
+      TEST_SYNC_POINT_CALLBACK(
+          "RandomAccessFileReader::MultiRead:IODebugContext",
+          const_cast<void*>(static_cast<void*>(dbg)));
+      io_s = file_->MultiRead(fs_reqs, num_fs_reqs, opts, dbg);
       RecordInHistogram(stats_, MULTIGET_IO_BATCH_SIZE, num_fs_reqs);
     }
 
@@ -475,18 +481,21 @@ IOStatus RandomAccessFileReader::MultiRead(const IOOptions& opts,
 }
 
 IOStatus RandomAccessFileReader::PrepareIOOptions(const ReadOptions& ro,
-                                                  IOOptions& opts) const {
+                                                  IOOptions& opts,
+                                                  IODebugContext* dbg) const {
   if (clock_ != nullptr) {
-    return PrepareIOFromReadOptions(ro, clock_, opts);
+    return PrepareIOFromReadOptions(ro, clock_, opts, dbg);
   } else {
-    return PrepareIOFromReadOptions(ro, SystemClock::Default().get(), opts);
+    return PrepareIOFromReadOptions(ro, SystemClock::Default().get(), opts,
+                                    dbg);
   }
 }
 
 IOStatus RandomAccessFileReader::ReadAsync(
     FSReadRequest& req, const IOOptions& opts,
     std::function<void(FSReadRequest&, void*)> cb, void* cb_arg,
-    void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf) {
+    void** io_handle, IOHandleDeleter* del_fn, AlignedBuf* aligned_buf,
+    IODebugContext* dbg) {
   IOStatus s;
   // Create a callback and populate info.
   auto read_async_callback =
@@ -532,14 +541,14 @@ IOStatus RandomAccessFileReader::ReadAsync(
                  (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                  true /*delay_enabled*/);
     s = file_->ReadAsync(aligned_req, opts, read_async_callback,
-                         read_async_info, io_handle, del_fn, nullptr /*dbg*/);
+                         read_async_info, io_handle, del_fn, dbg);
   } else {
     StopWatch sw(clock_, stats_, hist_type_,
                  GetFileReadHistograms(stats_, opts.io_activity),
                  (stats_ != nullptr) ? &elapsed : nullptr, true /*overwrite*/,
                  true /*delay_enabled*/);
     s = file_->ReadAsync(req, opts, read_async_callback, read_async_info,
-                         io_handle, del_fn, nullptr /*dbg*/);
+                         io_handle, del_fn, dbg);
   }
   RecordTick(stats_, READ_ASYNC_MICROS, elapsed);
 
diff --git a/file/random_access_file_reader.h b/file/random_access_file_reader.h
index 945e685e3d..c1de6b973f 100644
--- a/file/random_access_file_reader.h
+++ b/file/random_access_file_reader.h
@@ -164,7 +164,8 @@ class RandomAccessFileReader {
   // the internally allocated buffer on return, and the result refers to a
   // region in aligned_buf.
   IOStatus Read(const IOOptions& opts, uint64_t offset, size_t n, Slice* result,
-                char* scratch, AlignedBuf* aligned_buf) const;
+                char* scratch, AlignedBuf* aligned_buf,
+                IODebugContext* dbg = nullptr) const;
 
   // REQUIRES:
   // num_reqs > 0, reqs do not overlap, and offsets in reqs are increasing.
@@ -172,10 +173,12 @@ class RandomAccessFileReader {
   // In direct IO mode, aligned_buf stores the aligned buffer allocated inside
   // MultiRead, the result Slices in reqs refer to aligned_buf.
   IOStatus MultiRead(const IOOptions& opts, FSReadRequest* reqs,
-                     size_t num_reqs, AlignedBuf* aligned_buf) const;
+                     size_t num_reqs, AlignedBuf* aligned_buf,
+                     IODebugContext* dbg = nullptr) const;
 
-  IOStatus Prefetch(const IOOptions& opts, uint64_t offset, size_t n) const {
-    return file_->Prefetch(offset, n, opts, nullptr);
+  IOStatus Prefetch(const IOOptions& opts, uint64_t offset, size_t n,
+                    IODebugContext* dbg = nullptr) const {
+    return file_->Prefetch(offset, n, opts, dbg);
   }
 
   FSRandomAccessFile* file() { return file_.get(); }
@@ -184,12 +187,13 @@ class RandomAccessFileReader {
 
   bool use_direct_io() const { return file_->use_direct_io(); }
 
-  IOStatus PrepareIOOptions(const ReadOptions& ro, IOOptions& opts) const;
+  IOStatus PrepareIOOptions(const ReadOptions& ro, IOOptions& opts,
+                            IODebugContext* dbg = nullptr) const;
 
   IOStatus ReadAsync(FSReadRequest& req, const IOOptions& opts,
                      std::function<void(FSReadRequest&, void*)> cb,
                      void* cb_arg, void** io_handle, IOHandleDeleter* del_fn,
-                     AlignedBuf* aligned_buf);
+                     AlignedBuf* aligned_buf, IODebugContext* dbg = nullptr);
 
   void ReadAsyncCallback(FSReadRequest& req, void* cb_arg);
 };
diff --git a/file/random_access_file_reader_test.cc b/file/random_access_file_reader_test.cc
index f081795b9d..717e985f1a 100644
--- a/file/random_access_file_reader_test.cc
+++ b/file/random_access_file_reader_test.cc
@@ -147,8 +147,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
     AlignedBuf aligned_buf;
-    ASSERT_OK(
-        r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
+    IODebugContext dbg;
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
+                           &dbg));
 
     AssertResult(content, reqs);
 
@@ -192,8 +193,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r1));
     reqs.push_back(std::move(r2));
     AlignedBuf aligned_buf;
-    ASSERT_OK(
-        r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
+    IODebugContext dbg;
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
+                           &dbg));
 
     AssertResult(content, reqs);
 
@@ -237,8 +239,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r1));
     reqs.push_back(std::move(r2));
     AlignedBuf aligned_buf;
-    ASSERT_OK(
-        r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
+    IODebugContext dbg;
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
+                           &dbg));
 
     AssertResult(content, reqs);
 
@@ -274,8 +277,9 @@ TEST_F(RandomAccessFileReaderTest, MultiReadDirectIO) {
     reqs.push_back(std::move(r0));
     reqs.push_back(std::move(r1));
     AlignedBuf aligned_buf;
-    ASSERT_OK(
-        r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf));
+    IODebugContext dbg;
+    ASSERT_OK(r->MultiRead(IOOptions(), reqs.data(), reqs.size(), &aligned_buf,
+                           &dbg));
 
     AssertResult(content, reqs);
 
diff --git a/include/rocksdb/file_system.h b/include/rocksdb/file_system.h
index ec10a5f126..cb6ecee9f2 100644
--- a/include/rocksdb/file_system.h
+++ b/include/rocksdb/file_system.h
@@ -220,6 +220,8 @@ struct FileOptions : EnvOptions {
 
 // A structure to pass back some debugging information from the FileSystem
 // implementation to RocksDB in case of an IO error
+// TODO(virajthakur): Update all calls to FS APIs for writes to pass in
+// IODebugContext
 struct IODebugContext {
   // file_path to be filled in by RocksDB in case of an error
   std::string file_path;
@@ -230,8 +232,9 @@ struct IODebugContext {
   // To be set by the FileSystem implementation
   std::string msg;
 
-  // To be set by the underlying FileSystem implementation.
-  std::string request_id;
+  // To be set by the application, to allow tracing logs/metrics from user ->
+  // RocksDB -> FS.
+  const std::string* request_id = nullptr;
 
   // In order to log required information in IO tracing for different
   // operations, Each bit in trace_data stores which corresponding info from
@@ -255,7 +258,7 @@ struct IODebugContext {
 
   // Called by underlying file system to set request_id and log request_id in
   // IOTracing.
-  void SetRequestId(const std::string& _request_id) {
+  void SetRequestId(const std::string* _request_id) {
     request_id = _request_id;
     trace_data |= (1 << TraceData::kRequestID);
   }
diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h
index 6e21fbe7fb..ba5b98147a 100644
--- a/include/rocksdb/options.h
+++ b/include/rocksdb/options.h
@@ -2062,6 +2062,19 @@ struct ReadOptions {
 
   // *** END options for RocksDB internal use only ***
 
+  // *** BEGIN per-request settings for internal team use only ***
+
+  // TODO: create a new struct for per-request options, potentially including
+  // timestamps in point lookups/scans
+
+  // request_id is a unique id assigned by the application. It is used to allow
+  // us to link file system metrics/logs to rocksDB and application logs. This
+  // request_id may not be unique to each RocksDB api call - it could refer to
+  // an application level request that results in multiple RocksDB api calls
+  const std::string* request_id = nullptr;
+
+  // *** END per-request settings for internal team use only ***
+
   ReadOptions() {}
   ReadOptions(bool _verify_checksums, bool _fill_cache);
   explicit ReadOptions(Env::IOActivity _io_activity);
diff --git a/table/block_based/block_based_table_reader.cc b/table/block_based/block_based_table_reader.cc
index ebb895027a..baab81a333 100644
--- a/table/block_based/block_based_table_reader.cc
+++ b/table/block_based/block_based_table_reader.cc
@@ -683,7 +683,8 @@ Status BlockBasedTable::Open(
   //    6. [meta block: index]
   //    7. [meta block: filter]
   IOOptions opts;
-  s = file->PrepareIOOptions(ro, opts);
+  IODebugContext dbg;
+  s = file->PrepareIOOptions(ro, opts, &dbg);
   if (s.ok()) {
     s = ReadFooterFromFile(opts, file.get(), *ioptions.fs,
                            prefetch_buffer.get(), file_size, &footer,
@@ -941,7 +942,8 @@ Status BlockBasedTable::PrefetchTail(
 #endif  // NDEBUG
 
   IOOptions opts;
-  Status s = file->PrepareIOOptions(ro, opts);
+  IODebugContext dbg;
+  Status s = file->PrepareIOOptions(ro, opts, &dbg);
   // Try file system prefetch
   if (s.ok() && !file->use_direct_io() && !force_direct_prefetch) {
     if (!file->Prefetch(opts, prefetch_off, prefetch_len).IsNotSupported()) {
diff --git a/table/block_based/block_based_table_reader_sync_and_async.h b/table/block_based/block_based_table_reader_sync_and_async.h
index c6263e150d..7c331cbe82 100644
--- a/table/block_based/block_based_table_reader_sync_and_async.h
+++ b/table/block_based/block_based_table_reader_sync_and_async.h
@@ -138,17 +138,18 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
   AlignedBuf direct_io_buf;
   {
     IOOptions opts;
-    IOStatus s = file->PrepareIOOptions(options, opts);
+    IODebugContext dbg;
+    IOStatus s = file->PrepareIOOptions(options, opts, &dbg);
     if (s.ok()) {
 #if defined(WITH_COROUTINES)
       if (file->use_direct_io()) {
 #endif  // WITH_COROUTINES
         s = file->MultiRead(opts, &read_reqs[0], read_reqs.size(),
-                            &direct_io_buf);
+                            &direct_io_buf, &dbg);
 #if defined(WITH_COROUTINES)
       } else {
         co_await batch->context()->reader().MultiReadAsync(
-            file, opts, &read_reqs[0], read_reqs.size(), &direct_io_buf);
+            file, opts, &read_reqs[0], read_reqs.size(), &direct_io_buf, &dbg);
       }
 #endif  // WITH_COROUTINES
     }
@@ -240,10 +241,11 @@ DEFINE_SYNC_AND_ASYNC(void, BlockBasedTable::RetrieveMultipleBlocks)
           // its not a memory mapped file
           Slice result;
           IOOptions opts;
-          IOStatus io_s = file->PrepareIOOptions(options, opts);
+          IODebugContext dbg;
+          IOStatus io_s = file->PrepareIOOptions(options, opts, &dbg);
           opts.verify_and_reconstruct_read = true;
           io_s = file->Read(opts, handle.offset(), BlockSizeWithTrailer(handle),
-                            &result, const_cast<char*>(data), nullptr);
+                            &result, const_cast<char*>(data), nullptr, &dbg);
           if (io_s.ok()) {
             assert(result.data() == data);
             assert(result.size() == BlockSizeWithTrailer(handle));
diff --git a/table/block_based/block_prefetcher.cc b/table/block_based/block_prefetcher.cc
index a4cfb027b2..38ec3a0441 100644
--- a/table/block_based/block_prefetcher.cc
+++ b/table/block_based/block_prefetcher.cc
@@ -39,7 +39,8 @@ void BlockPrefetcher::PrefetchIfNeeded(
         return;
       }
       IOOptions opts;
-      Status s = rep->file->PrepareIOOptions(read_options, opts);
+      IODebugContext dbg;
+      Status s = rep->file->PrepareIOOptions(read_options, opts, &dbg);
       if (!s.ok()) {
         return;
       }
diff --git a/table/block_based/partitioned_filter_block.cc b/table/block_based/partitioned_filter_block.cc
index a554364e50..42cfce462a 100644
--- a/table/block_based/partitioned_filter_block.cc
+++ b/table/block_based/partitioned_filter_block.cc
@@ -591,7 +591,8 @@ Status PartitionedFilterBlockReader::CacheDependencies(
                                   /*usage=*/FilePrefetchBufferUsage::kUnknown);
 
     IOOptions opts;
-    s = rep->file->PrepareIOOptions(ro, opts);
+    IODebugContext dbg;
+    s = rep->file->PrepareIOOptions(ro, opts, &dbg);
     if (s.ok()) {
       s = prefetch_buffer->Prefetch(opts, rep->file.get(), prefetch_off,
                                     static_cast<size_t>(prefetch_len));
diff --git a/table/block_fetcher.cc b/table/block_fetcher.cc
index af564063ca..6c73df23be 100644
--- a/table/block_fetcher.cc
+++ b/table/block_fetcher.cc
@@ -74,7 +74,8 @@ inline bool BlockFetcher::TryGetUncompressBlockFromPersistentCache() {
 inline bool BlockFetcher::TryGetFromPrefetchBuffer() {
   if (prefetch_buffer_ != nullptr) {
     IOOptions opts;
-    IOStatus io_s = file_->PrepareIOOptions(read_options_, opts);
+    IODebugContext dbg;
+    IOStatus io_s = file_->PrepareIOOptions(read_options_, opts, &dbg);
     if (io_s.ok()) {
       bool read_from_prefetch_buffer = prefetch_buffer_->TryReadFromCache(
           opts, file_, handle_.offset(), block_size_with_trailer_, &slice_,
@@ -246,7 +247,8 @@ inline void BlockFetcher::GetBlockContents() {
 void BlockFetcher::ReadBlock(bool retry) {
   FSReadRequest read_req;
   IOOptions opts;
-  io_status_ = file_->PrepareIOOptions(read_options_, opts);
+  IODebugContext dbg;
+  io_status_ = file_->PrepareIOOptions(read_options_, opts, &dbg);
   opts.verify_and_reconstruct_read = retry;
   read_req.status.PermitUncheckedError();
   // Actual file read
@@ -256,8 +258,9 @@ void BlockFetcher::ReadBlock(bool retry) {
       PERF_CPU_TIMER_GUARD(
           block_read_cpu_time,
           ioptions_.env ? ioptions_.env->GetSystemClock().get() : nullptr);
-      io_status_ = file_->Read(opts, handle_.offset(), block_size_with_trailer_,
-                               &slice_, /*scratch=*/nullptr, &direct_io_buf_);
+      io_status_ =
+          file_->Read(opts, handle_.offset(), block_size_with_trailer_, &slice_,
+                      /*scratch=*/nullptr, &direct_io_buf_, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
       used_buf_ = const_cast<char*>(slice_.data());
     } else if (use_fs_scratch_) {
@@ -269,7 +272,7 @@ void BlockFetcher::ReadBlock(bool retry) {
       read_req.len = block_size_with_trailer_;
       read_req.scratch = nullptr;
       io_status_ = file_->MultiRead(opts, &read_req, /*num_reqs=*/1,
-                                    /*AlignedBuf* =*/nullptr);
+                                    /*AlignedBuf* =*/nullptr, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
 
       slice_ = Slice(read_req.result.data(), read_req.result.size());
@@ -286,7 +289,7 @@ void BlockFetcher::ReadBlock(bool retry) {
       io_status_ =
           file_->Read(opts, handle_.offset(), /*size*/ block_size_with_trailer_,
                       /*result*/ &slice_, /*scratch*/ used_buf_,
-                      /*aligned_buf=*/nullptr);
+                      /*aligned_buf=*/nullptr, &dbg);
       PERF_COUNTER_ADD(block_read_count, 1);
 #ifndef NDEBUG
       if (slice_.data() == &stack_buf_[0]) {
@@ -417,7 +420,8 @@ IOStatus BlockFetcher::ReadAsyncBlockContents() {
     assert(prefetch_buffer_ != nullptr);
     if (!for_compaction_) {
       IOOptions opts;
-      IOStatus io_s = file_->PrepareIOOptions(read_options_, opts);
+      IODebugContext dbg;
+      IOStatus io_s = file_->PrepareIOOptions(read_options_, opts, &dbg);
       if (!io_s.ok()) {
         return io_s;
       }
diff --git a/table/meta_blocks.cc b/table/meta_blocks.cc
index 89be11be21..73764ae4bb 100644
--- a/table/meta_blocks.cc
+++ b/table/meta_blocks.cc
@@ -434,15 +434,16 @@ Status ReadTablePropertiesHelper(
       // If retrying, use a stronger file system read to check and correct
       // data corruption
       IOOptions opts;
-      if (PrepareIOFromReadOptions(ro, ioptions.clock, opts) !=
+      IODebugContext dbg;
+      if (PrepareIOFromReadOptions(ro, ioptions.clock, opts, &dbg) !=
           IOStatus::OK()) {
         return s;
       }
       opts.verify_and_reconstruct_read = true;
       std::unique_ptr<char[]> data(new char[len]);
       Slice result;
-      IOStatus io_s =
-          file->Read(opts, handle.offset(), len, &result, data.get(), nullptr);
+      IOStatus io_s = file->Read(opts, handle.offset(), len, &result,
+                                 data.get(), nullptr, &dbg);
       RecordTick(ioptions.stats, FILE_READ_CORRUPTION_RETRY_COUNT);
       if (!io_s.ok()) {
         ROCKS_LOG_INFO(ioptions.info_log,
@@ -574,8 +575,9 @@ Status ReadMetaIndexBlockInFile(RandomAccessFileReader* file,
                                 Footer* footer_out) {
   Footer footer;
   IOOptions opts;
+  IODebugContext dbg;
   Status s;
-  s = file->PrepareIOOptions(read_options, opts);
+  s = file->PrepareIOOptions(read_options, opts, &dbg);
   if (!s.ok()) {
     return s;
   }
diff --git a/trace_replay/io_tracer.cc b/trace_replay/io_tracer.cc
index a860130f85..e72b80c4f1 100644
--- a/trace_replay/io_tracer.cc
+++ b/trace_replay/io_tracer.cc
@@ -82,7 +82,7 @@ Status IOTraceWriter::WriteIOOp(const IOTraceRecord& record,
     uint32_t set_pos = static_cast<uint32_t>(log2(trace_data & -trace_data));
     switch (set_pos) {
       case IODebugContext::TraceData::kRequestID: {
-        Slice request_id(dbg->request_id);
+        Slice request_id(*dbg->request_id);
         PutLengthPrefixedSlice(&trace.payload, request_id);
       } break;
       default:
diff --git a/trace_replay/io_tracer_test.cc b/trace_replay/io_tracer_test.cc
index be3af4fb35..6946fa4be1 100644
--- a/trace_replay/io_tracer_test.cc
+++ b/trace_replay/io_tracer_test.cc
@@ -145,7 +145,8 @@ TEST_F(IOTracerTest, MultipleRecordsWithDifferentIOOpOptions) {
     // Write record with IODebugContext.
     io_op_data = 0;
     IODebugContext dbg;
-    dbg.SetRequestId("request_id_1");
+    const std::string test_request_id = "request_id_1";
+    dbg.SetRequestId(&test_request_id);
     IOTraceRecord record5(0, TraceType::kIOTracer, io_op_data,
                           GetFileOperation(5), 10 /*latency*/,
                           IOStatus::OK().ToString(), file_name);
diff --git a/unreleased_history/new_features/plumb_application_request_id_to_fs.md b/unreleased_history/new_features/plumb_application_request_id_to_fs.md
new file mode 100644
index 0000000000..3144ae8fd2
--- /dev/null
+++ b/unreleased_history/new_features/plumb_application_request_id_to_fs.md
@@ -0,0 +1,2 @@
+[internal team use only] 
+allow an application-defined request_id to be passed to RocksDB and propagated to the filesystem via IODebugContext
diff --git a/util/async_file_reader.cc b/util/async_file_reader.cc
index 8fa4d19933..67acc978b9 100644
--- a/util/async_file_reader.cc
+++ b/util/async_file_reader.cc
@@ -31,7 +31,7 @@ bool AsyncFileReader::MultiReadAsyncImpl(ReadAwaiter* awaiter) {
           }
         },
         &awaiter->read_reqs_[i], &awaiter->io_handle_[i], &awaiter->del_fn_[i],
-        /*aligned_buf=*/nullptr);
+        /*aligned_buf=*/nullptr, awaiter->dbg_);
     if (!s.ok()) {
       // For any non-ok status, the FileSystem will not call the callback
       // So let's update the status ourselves
diff --git a/util/async_file_reader.h b/util/async_file_reader.h
index 50a5951949..989f392cac 100644
--- a/util/async_file_reader.h
+++ b/util/async_file_reader.h
@@ -36,9 +36,10 @@ class AsyncFileReader {
                                             const IOOptions& opts,
                                             FSReadRequest* read_reqs,
                                             size_t num_reqs,
-                                            AlignedBuf* aligned_buf) noexcept {
-    return ReadOperation<ReadAwaiter>{*this,     file,     opts,
-                                      read_reqs, num_reqs, aligned_buf};
+                                            AlignedBuf* aligned_buf,
+                                            IODebugContext* dbg) noexcept {
+    return ReadOperation<ReadAwaiter>{*this,    file,        opts, read_reqs,
+                                      num_reqs, aligned_buf, dbg};
   }
 
  private:
@@ -49,12 +50,14 @@ class AsyncFileReader {
    public:
     explicit ReadAwaiter(AsyncFileReader& reader, RandomAccessFileReader* file,
                          const IOOptions& opts, FSReadRequest* read_reqs,
-                         size_t num_reqs, AlignedBuf* /*aligned_buf*/) noexcept
+                         size_t num_reqs, AlignedBuf* /*aligned_buf*/,
+                         IODebugContext* dbg) noexcept
         : reader_(reader),
           file_(file),
           opts_(opts),
           read_reqs_(read_reqs),
           num_reqs_(num_reqs),
+          dbg_(dbg),
           next_(nullptr) {}
 
     bool await_ready() noexcept { return false; }
@@ -82,6 +85,7 @@ class AsyncFileReader {
     const IOOptions& opts_;
     FSReadRequest* read_reqs_;
     size_t num_reqs_;
+    IODebugContext* dbg_;
     autovector<void*, 32> io_handle_;
     autovector<IOHandleDeleter, 32> del_fn_;
     folly::coro::impl::coroutine_handle<> awaiting_coro_;
@@ -101,18 +105,20 @@ class AsyncFileReader {
     explicit ReadOperation(AsyncFileReader& reader,
                            RandomAccessFileReader* file, const IOOptions& opts,
                            FSReadRequest* read_reqs, size_t num_reqs,
-                           AlignedBuf* aligned_buf) noexcept
+                           AlignedBuf* aligned_buf,
+                           IODebugContext* dbg) noexcept
         : reader_(reader),
           file_(file),
           opts_(opts),
           read_reqs_(read_reqs),
           num_reqs_(num_reqs),
-          aligned_buf_(aligned_buf) {}
+          aligned_buf_(aligned_buf),
+          dbg_(dbg) {}
 
     auto viaIfAsync(folly::Executor::KeepAlive<> executor) const {
       return folly::coro::co_viaIfAsync(
-          std::move(executor),
-          Awaiter{reader_, file_, opts_, read_reqs_, num_reqs_, aligned_buf_});
+          std::move(executor), Awaiter{reader_, file_, opts_, read_reqs_,
+                                       num_reqs_, aligned_buf_, dbg_});
     }
 
    private:
@@ -122,6 +128,7 @@ class AsyncFileReader {
     FSReadRequest* read_reqs_;
     size_t num_reqs_;
     AlignedBuf* aligned_buf_;
+    IODebugContext* dbg_;
   };
 
   // This function does the actual work when this awaitable starts execution