feat: enable ZSTD compression for write buffer payload

Closes #5981.

Cargo.lock

@@ -4283,10 +4283,10 @@ dependencies = [
  "parking_lot 0.12.1",
  "pin-project-lite",
  "rand",
- "snap",
  "thiserror",
  "tokio",
  "tracing",
+ "zstd",
 ]
 
 [[package]]

write_buffer/Cargo.toml

@@ -24,7 +24,7 @@ observability_deps = { path = "../observability_deps" }
 parking_lot = "0.12"
 pin-project = "1.0"
 prost = "0.11"
-rskafka = { git = "https://github.com/influxdata/rskafka.git", rev="8678dfe049de05415929ffec7c1be8921bb057f7", default-features = false, features = ["compression-snappy", "transport-socks5"] }
+rskafka = { git = "https://github.com/influxdata/rskafka.git", rev="8678dfe049de05415929ffec7c1be8921bb057f7", default-features = false, features = ["compression-zstd", "transport-socks5"] }
 schema = { path = "../schema" }
 tokio = { version = "1.21", features = ["fs", "macros", "parking_lot", "rt", "sync", "time"] }
 tokio-util = "0.7.4"

write_buffer/src/kafka/instrumentation.rs

@@ -211,7 +211,7 @@ mod tests {
         };
 
         wrapper
-            .produce(vec![record.clone()], Compression::Snappy)
+            .produce(vec![record.clone()], Compression::Zstd)
             .await
             .expect("produce call should succeed");
 
@@ -261,7 +261,7 @@ mod tests {
                 .with_time_provider(Arc::clone(&clock));
 
         wrapper
-            .produce(Vec::new(), Compression::Snappy)
+            .produce(Vec::new(), Compression::Zstd)
             .await
             .expect_err("produce call should fail");
 

write_buffer/src/kafka/mod.rs

@@ -691,7 +691,7 @@ mod tests {
                     headers: Default::default(),
                     timestamp: rskafka::chrono::Utc::now(),
                 }],
-                Compression::NoCompression,
+                Compression::Zstd,
             )
             .await
             .unwrap();