[rust] Tighten down internal API visibility to pub(crate) (#377)

crates/fluss/src/client/admin.rs

@@ -56,7 +56,7 @@ pub struct FlussAdmin {
 }
 
 impl FlussAdmin {
-    pub fn new(connections: Arc<RpcClient>, metadata: Arc<Metadata>) -> Self {
+    pub(crate) fn new(connections: Arc<RpcClient>, metadata: Arc<Metadata>) -> Self {
         FlussAdmin {
             metadata,
             rpc_client: connections,

crates/fluss/src/client/connection.rs

@@ -86,11 +86,11 @@ impl FlussConnection {
         Ok(())
     }
 
-    pub fn get_metadata(&self) -> Arc<Metadata> {
+    pub(crate) fn get_metadata(&self) -> Arc<Metadata> {
         self.metadata.clone()
     }
 
-    pub fn get_connections(&self) -> Arc<RpcClient> {
+    pub(crate) fn get_connections(&self) -> Arc<RpcClient> {
         self.network_connects.clone()
     }
 
@@ -121,7 +121,7 @@ impl FlussConnection {
         Ok(admin)
     }
 
-    pub fn get_or_create_writer_client(&self) -> Result<Arc<WriterClient>> {
+    pub(crate) fn get_or_create_writer_client(&self) -> Result<Arc<WriterClient>> {
         // 1. Fast path: Attempt to acquire a read lock to check if the client already exists.
         if let Some(client) = self.writer_client.read().as_ref() {
             return Ok(client.clone());
@@ -145,7 +145,7 @@ impl FlussConnection {
     }
 
     /// Gets or creates a lookup client for batched lookup operations.
-    pub fn get_or_create_lookup_client(&self) -> Result<Arc<LookupClient>> {
+    pub(crate) fn get_or_create_lookup_client(&self) -> Result<Arc<LookupClient>> {
         // 1. Fast path: Attempt to acquire a read lock to check if the client already exists.
         if let Some(client) = self.lookup_client.read().as_ref() {
             return Ok(client.clone());

crates/fluss/src/client/credentials.rs

@@ -42,7 +42,7 @@ const DEFAULT_NON_EXPIRING_REFRESH_INTERVAL: Duration = Duration::from_secs(7 *
 /// Type alias for credentials properties receiver
 /// - `None` = not yet fetched, should wait
 /// - `Some(HashMap)` = fetched (may be empty if no auth needed)
-pub type CredentialsReceiver = watch::Receiver<Option<HashMap<String, String>>>;
+pub(crate) type CredentialsReceiver = watch::Receiver<Option<HashMap<String, String>>>;
 
 #[derive(Debug, Deserialize)]
 struct Credentials {
@@ -136,7 +136,7 @@ fn build_remote_fs_props(
 /// // Consumer can get latest credentials via:
 /// let props = credentials_rx.borrow().clone();
 /// ```
-pub struct SecurityTokenManager {
+pub(crate) struct SecurityTokenManager {
     rpc_client: Arc<RpcClient>,
     metadata: Arc<Metadata>,
     token_renewal_ratio: f64,
@@ -152,7 +152,7 @@ pub struct SecurityTokenManager {
 }
 
 impl SecurityTokenManager {
-    pub fn new(rpc_client: Arc<RpcClient>, metadata: Arc<Metadata>) -> Self {
+    pub(crate) fn new(rpc_client: Arc<RpcClient>, metadata: Arc<Metadata>) -> Self {
         let (credentials_tx, credentials_rx) = watch::channel(None);
         Self {
             rpc_client,
@@ -169,13 +169,13 @@ impl SecurityTokenManager {
     /// Subscribe to credential updates.
     /// Returns a receiver that always contains the latest credentials.
     /// Consumers can call `receiver.borrow()` to get the current value.
-    pub fn subscribe(&self) -> CredentialsReceiver {
+    pub(crate) fn subscribe(&self) -> CredentialsReceiver {
         self.credentials_rx.clone()
     }
 
     /// Start the background token refresh task.
     /// This should be called once after creating the manager.
-    pub fn start(&self) {
+    pub(crate) fn start(&self) {
         if self.task_handle.read().is_some() {
             warn!("SecurityTokenManager is already started");
             return;
@@ -207,7 +207,7 @@ impl SecurityTokenManager {
     }
 
     /// Stop the background token refresh task.
-    pub fn stop(&self) {
+    pub(crate) fn stop(&self) {
         if let Some(tx) = self.shutdown_tx.write().take() {
             let _ = tx.send(());
         }

crates/fluss/src/client/lookup/lookup_client.rs

@@ -48,7 +48,7 @@ use tokio::task::JoinHandle;
 /// let lookup_client = LookupClient::new(config, metadata);
 /// let result = lookup_client.lookup(table_path, table_bucket, key_bytes).await?;
 /// ```
-pub struct LookupClient {
+pub(crate) struct LookupClient {
     /// Channel to send lookup requests to the queue
     lookup_tx: mpsc::Sender<QueuedLookup>,
     /// Handle to the sender task
@@ -61,7 +61,7 @@ pub struct LookupClient {
 
 impl LookupClient {
     /// Creates a new lookup client.
-    pub fn new(config: &Config, metadata: Arc<Metadata>) -> Self {
+    pub(crate) fn new(config: &Config, metadata: Arc<Metadata>) -> Self {
         // Extract configuration values
         let queue_size = config.lookup_queue_size;
         let max_batch_size = config.lookup_max_batch_size;
@@ -116,7 +116,7 @@ impl LookupClient {
     /// * `Ok(Some(bytes))` - The value bytes if found
     /// * `Ok(None)` - If the key was not found
     /// * `Err(Error)` - If the lookup fails
-    pub async fn lookup(
+    pub(crate) async fn lookup(
         &self,
         table_path: TablePath,
         table_bucket: TableBucket,
@@ -160,7 +160,7 @@ impl LookupClient {
     /// # Returns
     /// * `Ok(rows)` - Every row matching the prefix (possibly empty)
     /// * `Err(Error)` - If the lookup fails
-    pub async fn prefix_lookup(
+    pub(crate) async fn prefix_lookup(
         &self,
         table_path: TablePath,
         table_bucket: TableBucket,

crates/fluss/src/client/lookup/mod.rs

@@ -34,6 +34,6 @@ mod lookup_query;
 mod lookup_queue;
 mod lookup_sender;
 
-pub use lookup_client::LookupClient;
+pub(crate) use lookup_client::LookupClient;
 pub(crate) use lookup_query::{PrefixLookupQuery, PrimaryLookupQuery, QueuedLookup};
 pub(crate) use lookup_queue::LookupQueue;

crates/fluss/src/client/metadata.rs

@@ -37,7 +37,7 @@ pub struct Metadata {
 }
 
 impl Metadata {
-    pub async fn new(bootstrap: &str, connections: Arc<RpcClient>) -> Result<Self> {
+    pub(crate) async fn new(bootstrap: &str, connections: Arc<RpcClient>) -> Result<Self> {
         let cluster = Self::init_cluster(bootstrap, connections.clone()).await?;
         let (cluster_version_tx, _) = watch::channel(0);
         Ok(Metadata {
@@ -48,7 +48,7 @@ impl Metadata {
         })
     }
 
-    pub fn subscribe_cluster_changes(&self) -> watch::Receiver<u64> {
+    pub(crate) fn subscribe_cluster_changes(&self) -> watch::Receiver<u64> {
         self.cluster_version_tx.subscribe()
     }
 
@@ -109,7 +109,7 @@ impl Metadata {
         Ok(())
     }
 
-    pub fn invalidate_server(&self, server_id: &i32, table_ids: Vec<i64>) {
+    pub(crate) fn invalidate_server(&self, server_id: &i32, table_ids: Vec<i64>) {
         {
             let mut cluster_guard = self.cluster.write();
             let updated_cluster = cluster_guard.invalidate_server(server_id, table_ids);
@@ -118,7 +118,7 @@ impl Metadata {
         self.notify_cluster_changed();
     }
 
-    pub fn invalidate_physical_table_meta(
+    pub(crate) fn invalidate_physical_table_meta(
         &self,
         physical_tables_to_invalid: &HashSet<PhysicalTablePath>,
     ) {
@@ -131,7 +131,7 @@ impl Metadata {
         self.notify_cluster_changed();
     }
 
-    pub async fn update(&self, metadata_response: MetadataResponse) -> Result<()> {
+    pub(crate) async fn update(&self, metadata_response: MetadataResponse) -> Result<()> {
         let origin_cluster = self.cluster.read().clone();
         let new_cluster =
             Cluster::from_metadata_response(metadata_response, Some(&origin_cluster))?;
@@ -143,7 +143,7 @@ impl Metadata {
         Ok(())
     }
 
-    pub async fn update_tables_metadata(
+    pub(crate) async fn update_tables_metadata(
         &self,
         table_paths: &HashSet<&TablePath>,
         physical_table_paths: &HashSet<&Arc<PhysicalTablePath>>,
@@ -178,12 +178,12 @@ impl Metadata {
         Ok(())
     }
 
-    pub async fn update_table_metadata(&self, table_path: &TablePath) -> Result<()> {
+    pub(crate) async fn update_table_metadata(&self, table_path: &TablePath) -> Result<()> {
         self.update_tables_metadata(&HashSet::from([table_path]), &HashSet::new(), vec![])
             .await
     }
 
-    pub async fn update_physical_table_metadata(
+    pub(crate) async fn update_physical_table_metadata(
         &self,
         physical_table_paths: &[Arc<PhysicalTablePath>],
     ) -> Result<()> {
@@ -204,7 +204,7 @@ impl Metadata {
             .await
     }
 
-    pub async fn check_and_update_table_metadata(&self, table_paths: &[TablePath]) -> Result<()> {
+    pub(crate) async fn check_and_update_table_metadata(&self, table_paths: &[TablePath]) -> Result<()> {
         let cluster_binding = self.cluster.read().clone();
         let need_update_table_paths: HashSet<&TablePath> = table_paths
             .iter()
@@ -221,7 +221,7 @@ impl Metadata {
     /// Resolves the partition id, refreshing metadata once if not cached.
     /// Returns `None` when the partition does not exist — `PartitionNotExists`
     /// server errors are swallowed so callers can short-circuit to an empty result.
-    pub async fn check_and_update_partition_metadata(
+    pub(crate) async fn check_and_update_partition_metadata(
         &self,
         physical_table_path: &PhysicalTablePath,
     ) -> Result<Option<PartitionId>> {
@@ -247,14 +247,14 @@ impl Metadata {
         Ok(result)
     }
 
-    pub fn get_cluster(&self) -> Arc<Cluster> {
+    pub(crate) fn get_cluster(&self) -> Arc<Cluster> {
         let guard = self.cluster.read();
         guard.clone()
     }
 
     const MAX_RETRY_TIMES: u8 = 3;
 
-    pub async fn leader_for(
+    pub(crate) async fn leader_for(
         &self,
         table_path: &TablePath,
         table_bucket: &TableBucket,

crates/fluss/src/client/mod.rs

@@ -18,17 +18,19 @@
 mod admin;
 mod connection;
 mod credentials;
-pub mod lookup;
+mod lookup;
 mod metadata;
 mod schema_getter;
 mod table;
 mod write;
 
 pub use admin::*;
 pub use connection::*;
-pub use credentials::*;
-pub use lookup::LookupClient;
+pub(crate) use credentials::*;
+pub(crate) use lookup::LookupClient;
+pub use crate::rpc::message::OffsetSpec;
 pub use metadata::*;
 pub(crate) use schema_getter::ClientSchemaGetter;
 pub use table::*;
-pub use write::*;
+pub use write::WriteResultFuture;
+pub(crate) use write::*;

crates/fluss/src/client/schema_getter.rs

@@ -35,7 +35,7 @@ pub(crate) struct ClientSchemaGetter {
 }
 
 impl ClientSchemaGetter {
-    pub fn new(table_path: TablePath, admin: Arc<FlussAdmin>, latest: SchemaInfo) -> Self {
+    pub(crate) fn new(table_path: TablePath, admin: Arc<FlussAdmin>, latest: SchemaInfo) -> Self {
         let mut map = HashMap::new();
         let (schema, schema_id) = latest.into_parts();
         map.insert(schema_id, Arc::new(schema));
@@ -50,7 +50,7 @@ impl ClientSchemaGetter {
     /// accept one redundant RPC in exchange for staying off
     /// `tokio::sync` machinery. Schemas are immutable per id, so
     /// last-write-wins on the cache insert is correct.
-    pub async fn get_schema(&self, schema_id: i32) -> Result<Arc<Schema>> {
+    pub(crate) async fn get_schema(&self, schema_id: i32) -> Result<Arc<Schema>> {
         if let Some(schema) = self.cache.read().get(&schema_id).cloned() {
             return Ok(schema);
         }

crates/fluss/src/client/table/log_fetch_buffer.rs

@@ -60,7 +60,7 @@ pub(crate) struct FetchErrorContext {
 }
 
 /// Represents a completed fetch that can be consumed
-pub trait CompletedFetch: Send + Sync {
+pub(crate) trait CompletedFetch: Send + Sync {
     fn table_bucket(&self) -> &TableBucket;
     fn api_error(&self) -> Option<&ApiError>;
     fn fetch_error_context(&self) -> Option<&FetchErrorContext>;
@@ -101,14 +101,14 @@ enum FetchStep<T> {
 }
 
 /// Represents a pending fetch that is waiting to be completed
-pub trait PendingFetch: Send + Sync {
+pub(crate) trait PendingFetch: Send + Sync {
     fn table_bucket(&self) -> &TableBucket;
     fn is_completed(&self) -> bool;
     fn to_completed_fetch(self: Box<Self>) -> Result<Box<dyn CompletedFetch>>;
 }
 
 /// Thread-safe buffer for completed fetches
-pub struct LogFetchBuffer {
+pub(crate) struct LogFetchBuffer {
     resolver: Arc<ReadContextResolver>,
     completed_fetches: Mutex<VecDeque<Box<dyn CompletedFetch>>>,
     pending_fetches: Mutex<HashMap<TableBucket, VecDeque<Box<dyn PendingFetch>>>>,
@@ -357,7 +357,7 @@ impl PendingFetch for CompletedPendingFetch {
 
 /// Default implementation of CompletedFetch for in-memory log records
 /// Used for local fetches from tablet server
-pub struct DefaultCompletedFetch {
+pub(crate) struct DefaultCompletedFetch {
     table_bucket: TableBucket,
     api_error: Option<ApiError>,
     fetch_error_context: Option<FetchErrorContext>,
@@ -807,7 +807,7 @@ impl CompletedFetch for DefaultCompletedFetch {
 /// Completed fetch for remote log segments
 /// Matches Java's RemoteCompletedFetch design - separate class for remote vs local
 /// Holds RAII permit until consumed (data is in inner)
-pub struct RemoteCompletedFetch {
+pub(crate) struct RemoteCompletedFetch {
     inner: DefaultCompletedFetch,
     permit: Option<PrefetchPermit>,
 }
@@ -887,7 +887,7 @@ impl CompletedFetch for RemoteCompletedFetch {
 // Permit released explicitly in drain() or automatically when struct drops
 
 /// Pending fetch that waits for remote log file to be downloaded
-pub struct RemotePendingFetch {
+pub(crate) struct RemotePendingFetch {
     segment: RemoteLogSegment,
     download_future: RemoteLogDownloadFuture,
     pos_in_log_segment: i32,

crates/fluss/src/client/table/mod.rs

@@ -40,7 +40,7 @@ pub use append::{AppendWriter, TableAppend};
 pub use batch_scanner::LimitBatchScanner;
 pub use lookup::{LookupResult, Lookuper, PrefixKeyLookuper, TableLookup, TablePrefixLookup};
 pub use reader::{RecordBatchLogReader, SyncRecordBatchLogReader};
-pub use remote_log::{
+pub(crate) use remote_log::{
     DEFAULT_REMOTE_FILE_DOWNLOAD_THREAD_NUM, DEFAULT_SCANNER_REMOTE_LOG_PREFETCH_NUM,
 };
 pub use scanner::{LogScanner, RecordBatchLogScanner, TableScan};

crates/fluss/src/client/table/partition_getter.rs

@@ -24,7 +24,7 @@ use crate::util::partition;
 use std::sync::Arc;
 
 /// Get the physical table path for a row, handling partitioned vs non-partitioned tables.
-pub fn get_physical_path<R: InternalRow>(
+pub(crate) fn get_physical_path<R: InternalRow>(
     table_path: &Arc<TablePath>,
     partition_getter: Option<&PartitionGetter>,
     row: &R,
@@ -42,7 +42,7 @@ pub fn get_physical_path<R: InternalRow>(
 
 /// A getter to get partition name from a row.
 #[allow(dead_code)]
-pub struct PartitionGetter {
+pub(crate) struct PartitionGetter {
     partition_keys: Arc<[String]>,
     partitions: Vec<(DataType, FieldGetter)>,
 }