feat(sink): kafka async truncate log (#12587)

src/common/src/util/future_utils.rs

@@ -14,6 +14,7 @@
 
 use std::future::pending;
 
+use futures::future::Either;
 use futures::{Future, FutureExt, Stream};
 
 /// Convert a list of streams into a [`Stream`] of results from the streams.
@@ -33,3 +34,12 @@ pub fn pending_on_none<I>(future: impl Future<Output = Option<I>>) -> impl Futur
         }
     })
 }
+
+pub fn drop_either_future<A, B>(
+    either: Either<(A, impl Future), (B, impl Future)>,
+) -> Either<A, B> {
+    match either {
+        Either::Left((left, _)) => Either::Left(left),
+        Either::Right((right, _)) => Either::Right(right),
+    }
+}

src/common/src/util/mod.rs

@@ -45,7 +45,7 @@ pub mod tracing;
 pub mod value_encoding;
 pub mod worker_util;
 
-pub use future_utils::{pending_on_none, select_all};
+pub use future_utils::{drop_either_future, pending_on_none, select_all};
 #[macro_use]
 pub mod match_util;