fix opaque type

Signed-off-by: xxchan <xxchan22f@gmail.com>

Author: xxchan

src/connector/src/sink/kafka.rs

@@ -403,7 +403,16 @@ struct KafkaPayloadWriter<'a> {
     config: &'a KafkaConfig,
 }
 
-pub type KafkaSinkDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
+mod opaque_type {
+    use super::*;
+    pub type KafkaSinkDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
+
+    pub(super) fn map_delivery_future(future: DeliveryFuture) -> KafkaSinkDeliveryFuture {
+        future.map(KafkaPayloadWriter::<'static>::map_future_result)
+    }
+}
+use opaque_type::map_delivery_future;
+pub use opaque_type::KafkaSinkDeliveryFuture;
 
 pub struct KafkaSinkWriter {
     formatter: SinkFormatterImpl,
@@ -482,7 +491,7 @@ impl<'w> KafkaPayloadWriter<'w> {
                 Ok(delivery_future) => {
                     if self
                         .add_future
-                        .add_future_may_await(Self::map_delivery_future(delivery_future))
+                        .add_future_may_await(map_delivery_future(delivery_future))
                         .await?
                     {
                         tracing::warn!(
@@ -567,10 +576,6 @@ impl<'w> KafkaPayloadWriter<'w> {
             Err(_) => Err(KafkaError::Canceled.into()),
         }
     }
-
-    fn map_delivery_future(future: DeliveryFuture) -> KafkaSinkDeliveryFuture {
-        future.map(KafkaPayloadWriter::<'static>::map_future_result)
-    }
 }
 
 impl<'a> FormattedSink for KafkaPayloadWriter<'a> {

src/connector/src/sink/kinesis.rs

@@ -201,8 +201,36 @@ impl KinesisSinkWriter {
     }
 }
 
-pub type KinesisSinkPayloadWriterDeliveryFuture =
-    impl TryFuture<Ok = (), Error = SinkError> + Unpin + Send + 'static;
+mod opaque_type {
+    use super::*;
+    pub type KinesisSinkPayloadWriterDeliveryFuture =
+        impl TryFuture<Ok = (), Error = SinkError> + Unpin + Send + 'static;
+
+    impl KinesisSinkPayloadWriter {
+        pub(super) fn finish(self) -> KinesisSinkPayloadWriterDeliveryFuture {
+            async move {
+                let builder = self.builder.expect("should not be None");
+                let context_fmt = format!(
+                    "failed to put record to {}",
+                    builder
+                        .get_stream_name()
+                        .as_ref()
+                        .expect("should have set stream name")
+                );
+                Retry::spawn(
+                    ExponentialBackoff::from_millis(100).map(jitter).take(3),
+                    || builder.clone().send(),
+                )
+                .await
+                .with_context(|| context_fmt.clone())
+                .map_err(SinkError::Kinesis)?;
+                Ok(())
+            }
+            .boxed()
+        }
+    }
+}
+pub use opaque_type::KinesisSinkPayloadWriterDeliveryFuture;
 
 impl KinesisSinkPayloadWriter {
     fn put_record(&mut self, key: String, payload: Vec<u8>) {
@@ -216,28 +244,6 @@ impl KinesisSinkPayloadWriter {
             ),
         );
     }
-
-    fn finish(self) -> KinesisSinkPayloadWriterDeliveryFuture {
-        async move {
-            let builder = self.builder.expect("should not be None");
-            let context_fmt = format!(
-                "failed to put record to {}",
-                builder
-                    .get_stream_name()
-                    .as_ref()
-                    .expect("should have set stream name")
-            );
-            Retry::spawn(
-                ExponentialBackoff::from_millis(100).map(jitter).take(3),
-                || builder.clone().send(),
-            )
-            .await
-            .with_context(|| context_fmt.clone())
-            .map_err(SinkError::Kinesis)?;
-            Ok(())
-        }
-        .boxed()
-    }
 }
 
 impl FormattedSink for KinesisSinkPayloadWriter {

src/connector/src/sink/pulsar.rs

@@ -235,15 +235,20 @@ struct PulsarPayloadWriter<'w> {
     add_future: DeliveryFutureManagerAddFuture<'w, PulsarDeliveryFuture>,
 }
 
-pub type PulsarDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
-
-fn may_delivery_future(future: SendFuture) -> PulsarDeliveryFuture {
-    future.map(|result| {
-        result
-            .map(|_| ())
-            .map_err(|e: pulsar::Error| SinkError::Pulsar(anyhow!(e)))
-    })
+mod opaque_type {
+    use super::*;
+    pub type PulsarDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;
+
+    pub(super) fn may_delivery_future(future: SendFuture) -> PulsarDeliveryFuture {
+        future.map(|result| {
+            result
+                .map(|_| ())
+                .map_err(|e: pulsar::Error| SinkError::Pulsar(anyhow!(e)))
+        })
+    }
 }
+use opaque_type::may_delivery_future;
+pub use opaque_type::PulsarDeliveryFuture;
 
 impl PulsarSinkWriter {
     pub async fn new(

src/jni_core/src/lib.rs

@@ -242,25 +242,28 @@ struct JavaClassMethodCache {
     utc: OnceLock<GlobalRef>,
 }
 
-// TODO: may only return a RowRef
-pub type StreamChunkRowIterator<'a> = impl Iterator<Item = (Op, OwnedRow)> + 'a;
+mod opaque_type {
+    use super::*;
+    // TODO: may only return a RowRef
+    pub type StreamChunkRowIterator<'a> = impl Iterator<Item = (Op, OwnedRow)> + 'a;
+
+    impl<'a> JavaBindingIteratorInner<'a> {
+        pub(super) fn from_chunk(chunk: &'a StreamChunk) -> JavaBindingIteratorInner<'a> {
+            JavaBindingIteratorInner::StreamChunk(
+                chunk
+                    .rows()
+                    .map(|(op, row)| (op.to_protobuf(), row.to_owned_row())),
+            )
+        }
+    }
+}
+pub use opaque_type::StreamChunkRowIterator;
 pub type HummockJavaBindingIterator = BoxStream<'static, anyhow::Result<(Bytes, OwnedRow)>>;
-
 pub enum JavaBindingIteratorInner<'a> {
     Hummock(HummockJavaBindingIterator),
     StreamChunk(StreamChunkRowIterator<'a>),
 }
 
-impl<'a> JavaBindingIteratorInner<'a> {
-    fn from_chunk(chunk: &'a StreamChunk) -> JavaBindingIteratorInner<'a> {
-        JavaBindingIteratorInner::StreamChunk(
-            chunk
-                .rows()
-                .map(|(op, row)| (op.to_protobuf(), row.to_owned_row())),
-        )
-    }
-}
-
 enum RowExtra {
     Op(Op),
     Key(Bytes),

src/storage/src/store_impl.rs

@@ -42,9 +42,27 @@ use crate::monitor::{
 use crate::opts::StorageOpts;
 use crate::StateStore;
 
-pub type HummockStorageType = impl StateStore + AsHummock;
-pub type MemoryStateStoreType = impl StateStore + AsHummock;
-pub type SledStateStoreType = impl StateStore + AsHummock;
+mod opaque_type {
+    use super::*;
+
+    pub type HummockStorageType = impl StateStore + AsHummock;
+    pub type MemoryStateStoreType = impl StateStore + AsHummock;
+    pub type SledStateStoreType = impl StateStore + AsHummock;
+
+    pub fn in_memory(state_store: MemoryStateStore) -> MemoryStateStoreType {
+        may_dynamic_dispatch(state_store)
+    }
+
+    pub fn hummock(state_store: HummockStorage) -> HummockStorageType {
+        may_dynamic_dispatch(may_verify(state_store))
+    }
+
+    pub fn sled(state_store: SledStateStore) -> SledStateStoreType {
+        may_dynamic_dispatch(state_store)
+    }
+}
+use opaque_type::{hummock, in_memory, sled};
+pub use opaque_type::{HummockStorageType, MemoryStateStoreType, SledStateStoreType};
 
 /// The type erased [`StateStore`].
 #[derive(Clone, EnumAsInner)]
@@ -114,24 +132,22 @@ impl StateStoreImpl {
         storage_metrics: Arc<MonitoredStorageMetrics>,
     ) -> Self {
         // The specific type of MemoryStateStoreType in deducted here.
-        Self::MemoryStateStore(may_dynamic_dispatch(state_store).monitored(storage_metrics))
+        Self::MemoryStateStore(in_memory(state_store).monitored(storage_metrics))
     }
 
     pub fn hummock(
         state_store: HummockStorage,
         storage_metrics: Arc<MonitoredStorageMetrics>,
     ) -> Self {
         // The specific type of HummockStateStoreType in deducted here.
-        Self::HummockStateStore(
-            may_dynamic_dispatch(may_verify(state_store)).monitored(storage_metrics),
-        )
+        Self::HummockStateStore(hummock(state_store).monitored(storage_metrics))
     }
 
     pub fn sled(
         state_store: SledStateStore,
         storage_metrics: Arc<MonitoredStorageMetrics>,
     ) -> Self {
-        Self::SledStateStore(may_dynamic_dispatch(state_store).monitored(storage_metrics))
+        Self::SledStateStore(sled(state_store).monitored(storage_metrics))
     }
 
     pub fn shared_in_memory_store(storage_metrics: Arc<MonitoredStorageMetrics>) -> Self {