chore: use rdkafka compression (#90)

* Revert "feat: enable producer side compression for kafka (#82)"

This reverts commit 336e9ee.

* zstd

Author: wlawt

Cargo.lock

@@ -55,12 +55,11 @@ wiremock = "0.6.2"
 axum = "0.8.3"
 
 # Kafka and S3 dependencies
-rdkafka = { version = "0.37.0", features = ["libz-static", "ssl-vendored"] }
+rdkafka = { version = "0.37.0", features = ["zstd", "ssl-vendored"] }
 aws-config = "1.1.7"
 aws-sdk-s3 = "1.106.0"
 aws-credential-types = "1.1.7"
 bytes = { version = "1.8.0", features = ["serde"] }
-lz4_flex = "0.12"
 
 # tips-ingress
 backon = "1.5.2"

Cargo.toml

@@ -32,7 +32,6 @@ aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
 aws-credential-types = { workspace = true }
 bytes = { workspace = true }
-lz4_flex = { workspace = true }
 
 [dev-dependencies]
 testcontainers = { workspace = true }

crates/audit/Cargo.toml

@@ -26,8 +26,7 @@ impl KafkaBundleEventPublisher {
     async fn send_event(&self, event: &BundleEvent) -> Result<()> {
         let bundle_id = event.bundle_id();
         let key = event.generate_event_key();
-        let json_bytes = serde_json::to_vec(event)?;
-        let payload = lz4_flex::compress_prepend_size(&json_bytes);
+        let payload = serde_json::to_vec(event)?;
 
         let record = FutureRecord::to(&self.topic).key(&key).payload(&payload);
 

crates/audit/src/publisher.rs

@@ -77,9 +77,7 @@ impl EventReader for KafkaAuditLogReader {
                         .as_millis() as i64,
                 };
 
-                let json_bytes = lz4_flex::decompress_size_prepended(payload)
-                    .map_err(|e| anyhow::anyhow!("Failed to decompress LZ4: {e}"))?;
-                let event: BundleEvent = serde_json::from_slice(&json_bytes)?;
+                let event: BundleEvent = serde_json::from_slice(payload)?;
 
                 debug!(
                     bundle_id = %event.bundle_id(),

crates/audit/src/reader.rs

@@ -81,7 +81,7 @@ mod tests {
                 assert_eq!(user_operation.signature, Bytes::from_str("0x01").unwrap());
             }
             _ => {
-                panic!("Expected EntryPointV06, got {user_operation:?}");
+                panic!("Expected EntryPointV06, got {:?}", user_operation);
             }
         }
     }
@@ -132,7 +132,7 @@ mod tests {
                 assert_eq!(user_operation.pre_verification_gas, Uint::from(0x186a0));
             }
             _ => {
-                panic!("Expected EntryPointV07, got {user_operation:?}");
+                panic!("Expected EntryPointV07, got {:?}", user_operation);
             }
         }
     }

crates/core/src/user_ops_types.rs

@@ -1,3 +1,4 @@
 # Kafka configuration properties for ingress audit events
 bootstrap.servers=host.docker.internal:9094
 message.timeout.ms=5000
+compression.type=zstd