[Ingress-Kafka] Refactor to use ingestion-client

Summary:
Refactor ingress-kafka to leverage on `ingestion-client` implementation. This replaces
the previous direct write to bifrost which allows:
- Batching, which increases throughput
- PP becomes the sole writer of its logs (WIP #3965)

Author: muhamadazmy

Cargo.lock

@@ -8,9 +8,55 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use bilrost::{
+    DecodeErrorKind,
+    encoding::{EmptyState, ForOverwrite, Proxiable},
+};
 use restate_encoding_derive::BilrostNewType;
 
-use crate::NetSerde;
+use crate::{NetSerde, bilrost_encodings::RestateEncoding};
+
+struct U128Tag;
+
+impl Proxiable<U128Tag> for u128 {
+    type Proxy = (u64, u64);
+
+    fn encode_proxy(&self) -> Self::Proxy {
+        ((*self >> 64) as u64, *self as u64)
+    }
+
+    fn decode_proxy(&mut self, proxy: Self::Proxy) -> Result<(), DecodeErrorKind> {
+        *self = (proxy.0 as u128) << 64 | proxy.1 as u128;
+        Ok(())
+    }
+}
+
+impl ForOverwrite<RestateEncoding, u128> for () {
+    fn for_overwrite() -> u128 {
+        0
+    }
+}
+
+impl EmptyState<RestateEncoding, u128> for () {
+    fn empty() -> u128 {
+        0
+    }
+
+    fn is_empty(val: &u128) -> bool {
+        *val == 0
+    }
+
+    fn clear(val: &mut u128) {
+        *val = 0;
+    }
+}
+
+bilrost::delegate_proxied_encoding!(
+    use encoding (::bilrost::encoding::General)
+    to encode proxied type (u128)
+    using proxy tag (U128Tag)
+    with encoding (RestateEncoding)
+);
 
 /// A Bilrost compatible U128 type.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, BilrostNewType)]

crates/encoding/src/common.rs

@@ -20,6 +20,7 @@ restate-workspace-hack = { workspace = true }
 
 restate-bifrost = { workspace = true }
 restate-core = { workspace = true }
+restate-ingestion-client = { workspace = true }
 restate-storage-api = { workspace = true }
 restate-timer-queue = { workspace = true }
 restate-types = { workspace = true }
@@ -30,6 +31,7 @@ anyhow = { workspace = true }
 base64 = { workspace = true }
 bytes = { workspace = true }
 derive_more = { workspace = true }
+futures = { workspace = true }
 metrics = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry_sdk = { workspace = true }
@@ -44,8 +46,9 @@ thiserror = { workspace = true }
 tokio = { workspace = true, features = ["sync", "rt"] }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
+xxhash-rust = { workspace = true, features = ["xxh3", "std"] }
 
 [dev-dependencies]
 restate-types = { workspace = true, features = ["test-util"] }
 
-base64 = { workspace = true }
+base64 = { workspace = true }