diff --git a/kvrocks.conf b/kvrocks.conf
index b6a2e0279f9..3c460c881e4 100644
--- a/kvrocks.conf
+++ b/kvrocks.conf
@@ -718,6 +718,16 @@ rocksdb.max_background_flushes -1
 # Default: 2
 rocksdb.max_subcompactions 2
 
+# If enabled WAL records will be compressed before they are written. Only
+# ZSTD (= kZSTD) is supported (until streaming support is adapted for other
+# compression types). Compressed WAL records will be read in supported
+# versions (>= RocksDB 7.4.0 for ZSTD) regardless of this setting when
+# the WAL is read.
+#
+# Accept value: "no", "zstd"
+# Default is no
+rocksdb.wal_compression no
+
 # In order to limit the size of WALs, RocksDB uses DBOptions::max_total_wal_size
 # as the trigger of column family flush. Once WALs exceed this size, RocksDB
 # will start forcing the flush of column families to allow deletion of some
diff --git a/src/config/config.cc b/src/config/config.cc
index 38b8fbaca54..93866bca88e 100644
--- a/src/config/config.cc
+++ b/src/config/config.cc
@@ -79,6 +79,15 @@ const std::vector<ConfigEnum<rocksdb::CompressionType>> compression_types{[] {
   return res;
 }()};
 
+const std::vector<ConfigEnum<rocksdb::CompressionType>> wal_compression_types{[] {
+  std::vector<ConfigEnum<rocksdb::CompressionType>> res;
+  res.reserve(engine::WalCompressionOptions.size());
+  for (const auto &e : engine::WalCompressionOptions) {
+    res.push_back({e.name, e.type});
+  }
+  return res;
+}()};
+
 const std::vector<ConfigEnum<BlockCacheType>> cache_types{[] {
   std::vector<ConfigEnum<BlockCacheType>> res;
   res.reserve(engine::CacheOptions.size());
@@ -211,6 +220,9 @@ Config::Config() {
       {"rocksdb.max_background_flushes", true, new IntField(&rocks_db.max_background_flushes, 2, -1, 32)},
       {"rocksdb.max_subcompactions", false, new IntField(&rocks_db.max_subcompactions, 2, 0, 16)},
       {"rocksdb.delayed_write_rate", false, new Int64Field(&rocks_db.delayed_write_rate, 0, 0, INT64_MAX)},
+      {"rocksdb.wal_compression", true,
+       new EnumField<rocksdb::CompressionType>(&rocks_db.wal_compression, wal_compression_types,
+                                               rocksdb::CompressionType::kNoCompression)},
       {"rocksdb.wal_ttl_seconds", true, new IntField(&rocks_db.wal_ttl_seconds, 3 * 3600, 0, INT_MAX)},
       {"rocksdb.wal_size_limit_mb", true, new IntField(&rocks_db.wal_size_limit_mb, 16384, 0, INT_MAX)},
       {"rocksdb.max_total_wal_size", false, new IntField(&rocks_db.max_total_wal_size, 64 * 4 * 2, 0, INT_MAX)},
diff --git a/src/config/config.h b/src/config/config.h
index 1b7311e4b18..5969ef5537d 100644
--- a/src/config/config.h
+++ b/src/config/config.h
@@ -192,6 +192,7 @@ struct Config {
     int64_t delayed_write_rate;
     int compaction_readahead_size;
     int target_file_size_base;
+    rocksdb::CompressionType wal_compression;
     int wal_ttl_seconds;
     int wal_size_limit_mb;
     int max_total_wal_size;
diff --git a/src/storage/storage.cc b/src/storage/storage.cc
index 116b452708e..9e1b315d073 100644
--- a/src/storage/storage.cc
+++ b/src/storage/storage.cc
@@ -168,6 +168,7 @@ rocksdb::Options Storage::InitRocksDBOptions() {
   options.num_levels = 7;
   options.compression_opts.level = config_->rocks_db.compression_level;
   options.compression_per_level.resize(options.num_levels);
+  options.wal_compression = config_->rocks_db.wal_compression;
   // only compress levels >= 2
   for (int i = 0; i < options.num_levels; ++i) {
     if (i < 2) {
diff --git a/src/storage/storage.h b/src/storage/storage.h
index 3ce9b78a548..c29b4689de5 100644
--- a/src/storage/storage.h
+++ b/src/storage/storage.h
@@ -94,6 +94,11 @@ inline const std::vector<CompressionOption> CompressionOptions = {
     {rocksdb::kZSTD, "zstd", "kZSTD"},
 };
 
+inline const std::vector<CompressionOption> WalCompressionOptions = {
+    {rocksdb::kNoCompression, "no", "kNoCompression"},
+    {rocksdb::kZSTD, "zstd", "kZSTD"},
+};
+
 struct CacheOption {
   BlockCacheType type;
   const std::string name;
diff --git a/tests/cppunit/config_test.cc b/tests/cppunit/config_test.cc
index 68919ed8415..dfe631189a9 100644
--- a/tests/cppunit/config_test.cc
+++ b/tests/cppunit/config_test.cc
@@ -128,6 +128,7 @@ TEST(Config, GetAndSet) {
       {"rocksdb.row_cache_size", "100"},
       {"rocksdb.rate_limiter_auto_tuned", "yes"},
       {"rocksdb.compression_level", "32767"},
+      {"rocksdb.wal_compression", "no"},
   };
   for (const auto &iter : immutable_cases) {
     s = config.Set(nullptr, iter.first, iter.second);