[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

@@ -8,8 +8,12 @@ license.workspace = true
 publish = false
 
 [features]
+# todo(azmy): remove "legacy-ingestion" from default features 
+# in v1.7.0
 default = []
 options_schema = ["dep:schemars"]
+# legacy-ingestion introduced in v1.6.0
+legacy-ingestion = []
 # support for OIDC based authentication via the Kafka consumer, building with rdkafka/curl-static does not work as it
 # fails with "SSL certificate OpenSSL verify result: unable to get local issuer certificate (20) (-1)" when trying to
 # obtain the OIDC token.
@@ -20,6 +24,7 @@ restate-workspace-hack = { workspace = true }
 
 restate-bifrost = { workspace = true }
 restate-core = { workspace = true }
+restate-ingestion-client = { workspace = true }
 restate-serde-util = { workspace = true }
 restate-storage-api = { workspace = true }
 restate-timer-queue = { workspace = true }
@@ -31,6 +36,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 }
@@ -39,14 +45,19 @@ parking_lot = { workspace = true }
 # https://github.com/fede1024/rust-rdkafka/pull/803. The PR bumps librdkafka to 2.12.1 and enables WITH_CURL for
 # librdkafka if the feature curl-static is enabled. Additionally, it cherry-picks https://github.com/confluentinc/librdkafka/pull/5182
 # which prevents pulling in curl if it is not activated. The additional fixes in fix-build-script fix the musl build.
-rdkafka = { version = "0.38", git = "https://github.com/restatedev/rust-rdkafka.git", rev = "e92cad90eff797a0dc29fa524cabb89b602ae234", features = ["libz-static", "cmake-build", "ssl-vendored"] }
+rdkafka = { version = "0.38", git = "https://github.com/restatedev/rust-rdkafka.git", rev = "e92cad90eff797a0dc29fa524cabb89b602ae234", features = [
+    "libz-static",
+    "cmake-build",
+    "ssl-vendored",
+] }
 schemars = { workspace = true, optional = true }
 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 }