diff --git a/crates/fluss/src/client/admin.rs b/crates/fluss/src/client/admin.rs
index 7015e573..7d526490 100644
--- a/crates/fluss/src/client/admin.rs
+++ b/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,
diff --git a/crates/fluss/src/client/connection.rs b/crates/fluss/src/client/connection.rs
index c31104c4..bbe44217 100644
--- a/crates/fluss/src/client/connection.rs
+++ b/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());
diff --git a/crates/fluss/src/client/credentials.rs b/crates/fluss/src/client/credentials.rs
index a954e2a9..d7a2c5f8 100644
--- a/crates/fluss/src/client/credentials.rs
+++ b/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(());
         }
diff --git a/crates/fluss/src/client/lookup/lookup_client.rs b/crates/fluss/src/client/lookup/lookup_client.rs
index 4d507aa9..243ba844 100644
--- a/crates/fluss/src/client/lookup/lookup_client.rs
+++ b/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,
diff --git a/crates/fluss/src/client/lookup/mod.rs b/crates/fluss/src/client/lookup/mod.rs
index ac2446a9..4f816ce0 100644
--- a/crates/fluss/src/client/lookup/mod.rs
+++ b/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;
diff --git a/crates/fluss/src/client/metadata.rs b/crates/fluss/src/client/metadata.rs
index 20a8a159..8f9e9dcc 100644
--- a/crates/fluss/src/client/metadata.rs
+++ b/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,
diff --git a/crates/fluss/src/client/mod.rs b/crates/fluss/src/client/mod.rs
index f8027948..fed0b99a 100644
--- a/crates/fluss/src/client/mod.rs
+++ b/crates/fluss/src/client/mod.rs
@@ -18,7 +18,7 @@
 mod admin;
 mod connection;
 mod credentials;
-pub mod lookup;
+mod lookup;
 mod metadata;
 mod schema_getter;
 mod table;
@@ -26,9 +26,11 @@ 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::*;
diff --git a/crates/fluss/src/client/schema_getter.rs b/crates/fluss/src/client/schema_getter.rs
index 4b643c0b..b0d6060e 100644
--- a/crates/fluss/src/client/schema_getter.rs
+++ b/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);
         }
diff --git a/crates/fluss/src/client/table/log_fetch_buffer.rs b/crates/fluss/src/client/table/log_fetch_buffer.rs
index 794fe9df..c991de5e 100644
--- a/crates/fluss/src/client/table/log_fetch_buffer.rs
+++ b/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,
diff --git a/crates/fluss/src/client/table/mod.rs b/crates/fluss/src/client/table/mod.rs
index 41cbadab..7ec65427 100644
--- a/crates/fluss/src/client/table/mod.rs
+++ b/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};
diff --git a/crates/fluss/src/client/table/partition_getter.rs b/crates/fluss/src/client/table/partition_getter.rs
index 1115ded3..e705f746 100644
--- a/crates/fluss/src/client/table/partition_getter.rs
+++ b/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)>,
 }
diff --git a/crates/fluss/src/client/table/read_context_resolver.rs b/crates/fluss/src/client/table/read_context_resolver.rs
index 655891f4..cd6fbf33 100644
--- a/crates/fluss/src/client/table/read_context_resolver.rs
+++ b/crates/fluss/src/client/table/read_context_resolver.rs
@@ -62,7 +62,7 @@ struct ResolvedContexts {
 
 impl ReadContextResolver {
     /// Create a new resolver with the initial schema's ReadContexts.
-    pub fn new(
+    pub(crate) fn new(
         initial_schema_id: i16,
         local_context: Arc<ReadContext>,
         remote_context: Arc<ReadContext>,
@@ -87,12 +87,12 @@ impl ReadContextResolver {
         }
     }
 
-    pub fn with_schema_getter(mut self, schema_getter: Arc<ClientSchemaGetter>) -> Self {
+    pub(crate) fn with_schema_getter(mut self, schema_getter: Arc<ClientSchemaGetter>) -> Self {
         self.schema_getter = Some(schema_getter);
         self
     }
 
-    pub fn with_fixed_schema(mut self, fixed_schema: bool) -> Self {
+    pub(crate) fn with_fixed_schema(mut self, fixed_schema: bool) -> Self {
         self.fixed_schema = fixed_schema;
         if fixed_schema {
             let fixed_target = {
@@ -115,7 +115,7 @@ impl ReadContextResolver {
     /// Resolve the ReadContext for the given schema_id.
     /// Returns the initial context if projection is active (schema pinned).
     /// Returns None if the schema_id is not yet cached.
-    pub fn resolve(&self, schema_id: i16, is_remote: bool) -> Option<Arc<ReadContext>> {
+    pub(crate) fn resolve(&self, schema_id: i16, is_remote: bool) -> Option<Arc<ReadContext>> {
         // If projection is active, always return the initial context
         let effective_id = if self.projected_fields.is_some() {
             self.initial_schema_id
@@ -138,7 +138,7 @@ impl ReadContextResolver {
     /// Callers should invoke this only after encountering a concrete batch
     /// whose schema is absent. This keeps remote log files streaming: no scan
     /// of the remaining segment is needed to discover all schema IDs up front.
-    pub async fn fetch_and_register(&self, schema_id: i16) -> Result<()> {
+    pub(crate) async fn fetch_and_register(&self, schema_id: i16) -> Result<()> {
         if self.projected_fields.is_some() || self.contexts.read().contains_key(&schema_id) {
             return Ok(());
         }
@@ -158,7 +158,7 @@ impl ReadContextResolver {
 
     /// Register a new schema by its ID. Builds ReadContexts from the Schema.
     /// No-op if already cached or if projection is active.
-    pub fn register_schema(&self, schema_id: i16, schema: &Schema) -> Result<()> {
+    pub(crate) fn register_schema(&self, schema_id: i16, schema: &Schema) -> Result<()> {
         if self.projected_fields.is_some() {
             // Projection pins the schema, no need to register new ones
             return Ok(());
@@ -203,13 +203,13 @@ impl ReadContextResolver {
     }
 
     /// Returns the initial schema ID.
-    pub fn initial_schema_id(&self) -> i16 {
+    pub(crate) fn initial_schema_id(&self) -> i16 {
         self.initial_schema_id
     }
 
     /// Returns the projection fields if active, used for fetch request pushdown.
     #[allow(dead_code)]
-    pub fn projected_fields(&self) -> Option<&[usize]> {
+    pub(crate) fn projected_fields(&self) -> Option<&[usize]> {
         self.projected_fields.as_deref()
     }
 }
diff --git a/crates/fluss/src/client/table/remote_log.rs b/crates/fluss/src/client/table/remote_log.rs
index c843d1a2..f9ad9680 100644
--- a/crates/fluss/src/client/table/remote_log.rs
+++ b/crates/fluss/src/client/table/remote_log.rs
@@ -45,11 +45,11 @@ use tokio::task::JoinSet;
 
 /// Default maximum number of remote log segments to prefetch
 /// Matches Java's CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM (default: 4)
-pub const DEFAULT_SCANNER_REMOTE_LOG_PREFETCH_NUM: usize = 4;
+pub(crate) const DEFAULT_SCANNER_REMOTE_LOG_PREFETCH_NUM: usize = 4;
 
 /// Default maximum concurrent remote log downloads
 /// Matches Java's REMOTE_FILE_DOWNLOAD_THREAD_NUM (default: 3)
-pub const DEFAULT_REMOTE_FILE_DOWNLOAD_THREAD_NUM: usize = 3;
+pub(crate) const DEFAULT_REMOTE_FILE_DOWNLOAD_THREAD_NUM: usize = 3;
 
 /// Initial retry backoff delay (milliseconds)
 /// Prevents hot-spin retry loops on persistent failures
@@ -83,13 +83,13 @@ fn calculate_backoff_delay(retry_count: u32) -> tokio::time::Duration {
 
 /// Result of a fetch operation containing file path and size
 #[derive(Debug)]
-pub struct FetchResult {
+pub(crate) struct FetchResult {
     pub file_path: PathBuf,
     pub file_size: usize,
 }
 
 /// Trait for fetching remote log segments (allows dependency injection for testing)
-pub trait RemoteLogFetcher: Send + Sync {
+pub(crate) trait RemoteLogFetcher: Send + Sync {
     fn fetch(
         &self,
         request: &RemoteLogDownloadRequest,
@@ -98,7 +98,7 @@ pub trait RemoteLogFetcher: Send + Sync {
 
 /// Represents a remote log segment that needs to be downloaded
 #[derive(Debug, Clone)]
-pub struct RemoteLogSegment {
+pub(crate) struct RemoteLogSegment {
     pub segment_id: String,
     pub start_offset: i64,
     #[allow(dead_code)]
@@ -133,7 +133,7 @@ impl RemoteLogSegment {
 
 /// Represents remote log fetch information
 #[derive(Debug, Clone)]
-pub struct RemoteLogFetchInfo {
+pub(crate) struct RemoteLogFetchInfo {
     pub remote_log_tablet_dir: String,
     #[allow(dead_code)]
     pub partition_name: Option<String>,
@@ -163,7 +163,7 @@ impl RemoteLogFetchInfo {
 /// NOTE: File deletion is now handled by FileSource::drop(), not here.
 /// This ensures the file is closed before deletion
 #[derive(Debug)]
-pub struct PrefetchPermit {
+pub(crate) struct PrefetchPermit {
     permit: Option<OwnedSemaphorePermit>,
     recycle_notify: Arc<Notify>,
 }
@@ -190,7 +190,7 @@ impl Drop for PrefetchPermit {
 /// Downloaded remote log file with prefetch permit
 /// File remains on disk for memory efficiency; file deletion is handled by FileCleanupGuard in FileSource
 #[derive(Debug)]
-pub struct RemoteLogFile {
+pub(crate) struct RemoteLogFile {
     /// Path to the downloaded file on local disk
     pub file_path: PathBuf,
     /// Size of the file in bytes
@@ -203,7 +203,7 @@ pub struct RemoteLogFile {
 
 /// Represents a request to download a remote log segment with priority ordering
 #[derive(Debug)]
-pub struct RemoteLogDownloadRequest {
+pub(crate) struct RemoteLogDownloadRequest {
     segment: RemoteLogSegment,
     remote_log_tablet_dir: String,
     result_sender: oneshot::Sender<Result<RemoteLogFile>>,
@@ -669,7 +669,7 @@ async fn coordinator_loop(
 type CompletionCallback = Box<dyn Fn() + Send + Sync>;
 
 /// Future for a remote log download request
-pub struct RemoteLogDownloadFuture {
+pub(crate) struct RemoteLogDownloadFuture {
     result: Arc<Mutex<Option<Result<RemoteLogFile>>>>,
     completion_callbacks: Arc<Mutex<Vec<CompletionCallback>>>,
 }
@@ -778,7 +778,7 @@ impl RemoteLogDownloadFuture {
 /// When the downloader is dropped, the request channel closes, signaling the coordinator
 /// to stop accepting new work. The coordinator will finish any in-flight downloads but
 /// won't wait for completion. Pending futures will fail.
-pub struct RemoteLogDownloader {
+pub(crate) struct RemoteLogDownloader {
     request_sender: Option<mpsc::UnboundedSender<RemoteLogDownloadRequest>>,
 }
 
diff --git a/crates/fluss/src/client/table/scanner.rs b/crates/fluss/src/client/table/scanner.rs
index aadd1cb8..40f2a584 100644
--- a/crates/fluss/src/client/table/scanner.rs
+++ b/crates/fluss/src/client/table/scanner.rs
@@ -72,7 +72,7 @@ pub struct TableScan<'a> {
 }
 
 impl<'a> TableScan<'a> {
-    pub fn new(conn: &'a FlussConnection, table_info: TableInfo, metadata: Arc<Metadata>) -> Self {
+    pub(crate) fn new(conn: &'a FlussConnection, table_info: TableInfo, metadata: Arc<Metadata>) -> Self {
         Self {
             conn,
             table_info,
@@ -1192,7 +1192,7 @@ struct FetchResponseContext {
 
 impl LogFetcher {
     #[allow(clippy::too_many_arguments)]
-    pub fn new(
+    pub(crate) fn new(
         table_info: TableInfo,
         conns: Arc<RpcClient>,
         metadata: Arc<Metadata>,
@@ -2206,7 +2206,7 @@ impl LogFetcher {
     }
 }
 
-pub struct LogScannerStatus {
+pub(crate) struct LogScannerStatus {
     bucket_status_map: Arc<RwLock<FairBucketStatusMap<BucketScanStatus>>>,
 }
 
@@ -2311,7 +2311,7 @@ impl Default for LogScannerStatus {
 
 #[derive(Debug)]
 #[allow(dead_code)]
-pub struct BucketScanStatus {
+pub(crate) struct BucketScanStatus {
     offset: RwLock<i64>,
     high_watermark: RwLock<i64>,
 }
diff --git a/crates/fluss/src/client/table/upsert.rs b/crates/fluss/src/client/table/upsert.rs
index 8a5547a7..a16b7615 100644
--- a/crates/fluss/src/client/table/upsert.rs
+++ b/crates/fluss/src/client/table/upsert.rs
@@ -38,7 +38,7 @@ pub struct TableUpsert {
 
 #[allow(dead_code)]
 impl TableUpsert {
-    pub fn new(
+    pub(crate) fn new(
         table_path: TablePath,
         table_info: TableInfo,
         writer_client: Arc<WriterClient>,
@@ -124,7 +124,7 @@ pub struct UpsertWriter {
 struct UpsertWriterFactory;
 
 impl UpsertWriterFactory {
-    pub fn create(
+    pub(crate) fn create(
         table_path: Arc<TablePath>,
         table_info: Arc<TableInfo>,
         partial_update_columns: Option<Arc<Vec<usize>>>,
@@ -560,10 +560,10 @@ mod tests {
 /// Currently this is an empty struct to allow for compatible evolution in the future
 #[derive(Default)]
 #[allow(dead_code)]
-pub struct UpsertResult;
+pub(crate) struct UpsertResult;
 
 /// The result of deleting a record
 /// Currently this is an empty struct to allow for compatible evolution in the future
 #[derive(Default)]
 #[allow(dead_code)]
-pub struct DeleteResult;
+pub(crate) struct DeleteResult;
diff --git a/crates/fluss/src/client/write/accumulator.rs b/crates/fluss/src/client/write/accumulator.rs
index ed5d31a3..7f984e86 100644
--- a/crates/fluss/src/client/write/accumulator.rs
+++ b/crates/fluss/src/client/write/accumulator.rs
@@ -181,7 +181,7 @@ impl Drop for MemoryPermit {
 type BucketBatches = Vec<(BucketId, Arc<Mutex<VecDeque<WriteBatch>>>)>;
 
 #[allow(dead_code)]
-pub struct RecordAccumulator {
+pub(crate) struct RecordAccumulator {
     config: Config,
     write_batches: DashMap<Arc<PhysicalTablePath>, BucketAndWriteBatches>,
     // batch_id -> (complete callback, memory permit)
@@ -980,7 +980,7 @@ impl RecordAccumulator {
     }
 }
 
-pub struct ReadyWriteBatch {
+pub(crate) struct ReadyWriteBatch {
     pub table_bucket: TableBucket,
     pub write_batch: WriteBatch,
 }
@@ -1027,7 +1027,7 @@ impl BucketAndWriteBatches {
     }
 }
 
-pub struct RecordAppendResult {
+pub(crate) struct RecordAppendResult {
     pub batch_is_full: bool,
     pub new_batch_created: bool,
     pub abort_record_for_new_batch: bool,
@@ -1063,7 +1063,7 @@ impl RecordAppendResult {
     }
 }
 
-pub struct ReadyCheckResult {
+pub(crate) struct ReadyCheckResult {
     pub ready_nodes: HashSet<ServerNode>,
     pub next_ready_check_delay_ms: i64,
     pub unknown_leader_tables: HashSet<Arc<PhysicalTablePath>>,
diff --git a/crates/fluss/src/client/write/batch.rs b/crates/fluss/src/client/write/batch.rs
index 50148835..269ab162 100644
--- a/crates/fluss/src/client/write/batch.rs
+++ b/crates/fluss/src/client/write/batch.rs
@@ -28,7 +28,7 @@ use std::cmp::max;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicI32, Ordering};
 
-pub struct InnerWriteBatch {
+pub(crate) struct InnerWriteBatch {
     batch_id: i64,
     physical_table_path: Arc<PhysicalTablePath>,
     create_ms: i64,
@@ -112,7 +112,7 @@ impl InnerWriteBatch {
     }
 }
 
-pub enum WriteBatch {
+pub(crate) enum WriteBatch {
     ArrowLog(ArrowLogWriteBatch),
     Kv(KvWriteBatch),
 }
@@ -236,7 +236,7 @@ impl WriteBatch {
     }
 }
 
-pub struct ArrowLogWriteBatch {
+pub(crate) struct ArrowLogWriteBatch {
     pub write_batch: InnerWriteBatch,
     pub arrow_builder: MemoryLogRecordsArrowBuilder,
     built_records: Option<Bytes>,
@@ -326,7 +326,7 @@ impl ArrowLogWriteBatch {
     }
 }
 
-pub struct KvWriteBatch {
+pub(crate) struct KvWriteBatch {
     write_batch: InnerWriteBatch,
     kv_batch_builder: KvRecordBatchBuilder,
     target_columns: Option<Arc<Vec<usize>>>,
diff --git a/crates/fluss/src/client/write/broadcast.rs b/crates/fluss/src/client/write/broadcast.rs
index 9e004035..30e6d0c0 100644
--- a/crates/fluss/src/client/write/broadcast.rs
+++ b/crates/fluss/src/client/write/broadcast.rs
@@ -20,12 +20,12 @@ use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::Notify;
 
-pub type Result<T, E = Error> = std::result::Result<T, E>;
+pub(crate) type Result<T, E = Error> = std::result::Result<T, E>;
 
-pub type BatchWriteResult = Result<(), Error>;
+pub(crate) type BatchWriteResult = Result<(), Error>;
 
 #[derive(Debug, Error, Clone, PartialEq, Eq)]
-pub enum Error {
+pub(crate) enum Error {
     #[error("BroadcastOnce dropped")]
     Dropped,
     #[error("Write failed: {message} (code {code})")]
@@ -35,7 +35,7 @@ pub enum Error {
 }
 
 #[derive(Debug, Clone)]
-pub struct BroadcastOnceReceiver<T> {
+pub(crate) struct BroadcastOnceReceiver<T> {
     shared: Arc<Shared<T>>,
 }
 
@@ -78,7 +78,7 @@ struct Shared<T> {
 }
 
 #[derive(Debug)]
-pub struct BroadcastOnce<T>
+pub(crate) struct BroadcastOnce<T>
 where
     T: Send + Sync,
 {
diff --git a/crates/fluss/src/client/write/bucket_assigner.rs b/crates/fluss/src/client/write/bucket_assigner.rs
index 0868a902..10931810 100644
--- a/crates/fluss/src/client/write/bucket_assigner.rs
+++ b/crates/fluss/src/client/write/bucket_assigner.rs
@@ -26,7 +26,7 @@ use rand::Rng;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicI32, Ordering};
 
-pub trait BucketAssigner: Sync + Send {
+pub(crate) trait BucketAssigner: Sync + Send {
     fn abort_if_batch_full(&self) -> bool;
 
     fn on_new_batch(&self, cluster: &Cluster, prev_bucket_id: BucketId);
@@ -35,7 +35,7 @@ pub trait BucketAssigner: Sync + Send {
 }
 
 #[derive(Debug)]
-pub struct StickyBucketAssigner {
+pub(crate) struct StickyBucketAssigner {
     table_path: Arc<PhysicalTablePath>,
     current_bucket_id: AtomicI32,
 }
@@ -109,7 +109,7 @@ impl BucketAssigner for StickyBucketAssigner {
 
 /// Unlike [StickyBucketAssigner], each record is assigned to the next bucket
 /// in a rotating sequence, providing even data distribution across all buckets.
-pub struct RoundRobinBucketAssigner {
+pub(crate) struct RoundRobinBucketAssigner {
     table_path: Arc<PhysicalTablePath>,
     num_buckets: i32,
     counter: AtomicI32,
@@ -146,7 +146,7 @@ impl BucketAssigner for RoundRobinBucketAssigner {
 }
 
 /// A [BucketAssigner] which assigns based on a modulo hashing function
-pub struct HashBucketAssigner {
+pub(crate) struct HashBucketAssigner {
     num_buckets: i32,
     bucketing_function: Box<dyn BucketingFunction>,
 }
diff --git a/crates/fluss/src/client/write/idempotence.rs b/crates/fluss/src/client/write/idempotence.rs
index eeec8761..b3a94bda 100644
--- a/crates/fluss/src/client/write/idempotence.rs
+++ b/crates/fluss/src/client/write/idempotence.rs
@@ -48,7 +48,7 @@ impl BucketEntry {
     }
 }
 
-pub struct IdempotenceManager {
+pub(crate) struct IdempotenceManager {
     writer_id: AtomicI64,
     bucket_entries: Mutex<HashMap<TableBucket, BucketEntry>>,
     enabled: bool,
diff --git a/crates/fluss/src/client/write/mod.rs b/crates/fluss/src/client/write/mod.rs
index a65b5d5a..c27a52d8 100644
--- a/crates/fluss/src/client/write/mod.rs
+++ b/crates/fluss/src/client/write/mod.rs
@@ -25,7 +25,7 @@ use crate::error::Error;
 use crate::metadata::{PhysicalTablePath, TableInfo};
 
 use crate::row::InternalRow;
-pub use accumulator::*;
+pub(crate) use accumulator::*;
 use arrow::array::RecordBatch;
 use bytes::Bytes;
 use std::future::Future;
@@ -41,11 +41,11 @@ mod write_format;
 mod writer_client;
 
 pub(crate) use idempotence::IdempotenceManager;
-pub use write_format::WriteFormat;
+pub(crate) use write_format::WriteFormat;
 pub(crate) use writer_client::WriterClient;
 
 #[allow(dead_code)]
-pub struct WriteRecord<'a> {
+pub(crate) struct WriteRecord<'a> {
     record: Record<'a>,
     physical_table_path: Arc<PhysicalTablePath>,
     bucket_key: Option<Bytes>,
@@ -80,18 +80,18 @@ impl<'a> WriteRecord<'a> {
     }
 }
 
-pub enum Record<'a> {
+pub(crate) enum Record<'a> {
     Log(LogWriteRecord<'a>),
     Kv(KvWriteRecord<'a>),
 }
 
-pub enum LogWriteRecord<'a> {
+pub(crate) enum LogWriteRecord<'a> {
     InternalRow(&'a dyn InternalRow),
     RecordBatch(Arc<RecordBatch>),
 }
 
 #[derive(Clone)]
-pub enum RowBytes<'a> {
+pub(crate) enum RowBytes<'a> {
     Borrowed(&'a [u8]),
     Owned(Bytes),
 }
@@ -105,7 +105,7 @@ impl<'a> RowBytes<'a> {
     }
 }
 
-pub struct KvWriteRecord<'a> {
+pub(crate) struct KvWriteRecord<'a> {
     key: Bytes,
     target_columns: Option<Arc<Vec<usize>>>,
     row_bytes: Option<RowBytes<'a>>,
@@ -185,7 +185,7 @@ impl<'a> WriteRecord<'a> {
 }
 
 #[derive(Debug, Clone)]
-pub struct ResultHandle {
+pub(crate) struct ResultHandle {
     receiver: BroadcastOnceReceiver<BatchWriteResult>,
 }
 
diff --git a/crates/fluss/src/client/write/sender.rs b/crates/fluss/src/client/write/sender.rs
index 2a4f34e6..7b00c4fb 100644
--- a/crates/fluss/src/client/write/sender.rs
+++ b/crates/fluss/src/client/write/sender.rs
@@ -53,7 +53,7 @@ type DrainResult<'a> = (
 );
 
 #[allow(dead_code)]
-pub struct Sender {
+pub(crate) struct Sender {
     running: AtomicBool,
     metadata: Arc<Metadata>,
     accumulator: Arc<RecordAccumulator>,
diff --git a/crates/fluss/src/client/write/write_format.rs b/crates/fluss/src/client/write/write_format.rs
index 147152ca..10284ce5 100644
--- a/crates/fluss/src/client/write/write_format.rs
+++ b/crates/fluss/src/client/write/write_format.rs
@@ -21,7 +21,7 @@ use crate::metadata::KvFormat;
 use std::fmt::Display;
 
 #[derive(Copy, Clone)]
-pub enum WriteFormat {
+pub(crate) enum WriteFormat {
     ArrowLog,
     CompactedLog,
     CompactedKv,
diff --git a/crates/fluss/src/client/write/writer_client.rs b/crates/fluss/src/client/write/writer_client.rs
index 73d91c7c..4bf7af27 100644
--- a/crates/fluss/src/client/write/writer_client.rs
+++ b/crates/fluss/src/client/write/writer_client.rs
@@ -40,7 +40,7 @@ use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 
 #[allow(dead_code)]
-pub struct WriterClient {
+pub(crate) struct WriterClient {
     config: Config,
     max_request_size: i32,
     accumulate: Arc<RecordAccumulator>,
diff --git a/crates/fluss/src/cluster/cluster.rs b/crates/fluss/src/cluster/cluster.rs
index 4370db67..68dfd5a7 100644
--- a/crates/fluss/src/cluster/cluster.rs
+++ b/crates/fluss/src/cluster/cluster.rs
@@ -30,7 +30,7 @@ use std::sync::Arc;
 static EMPTY: Vec<BucketLocation> = Vec::new();
 
 #[derive(Default)]
-pub struct Cluster {
+pub(crate) struct Cluster {
     coordinator_server: Option<ServerNode>,
     alive_tablet_servers_by_id: HashMap<i32, ServerNode>,
     alive_tablet_servers: Vec<ServerNode>,
@@ -45,7 +45,7 @@ pub struct Cluster {
 
 impl Cluster {
     #[allow(clippy::too_many_arguments)]
-    pub fn new(
+    pub(crate) fn new(
         coordinator_server: Option<ServerNode>,
         alive_tablet_servers_by_id: HashMap<i32, ServerNode>,
         available_locations_by_path: HashMap<Arc<PhysicalTablePath>, Vec<BucketLocation>>,
@@ -77,7 +77,7 @@ impl Cluster {
         }
     }
 
-    pub fn invalidate_server(&self, server_id: &i32, table_ids: Vec<TableId>) -> Self {
+    pub(crate) fn invalidate_server(&self, server_id: &i32, table_ids: Vec<TableId>) -> Self {
         let alive_tablet_servers_by_id = self
             .alive_tablet_servers_by_id
             .iter()
@@ -104,7 +104,7 @@ impl Cluster {
         )
     }
 
-    pub fn invalidate_physical_table_meta(
+    pub(crate) fn invalidate_physical_table_meta(
         &self,
         physical_tables_to_invalid: &HashSet<PhysicalTablePath>,
     ) -> Self {
@@ -126,7 +126,7 @@ impl Cluster {
         )
     }
 
-    pub fn update(&mut self, cluster: Cluster) {
+    pub(crate) fn update(&mut self, cluster: Cluster) {
         let Cluster {
             coordinator_server,
             alive_tablet_servers_by_id,
@@ -177,7 +177,7 @@ impl Cluster {
         (available_locations_by_path, available_locations_by_bucket)
     }
 
-    pub fn from_metadata_response(
+    pub(crate) fn from_metadata_response(
         metadata_response: MetadataResponse,
         origin_cluster: Option<&Cluster>,
     ) -> Result<Cluster> {
@@ -291,11 +291,11 @@ impl Cluster {
         ))
     }
 
-    pub fn get_coordinator_server(&self) -> Option<&ServerNode> {
+    pub(crate) fn get_coordinator_server(&self) -> Option<&ServerNode> {
         self.coordinator_server.as_ref()
     }
 
-    pub fn leader_for(&self, table_bucket: &TableBucket) -> Option<&ServerNode> {
+    pub(crate) fn leader_for(&self, table_bucket: &TableBucket) -> Option<&ServerNode> {
         let location = self.available_locations_by_bucket.get(table_bucket);
         if let Some(location) = location {
             location.leader().as_ref()
@@ -304,11 +304,11 @@ impl Cluster {
         }
     }
 
-    pub fn get_tablet_server(&self, id: i32) -> Option<&ServerNode> {
+    pub(crate) fn get_tablet_server(&self, id: i32) -> Option<&ServerNode> {
         self.alive_tablet_servers_by_id.get(&id)
     }
 
-    pub fn get_table_bucket(
+    pub(crate) fn get_table_bucket(
         &self,
         physical_table_path: &PhysicalTablePath,
         bucket_id: BucketId,
@@ -330,37 +330,37 @@ impl Cluster {
         ))
     }
 
-    pub fn get_partition_id(&self, physical_table_path: &PhysicalTablePath) -> Option<PartitionId> {
+    pub(crate) fn get_partition_id(&self, physical_table_path: &PhysicalTablePath) -> Option<PartitionId> {
         self.partitions_id_by_path.get(physical_table_path).copied()
     }
 
-    pub fn get_partition_name(&self, partition_id: PartitionId) -> Option<&String> {
+    pub(crate) fn get_partition_name(&self, partition_id: PartitionId) -> Option<&String> {
         self.partition_name_by_id.get(&partition_id)
     }
 
-    pub fn get_table_id(&self, table_path: &TablePath) -> Option<TableId> {
+    pub(crate) fn get_table_id(&self, table_path: &TablePath) -> Option<TableId> {
         self.table_id_by_path.get(table_path).copied()
     }
 
-    pub fn get_bucket_locations_by_path(
+    pub(crate) fn get_bucket_locations_by_path(
         &self,
     ) -> &HashMap<Arc<PhysicalTablePath>, Vec<BucketLocation>> {
         &self.available_locations_by_path
     }
 
-    pub fn get_table_info_by_path(&self) -> &HashMap<TablePath, TableInfo> {
+    pub(crate) fn get_table_info_by_path(&self) -> &HashMap<TablePath, TableInfo> {
         &self.table_info_by_path
     }
 
-    pub fn get_table_id_by_path(&self) -> &HashMap<TablePath, TableId> {
+    pub(crate) fn get_table_id_by_path(&self) -> &HashMap<TablePath, TableId> {
         &self.table_id_by_path
     }
 
-    pub fn get_table_path_by_id(&self, table_id: TableId) -> Option<&TablePath> {
+    pub(crate) fn get_table_path_by_id(&self, table_id: TableId) -> Option<&TablePath> {
         self.table_path_by_id.get(&table_id)
     }
 
-    pub fn get_available_buckets_for_table_path(
+    pub(crate) fn get_available_buckets_for_table_path(
         &self,
         table_path: &PhysicalTablePath,
     ) -> &Vec<BucketLocation> {
@@ -369,7 +369,7 @@ impl Cluster {
             .unwrap_or(&EMPTY)
     }
 
-    pub fn get_server_nodes(&self) -> Vec<ServerNode> {
+    pub(crate) fn get_server_nodes(&self) -> Vec<ServerNode> {
         let mut nodes = Vec::new();
         if let Some(coordinator) = &self.coordinator_server {
             nodes.push(coordinator.clone());
@@ -378,7 +378,7 @@ impl Cluster {
         nodes
     }
 
-    pub fn get_one_available_server(&self) -> Option<&ServerNode> {
+    pub(crate) fn get_one_available_server(&self) -> Option<&ServerNode> {
         if self.alive_tablet_servers.is_empty() {
             return None;
         }
@@ -386,24 +386,24 @@ impl Cluster {
         self.alive_tablet_servers.get(offset)
     }
 
-    pub fn get_bucket_count(&self, table_path: &TablePath) -> i32 {
+    pub(crate) fn get_bucket_count(&self, table_path: &TablePath) -> i32 {
         self.table_info_by_path
             .get(table_path)
             .unwrap_or_else(|| panic!("can't not table info by path {table_path}"))
             .num_buckets
     }
 
-    pub fn get_table(&self, table_path: &TablePath) -> Result<&TableInfo> {
+    pub(crate) fn get_table(&self, table_path: &TablePath) -> Result<&TableInfo> {
         self.table_info_by_path
             .get(table_path)
             .ok_or_else(|| Error::invalid_table(format!("Table info not found for {table_path}")))
     }
 
-    pub fn opt_get_table(&self, table_path: &TablePath) -> Option<&TableInfo> {
+    pub(crate) fn opt_get_table(&self, table_path: &TablePath) -> Option<&TableInfo> {
         self.table_info_by_path.get(table_path)
     }
 
-    pub fn get_partition_id_by_path(&self) -> &HashMap<Arc<PhysicalTablePath>, PartitionId> {
+    pub(crate) fn get_partition_id_by_path(&self) -> &HashMap<Arc<PhysicalTablePath>, PartitionId> {
         &self.partitions_id_by_path
     }
 }
diff --git a/crates/fluss/src/cluster/mod.rs b/crates/fluss/src/cluster/mod.rs
index d300f7f9..59969940 100644
--- a/crates/fluss/src/cluster/mod.rs
+++ b/crates/fluss/src/cluster/mod.rs
@@ -23,7 +23,7 @@ use std::sync::Arc;
 #[allow(clippy::module_inception)]
 mod cluster;
 
-pub use cluster::Cluster;
+pub(crate) use cluster::Cluster;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct ServerNode {
@@ -35,7 +35,7 @@ pub struct ServerNode {
 }
 
 impl ServerNode {
-    pub fn new(id: i32, host: String, port: u32, server_type: ServerType) -> ServerNode {
+    pub(crate) fn new(id: i32, host: String, port: u32, server_type: ServerType) -> ServerNode {
         let uid = match &server_type {
             ServerType::CoordinatorServer => format!("cs-{id}"),
             ServerType::TabletServer => format!("ts-{id}"),
@@ -83,7 +83,7 @@ pub enum ServerType {
 }
 
 impl ServerType {
-    pub fn to_type_id(&self) -> i32 {
+    pub(crate) fn to_type_id(&self) -> i32 {
         match self {
             ServerType::CoordinatorServer => 1,
             ServerType::TabletServer => 2,
@@ -91,7 +91,7 @@ impl ServerType {
         }
     }
 
-    pub fn from_type_id(type_id: i32) -> ServerType {
+    pub(crate) fn from_type_id(type_id: i32) -> ServerType {
         match type_id {
             1 => ServerType::CoordinatorServer,
             2 => ServerType::TabletServer,
@@ -111,14 +111,14 @@ impl fmt::Display for ServerType {
 }
 
 #[derive(Debug, Clone)]
-pub struct BucketLocation {
-    pub table_bucket: TableBucket,
+pub(crate) struct BucketLocation {
+    pub(crate) table_bucket: TableBucket,
     leader: Option<ServerNode>,
     physical_table_path: Arc<PhysicalTablePath>,
 }
 
 impl BucketLocation {
-    pub fn new(
+    pub(crate) fn new(
         table_bucket: TableBucket,
         leader: Option<ServerNode>,
         physical_table_path: Arc<PhysicalTablePath>,
@@ -130,19 +130,19 @@ impl BucketLocation {
         }
     }
 
-    pub fn leader(&self) -> &Option<ServerNode> {
+    pub(crate) fn leader(&self) -> &Option<ServerNode> {
         &self.leader
     }
 
-    pub fn table_bucket(&self) -> &TableBucket {
+    pub(crate) fn table_bucket(&self) -> &TableBucket {
         &self.table_bucket
     }
 
-    pub fn bucket_id(&self) -> BucketId {
+    pub(crate) fn bucket_id(&self) -> BucketId {
         self.table_bucket.bucket_id()
     }
 
-    pub fn physical_table_path(&self) -> &Arc<PhysicalTablePath> {
+    pub(crate) fn physical_table_path(&self) -> &Arc<PhysicalTablePath> {
         &self.physical_table_path
     }
 }
diff --git a/crates/fluss/src/compression/mod.rs b/crates/fluss/src/compression/mod.rs
index 29923c0a..28da28e5 100644
--- a/crates/fluss/src/compression/mod.rs
+++ b/crates/fluss/src/compression/mod.rs
@@ -18,5 +18,5 @@
 mod arrow_compression;
 mod arrow_compression_ratio_estimator;
 
-pub use arrow_compression::*;
-pub use arrow_compression_ratio_estimator::*;
+pub(crate) use arrow_compression::*;
+pub(crate) use arrow_compression_ratio_estimator::*;
diff --git a/crates/fluss/src/config.rs b/crates/fluss/src/config.rs
index cad8d9cb..5be25413 100644
--- a/crates/fluss/src/config.rs
+++ b/crates/fluss/src/config.rs
@@ -324,13 +324,13 @@ impl Config {
     /// Returns true when the security protocol indicates SASL authentication
     /// should be performed. Matches Java's `SaslAuthenticationPlugin` which
     /// registers as `"sasl"` (case-insensitive).
-    pub fn is_sasl_enabled(&self) -> bool {
+    pub(crate) fn is_sasl_enabled(&self) -> bool {
         self.security_protocol.eq_ignore_ascii_case("sasl")
     }
     /// Validates security configuration. Returns `Ok(())` when the config is
     /// consistent, or an error message when SASL is enabled but the config is
     /// incomplete or uses an unsupported mechanism.
-    pub fn validate_security(&self) -> Result<(), String> {
+    pub(crate) fn validate_security(&self) -> Result<(), String> {
         if !self.is_sasl_enabled() {
             return Ok(());
         }
@@ -352,7 +352,7 @@ impl Config {
         }
         Ok(())
     }
-    pub fn validate_scanner(&self) -> Result<(), String> {
+    pub(crate) fn validate_scanner(&self) -> Result<(), String> {
         if self.scanner_remote_log_prefetch_num == 0 {
             return Err("scanner_remote_log_prefetch_num must be > 0".to_string());
         }
@@ -390,7 +390,7 @@ impl Config {
         Ok(())
     }
 
-    pub fn validate_writer(&self) -> Result<(), String> {
+    pub(crate) fn validate_writer(&self) -> Result<(), String> {
         if self.writer_request_max_size <= 0 {
             return Err("writer_request_max_size must be > 0".to_string());
         }
diff --git a/crates/fluss/src/error.rs b/crates/fluss/src/error.rs
index 4bd0690e..cbdcc639 100644
--- a/crates/fluss/src/error.rs
+++ b/crates/fluss/src/error.rs
@@ -133,7 +133,7 @@ pub enum Error {
 /// These create `FlussAPIError` with the correct protocol error code,
 /// consistent with Java where e.g. `InvalidTableException` always carries code 15.
 impl Error {
-    pub fn table_not_exist(message: impl Into<String>) -> Self {
+    pub(crate) fn table_not_exist(message: impl Into<String>) -> Self {
         Error::FlussAPIError {
             api_error: ApiError {
                 code: FlussError::TableNotExist.code(),
@@ -142,7 +142,7 @@ impl Error {
         }
     }
 
-    pub fn invalid_table(message: impl Into<String>) -> Self {
+    pub(crate) fn invalid_table(message: impl Into<String>) -> Self {
         Error::FlussAPIError {
             api_error: ApiError {
                 code: FlussError::InvalidTableException.code(),
@@ -151,7 +151,7 @@ impl Error {
         }
     }
 
-    pub fn partition_not_exist(message: impl Into<String>) -> Self {
+    pub(crate) fn partition_not_exist(message: impl Into<String>) -> Self {
         Error::FlussAPIError {
             api_error: ApiError {
                 code: FlussError::PartitionNotExists.code(),
@@ -160,7 +160,7 @@ impl Error {
         }
     }
 
-    pub fn invalid_partition(message: impl Into<String>) -> Self {
+    pub(crate) fn invalid_partition(message: impl Into<String>) -> Self {
         Error::FlussAPIError {
             api_error: ApiError {
                 code: FlussError::PartitionSpecInvalidException.code(),
@@ -169,7 +169,7 @@ impl Error {
         }
     }
 
-    pub fn leader_not_available(message: impl Into<String>) -> Self {
+    pub(crate) fn leader_not_available(message: impl Into<String>) -> Self {
         Error::FlussAPIError {
             api_error: ApiError {
                 code: FlussError::LeaderNotAvailableException.code(),
diff --git a/crates/fluss/src/io/mod.rs b/crates/fluss/src/io/mod.rs
index 74265017..c6f53cb4 100644
--- a/crates/fluss/src/io/mod.rs
+++ b/crates/fluss/src/io/mod.rs
@@ -18,10 +18,10 @@
 
 mod file_io;
 
-pub use file_io::*;
+pub(crate) use file_io::*;
 
 mod storage;
-pub use storage::*;
+pub(crate) use storage::*;
 
 #[cfg(feature = "storage-fs")]
 mod storage_fs;
diff --git a/crates/fluss/src/lib.rs b/crates/fluss/src/lib.rs
index 62d4c57d..da33ad29 100644
--- a/crates/fluss/src/lib.rs
+++ b/crates/fluss/src/lib.rs
@@ -136,7 +136,7 @@ pub mod metrics;
 
 mod bucketing;
 mod compression;
-pub mod io;
+mod io;
 mod util;
 
 #[cfg(test)]
diff --git a/crates/fluss/src/metadata/acl.rs b/crates/fluss/src/metadata/acl.rs
index f69ec952..161d8374 100644
--- a/crates/fluss/src/metadata/acl.rs
+++ b/crates/fluss/src/metadata/acl.rs
@@ -30,7 +30,7 @@ pub enum ResourceType {
 }
 
 impl ResourceType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Any => 1,
             Self::Cluster => 2,
@@ -39,7 +39,7 @@ impl ResourceType {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             1 => Ok(Self::Any),
             2 => Ok(Self::Cluster),
@@ -66,7 +66,7 @@ pub enum OperationType {
 }
 
 impl OperationType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Any => 1,
             Self::All => 2,
@@ -79,7 +79,7 @@ impl OperationType {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             1 => Ok(Self::Any),
             2 => Ok(Self::All),
@@ -104,14 +104,14 @@ pub enum PermissionType {
 }
 
 impl PermissionType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Any => 1,
             Self::Allow => 2,
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             1 => Ok(Self::Any),
             2 => Ok(Self::Allow),
@@ -135,7 +135,7 @@ pub struct AclInfo {
 }
 
 impl AclInfo {
-    pub fn to_pb(&self) -> PbAclInfo {
+    pub(crate) fn to_pb(&self) -> PbAclInfo {
         PbAclInfo {
             resource_name: self.resource_name.clone(),
             resource_type: self.resource_type.to_i32(),
@@ -147,7 +147,7 @@ impl AclInfo {
         }
     }
 
-    pub fn from_pb(pb: &PbAclInfo) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbAclInfo) -> Result<Self> {
         Ok(Self {
             resource_name: pb.resource_name.clone(),
             resource_type: ResourceType::try_from_i32(pb.resource_type)?,
@@ -174,7 +174,7 @@ pub struct AclFilter {
 }
 
 impl AclFilter {
-    pub fn to_pb(&self) -> PbAclFilter {
+    pub(crate) fn to_pb(&self) -> PbAclFilter {
         PbAclFilter {
             resource_name: self.resource_name.clone(),
             resource_type: self.resource_type.to_i32(),
@@ -186,7 +186,7 @@ impl AclFilter {
         }
     }
 
-    pub fn from_pb(pb: &PbAclFilter) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbAclFilter) -> Result<Self> {
         Ok(Self {
             resource_name: pb.resource_name.clone(),
             resource_type: ResourceType::try_from_i32(pb.resource_type)?,
@@ -213,7 +213,7 @@ pub struct AclError {
 }
 
 impl CreateAclResult {
-    pub fn from_pb(pb: &PbCreateAclRespInfo) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbCreateAclRespInfo) -> Result<Self> {
         Ok(Self {
             acl: AclInfo::from_pb(&pb.acl)?,
             error: pb.error_code.map(|code| AclError {
@@ -233,7 +233,7 @@ pub struct DropAclMatchingAcl {
 }
 
 impl DropAclMatchingAcl {
-    pub fn from_pb(pb: &PbDropAclsMatchingAcl) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbDropAclsMatchingAcl) -> Result<Self> {
         Ok(Self {
             acl: AclInfo::from_pb(&pb.acl)?,
             error: pb.error_code.map(|code| AclError {
@@ -253,7 +253,7 @@ pub struct DropAclsFilterResult {
 }
 
 impl DropAclsFilterResult {
-    pub fn from_pb(pb: &PbDropAclsFilterResult) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbDropAclsFilterResult) -> Result<Self> {
         let matching_acls = pb
             .matching_acls
             .iter()
diff --git a/crates/fluss/src/metadata/cluster_health.rs b/crates/fluss/src/metadata/cluster_health.rs
index 938d1f5f..5a259230 100644
--- a/crates/fluss/src/metadata/cluster_health.rs
+++ b/crates/fluss/src/metadata/cluster_health.rs
@@ -28,7 +28,7 @@ pub enum ClusterHealthStatus {
 }
 
 impl ClusterHealthStatus {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Green => 0,
             Self::Yellow => 1,
@@ -37,7 +37,7 @@ impl ClusterHealthStatus {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::Green),
             1 => Ok(Self::Yellow),
@@ -61,7 +61,7 @@ pub struct ClusterHealth {
 }
 
 impl ClusterHealth {
-    pub fn from_pb(pb: &GetClusterHealthResponse) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &GetClusterHealthResponse) -> Result<Self> {
         Ok(Self {
             num_replicas: pb.num_replicas,
             in_sync_replicas: pb.in_sync_replicas,
diff --git a/crates/fluss/src/metadata/config.rs b/crates/fluss/src/metadata/config.rs
index 4dd7d8bc..ba7859b9 100644
--- a/crates/fluss/src/metadata/config.rs
+++ b/crates/fluss/src/metadata/config.rs
@@ -28,7 +28,7 @@ pub enum AlterConfigOpType {
 }
 
 impl AlterConfigOpType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Set => 0,
             Self::Delete => 1,
@@ -37,7 +37,7 @@ impl AlterConfigOpType {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::Set),
             1 => Ok(Self::Delete),
@@ -71,7 +71,7 @@ impl AlterConfig {
         }
     }
 
-    pub fn to_pb(&self) -> PbAlterConfig {
+    pub(crate) fn to_pb(&self) -> PbAlterConfig {
         PbAlterConfig {
             config_key: self.config_key.clone(),
             config_value: self.config_value.clone(),
@@ -79,7 +79,7 @@ impl AlterConfig {
         }
     }
 
-    pub fn from_pb(pb: &PbAlterConfig) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbAlterConfig) -> Result<Self> {
         Ok(Self {
             config_key: pb.config_key.clone(),
             config_value: pb.config_value.clone(),
@@ -98,7 +98,7 @@ pub struct DescribeConfig {
 }
 
 impl DescribeConfig {
-    pub fn from_pb(pb: &PbDescribeConfig) -> Self {
+    pub(crate) fn from_pb(pb: &PbDescribeConfig) -> Self {
         Self {
             config_key: pb.config_key.clone(),
             config_value: pb.config_value.clone(),
@@ -106,7 +106,7 @@ impl DescribeConfig {
         }
     }
 
-    pub fn to_pb(&self) -> PbDescribeConfig {
+    pub(crate) fn to_pb(&self) -> PbDescribeConfig {
         PbDescribeConfig {
             config_key: self.config_key.clone(),
             config_value: self.config_value.clone(),
diff --git a/crates/fluss/src/metadata/database.rs b/crates/fluss/src/metadata/database.rs
index 76fe5114..68d11d31 100644
--- a/crates/fluss/src/metadata/database.rs
+++ b/crates/fluss/src/metadata/database.rs
@@ -190,7 +190,7 @@ impl JsonSerde for DatabaseDescriptor {
 
 impl DatabaseDescriptor {
     /// Create DatabaseDescriptor from JSON bytes (equivalent to Java's fromJsonBytes)
-    pub fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
+    pub(crate) fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
         let json_value: Value = serde_json::from_slice(bytes).map_err(|e| JsonSerdeError {
             message: format!("Failed to parse JSON: {e}"),
         })?;
@@ -198,7 +198,7 @@ impl DatabaseDescriptor {
     }
 
     /// Convert DatabaseDescriptor to JSON bytes
-    pub fn to_json_bytes(&self) -> Result<Vec<u8>> {
+    pub(crate) fn to_json_bytes(&self) -> Result<Vec<u8>> {
         let json_value = self.serialize_json()?;
         serde_json::to_vec(&json_value).map_err(|e| JsonSerdeError {
             message: format!("Failed to serialize to JSON: {e}"),
@@ -215,7 +215,7 @@ pub struct DatabaseSummary {
 }
 
 impl DatabaseSummary {
-    pub fn from_pb(pb: &PbDatabaseSummary) -> Self {
+    pub(crate) fn from_pb(pb: &PbDatabaseSummary) -> Self {
         Self {
             database_name: pb.database_name.clone(),
             created_time: pb.created_time,
diff --git a/crates/fluss/src/metadata/goal_type.rs b/crates/fluss/src/metadata/goal_type.rs
index 3880faae..24361b4f 100644
--- a/crates/fluss/src/metadata/goal_type.rs
+++ b/crates/fluss/src/metadata/goal_type.rs
@@ -26,7 +26,7 @@ pub enum GoalType {
 }
 
 impl GoalType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::ReplicaDistribution => 0,
             Self::LeaderDistribution => 1,
@@ -34,7 +34,7 @@ impl GoalType {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::ReplicaDistribution),
             1 => Ok(Self::LeaderDistribution),
diff --git a/crates/fluss/src/metadata/json_serde.rs b/crates/fluss/src/metadata/json_serde.rs
index b08159ae..76b934d1 100644
--- a/crates/fluss/src/metadata/json_serde.rs
+++ b/crates/fluss/src/metadata/json_serde.rs
@@ -32,7 +32,7 @@ pub trait JsonSerde: Sized {
 }
 
 impl DataType {
-    pub fn to_type_root(&self) -> &str {
+    pub(crate) fn to_type_root(&self) -> &str {
         match &self {
             DataType::Boolean(_) => "BOOLEAN",
             DataType::TinyInt(_) => "TINYINT",
diff --git a/crates/fluss/src/metadata/kv_snapshot.rs b/crates/fluss/src/metadata/kv_snapshot.rs
index 054254ad..9a2dcb6a 100644
--- a/crates/fluss/src/metadata/kv_snapshot.rs
+++ b/crates/fluss/src/metadata/kv_snapshot.rs
@@ -32,7 +32,7 @@ pub struct KvSnapshot {
 }
 
 impl KvSnapshot {
-    pub fn from_pb(pb: &PbKvSnapshot) -> Self {
+    pub(crate) fn from_pb(pb: &PbKvSnapshot) -> Self {
         Self {
             bucket_id: pb.bucket_id,
             snapshot_id: pb.snapshot_id,
@@ -50,7 +50,7 @@ pub struct LatestKvSnapshots {
 }
 
 impl LatestKvSnapshots {
-    pub fn from_pb(pb: &GetLatestKvSnapshotsResponse) -> Self {
+    pub(crate) fn from_pb(pb: &GetLatestKvSnapshotsResponse) -> Self {
         Self {
             table_id: pb.table_id,
             partition_id: pb.partition_id,
@@ -72,7 +72,7 @@ pub struct RemotePathAndLocalFile {
 }
 
 impl RemotePathAndLocalFile {
-    pub fn from_pb(pb: &PbRemotePathAndLocalFile) -> Self {
+    pub(crate) fn from_pb(pb: &PbRemotePathAndLocalFile) -> Self {
         Self {
             remote_path: pb.remote_path.clone(),
             local_file_name: pb.local_file_name.clone(),
@@ -88,7 +88,7 @@ pub struct KvSnapshotMetadata {
 }
 
 impl KvSnapshotMetadata {
-    pub fn from_pb(pb: &GetKvSnapshotMetadataResponse) -> Self {
+    pub(crate) fn from_pb(pb: &GetKvSnapshotMetadataResponse) -> Self {
         Self {
             log_offset: pb.log_offset,
             snapshot_files: pb
@@ -108,7 +108,7 @@ pub struct AcquireKvSnapshotLeaseResult {
 }
 
 impl AcquireKvSnapshotLeaseResult {
-    pub fn from_pb(pb: &AcquireKvSnapshotLeaseResponse) -> Self {
+    pub(crate) fn from_pb(pb: &AcquireKvSnapshotLeaseResponse) -> Self {
         Self {
             unavailable_snapshots: pb
                 .unavailable_snapshots
@@ -128,7 +128,7 @@ pub struct ActiveKvSnapshots {
 }
 
 impl ActiveKvSnapshots {
-    pub fn from_pb(pb: &ListKvSnapshotsResponse) -> Self {
+    pub(crate) fn from_pb(pb: &ListKvSnapshotsResponse) -> Self {
         Self {
             table_id: pb.table_id,
             partition_id: pb.partition_id,
diff --git a/crates/fluss/src/metadata/kv_snapshot_lease.rs b/crates/fluss/src/metadata/kv_snapshot_lease.rs
index 98638f5d..153c03ad 100644
--- a/crates/fluss/src/metadata/kv_snapshot_lease.rs
+++ b/crates/fluss/src/metadata/kv_snapshot_lease.rs
@@ -27,7 +27,7 @@ pub struct KvSnapshotLeaseForBucket {
 }
 
 impl KvSnapshotLeaseForBucket {
-    pub fn to_pb(&self) -> PbKvSnapshotLeaseForBucket {
+    pub(crate) fn to_pb(&self) -> PbKvSnapshotLeaseForBucket {
         PbKvSnapshotLeaseForBucket {
             partition_id: self.partition_id,
             bucket_id: self.bucket_id,
@@ -35,7 +35,7 @@ impl KvSnapshotLeaseForBucket {
         }
     }
 
-    pub fn from_pb(pb: &PbKvSnapshotLeaseForBucket) -> Self {
+    pub(crate) fn from_pb(pb: &PbKvSnapshotLeaseForBucket) -> Self {
         Self {
             partition_id: pb.partition_id,
             bucket_id: pb.bucket_id,
@@ -52,7 +52,7 @@ pub struct KvSnapshotLeaseForTable {
 }
 
 impl KvSnapshotLeaseForTable {
-    pub fn to_pb(&self) -> PbKvSnapshotLeaseForTable {
+    pub(crate) fn to_pb(&self) -> PbKvSnapshotLeaseForTable {
         PbKvSnapshotLeaseForTable {
             table_id: self.table_id,
             bucket_snapshots: self
@@ -63,7 +63,7 @@ impl KvSnapshotLeaseForTable {
         }
     }
 
-    pub fn from_pb(pb: &PbKvSnapshotLeaseForTable) -> Self {
+    pub(crate) fn from_pb(pb: &PbKvSnapshotLeaseForTable) -> Self {
         Self {
             table_id: pb.table_id,
             bucket_snapshots: pb
diff --git a/crates/fluss/src/metadata/lake_snapshot.rs b/crates/fluss/src/metadata/lake_snapshot.rs
index 5ff50ad6..ffe473eb 100644
--- a/crates/fluss/src/metadata/lake_snapshot.rs
+++ b/crates/fluss/src/metadata/lake_snapshot.rs
@@ -28,7 +28,7 @@ pub struct LakeBucketSnapshot {
 }
 
 impl LakeBucketSnapshot {
-    pub fn from_pb(pb: &PbLakeSnapshotForBucket) -> Self {
+    pub(crate) fn from_pb(pb: &PbLakeSnapshotForBucket) -> Self {
         Self {
             partition_id: pb.partition_id,
             bucket_id: pb.bucket_id,
@@ -49,7 +49,7 @@ pub struct LakeSnapshotInfo {
 }
 
 impl LakeSnapshotInfo {
-    pub fn from_pb(pb: &GetLakeSnapshotResponse) -> Self {
+    pub(crate) fn from_pb(pb: &GetLakeSnapshotResponse) -> Self {
         Self {
             table_id: pb.table_id,
             snapshot_id: pb.snapshot_id,
diff --git a/crates/fluss/src/metadata/partition.rs b/crates/fluss/src/metadata/partition.rs
index c63fe296..ef1c982b 100644
--- a/crates/fluss/src/metadata/partition.rs
+++ b/crates/fluss/src/metadata/partition.rs
@@ -45,7 +45,7 @@ impl PartitionSpec {
         &self.partition_spec
     }
 
-    pub fn to_pb(&self) -> PbPartitionSpec {
+    pub(crate) fn to_pb(&self) -> PbPartitionSpec {
         PbPartitionSpec {
             partition_key_values: self
                 .partition_spec
@@ -58,7 +58,7 @@ impl PartitionSpec {
         }
     }
 
-    pub fn from_pb(pb: &PbPartitionSpec) -> Self {
+    pub(crate) fn from_pb(pb: &PbPartitionSpec) -> Self {
         let partition_spec = pb
             .partition_key_values
             .iter()
@@ -212,7 +212,7 @@ impl ResolvedPartitionSpec {
         Ok(expected_partition_name == other_partition_name)
     }
 
-    pub fn to_pb(&self) -> PbPartitionSpec {
+    pub(crate) fn to_pb(&self) -> PbPartitionSpec {
         PbPartitionSpec {
             partition_key_values: self
                 .partition_keys
@@ -226,7 +226,7 @@ impl ResolvedPartitionSpec {
         }
     }
 
-    pub fn from_pb(pb: &PbPartitionSpec) -> Self {
+    pub(crate) fn from_pb(pb: &PbPartitionSpec) -> Self {
         let partition_keys = pb
             .partition_key_values
             .iter()
@@ -296,7 +296,7 @@ impl PartitionInfo {
         self.partition_spec.to_partition_spec()
     }
 
-    pub fn to_pb(&self) -> PbPartitionInfo {
+    pub(crate) fn to_pb(&self) -> PbPartitionInfo {
         PbPartitionInfo {
             partition_id: self.partition_id,
             partition_spec: self.partition_spec.to_pb(),
@@ -304,7 +304,7 @@ impl PartitionInfo {
         }
     }
 
-    pub fn from_pb(pb: &PbPartitionInfo) -> Self {
+    pub(crate) fn from_pb(pb: &PbPartitionInfo) -> Self {
         Self {
             partition_id: pb.partition_id,
             partition_spec: ResolvedPartitionSpec::from_pb(&pb.partition_spec),
diff --git a/crates/fluss/src/metadata/producer_offsets.rs b/crates/fluss/src/metadata/producer_offsets.rs
index f367c390..23d1ec3b 100644
--- a/crates/fluss/src/metadata/producer_offsets.rs
+++ b/crates/fluss/src/metadata/producer_offsets.rs
@@ -27,7 +27,7 @@ pub struct BucketOffset {
 }
 
 impl BucketOffset {
-    pub fn to_pb(&self) -> PbBucketOffset {
+    pub(crate) fn to_pb(&self) -> PbBucketOffset {
         PbBucketOffset {
             partition_id: self.partition_id,
             bucket_id: self.bucket_id,
@@ -35,7 +35,7 @@ impl BucketOffset {
         }
     }
 
-    pub fn from_pb(pb: &PbBucketOffset) -> Self {
+    pub(crate) fn from_pb(pb: &PbBucketOffset) -> Self {
         Self {
             partition_id: pb.partition_id,
             bucket_id: pb.bucket_id,
@@ -52,7 +52,7 @@ pub struct ProducerTableOffsets {
 }
 
 impl ProducerTableOffsets {
-    pub fn to_pb(&self) -> PbProducerTableOffsets {
+    pub(crate) fn to_pb(&self) -> PbProducerTableOffsets {
         PbProducerTableOffsets {
             table_id: self.table_id,
             bucket_offsets: self
@@ -63,7 +63,7 @@ impl ProducerTableOffsets {
         }
     }
 
-    pub fn from_pb(pb: &PbProducerTableOffsets) -> Self {
+    pub(crate) fn from_pb(pb: &PbProducerTableOffsets) -> Self {
         Self {
             table_id: pb.table_id,
             bucket_offsets: pb
@@ -84,7 +84,7 @@ pub struct ProducerOffsets {
 }
 
 impl ProducerOffsets {
-    pub fn from_pb(pb: &GetProducerOffsetsResponse) -> Self {
+    pub(crate) fn from_pb(pb: &GetProducerOffsetsResponse) -> Self {
         Self {
             producer_id: pb.producer_id.clone(),
             expiration_time: pb.expiration_time,
diff --git a/crates/fluss/src/metadata/rebalance.rs b/crates/fluss/src/metadata/rebalance.rs
index 13826f9e..fdce2113 100644
--- a/crates/fluss/src/metadata/rebalance.rs
+++ b/crates/fluss/src/metadata/rebalance.rs
@@ -34,7 +34,7 @@ pub enum RebalanceStatus {
 }
 
 impl RebalanceStatus {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::NotStarted => 0,
             Self::Rebalancing => 1,
@@ -45,7 +45,7 @@ impl RebalanceStatus {
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::NotStarted),
             1 => Ok(Self::Rebalancing),
@@ -73,7 +73,7 @@ pub struct BucketRebalancePlan {
 }
 
 impl BucketRebalancePlan {
-    pub fn from_pb(pb: &PbRebalancePlanForBucket) -> Self {
+    pub(crate) fn from_pb(pb: &PbRebalancePlanForBucket) -> Self {
         Self {
             partition_id: pb.partition_id,
             bucket_id: pb.bucket_id,
@@ -93,7 +93,7 @@ pub struct BucketRebalanceProgress {
 }
 
 impl BucketRebalanceProgress {
-    pub fn from_pb(pb: &PbRebalanceProgressForBucket) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbRebalanceProgressForBucket) -> Result<Self> {
         Ok(Self {
             rebalance_plan: BucketRebalancePlan::from_pb(&pb.rebalance_plan),
             rebalance_status: RebalanceStatus::try_from_i32(pb.rebalance_status)?,
@@ -109,7 +109,7 @@ pub struct TableRebalanceProgress {
 }
 
 impl TableRebalanceProgress {
-    pub fn from_pb(pb: &PbRebalanceProgressForTable) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbRebalanceProgressForTable) -> Result<Self> {
         Ok(Self {
             table_id: pb.table_id,
             buckets_progress: pb
@@ -130,7 +130,7 @@ pub struct RebalanceProgress {
 }
 
 impl RebalanceProgress {
-    pub fn from_pb(pb: &ListRebalanceProgressResponse) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &ListRebalanceProgressResponse) -> Result<Self> {
         Ok(Self {
             rebalance_id: pb.rebalance_id.clone(),
             rebalance_status: pb
diff --git a/crates/fluss/src/metadata/register_producer_result.rs b/crates/fluss/src/metadata/register_producer_result.rs
index 1c0c8e50..25dafa5d 100644
--- a/crates/fluss/src/metadata/register_producer_result.rs
+++ b/crates/fluss/src/metadata/register_producer_result.rs
@@ -27,14 +27,14 @@ pub enum RegisterProducerResult {
 }
 
 impl RegisterProducerResult {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Created => 0,
             Self::AlreadyExists => 1,
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::Created),
             1 => Ok(Self::AlreadyExists),
diff --git a/crates/fluss/src/metadata/remote_log.rs b/crates/fluss/src/metadata/remote_log.rs
index d3fd3e33..4960388b 100644
--- a/crates/fluss/src/metadata/remote_log.rs
+++ b/crates/fluss/src/metadata/remote_log.rs
@@ -27,7 +27,7 @@ pub struct RemoteLogManifestEntry {
 }
 
 impl RemoteLogManifestEntry {
-    pub fn from_pb(pb: &PbRemoteLogManifestEntry) -> Self {
+    pub(crate) fn from_pb(pb: &PbRemoteLogManifestEntry) -> Self {
         Self {
             table_bucket: TableBucket::from_pb(&pb.table_bucket),
             remote_log_manifest_path: pb.remote_log_manifest_path.clone(),
diff --git a/crates/fluss/src/metadata/server_tag.rs b/crates/fluss/src/metadata/server_tag.rs
index 642bbd6a..37a3e707 100644
--- a/crates/fluss/src/metadata/server_tag.rs
+++ b/crates/fluss/src/metadata/server_tag.rs
@@ -25,14 +25,14 @@ pub enum ServerTag {
 }
 
 impl ServerTag {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::PermanentOffline => 0,
             Self::TemporaryOffline => 1,
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::PermanentOffline),
             1 => Ok(Self::TemporaryOffline),
diff --git a/crates/fluss/src/metadata/table.rs b/crates/fluss/src/metadata/table.rs
index a796ed98..66997c84 100644
--- a/crates/fluss/src/metadata/table.rs
+++ b/crates/fluss/src/metadata/table.rs
@@ -1464,7 +1464,7 @@ impl TableBucket {
         self.partition_id
     }
 
-    pub fn to_pb(&self) -> crate::proto::PbTableBucket {
+    pub(crate) fn to_pb(&self) -> crate::proto::PbTableBucket {
         crate::proto::PbTableBucket {
             table_id: self.table_id,
             partition_id: self.partition_id,
@@ -1472,7 +1472,7 @@ impl TableBucket {
         }
     }
 
-    pub fn from_pb(pb: &crate::proto::PbTableBucket) -> Self {
+    pub(crate) fn from_pb(pb: &crate::proto::PbTableBucket) -> Self {
         Self {
             table_id: pb.table_id,
             partition_id: pb.partition_id,
diff --git a/crates/fluss/src/metadata/table_change.rs b/crates/fluss/src/metadata/table_change.rs
index 3e5866a2..6da649e1 100644
--- a/crates/fluss/src/metadata/table_change.rs
+++ b/crates/fluss/src/metadata/table_change.rs
@@ -30,13 +30,13 @@ pub enum ColumnPositionType {
 }
 
 impl ColumnPositionType {
-    pub fn to_i32(self) -> i32 {
+    pub(crate) fn to_i32(self) -> i32 {
         match self {
             Self::Last => 0,
         }
     }
 
-    pub fn try_from_i32(value: i32) -> Result<Self> {
+    pub(crate) fn try_from_i32(value: i32) -> Result<Self> {
         match value {
             0 => Ok(Self::Last),
             _ => Err(Error::IllegalArgument {
@@ -60,7 +60,7 @@ pub struct AddColumn {
 }
 
 impl AddColumn {
-    pub fn to_pb(&self) -> PbAddColumn {
+    pub(crate) fn to_pb(&self) -> PbAddColumn {
         PbAddColumn {
             column_name: self.column_name.clone(),
             data_type_json: self.data_type_json.clone(),
@@ -69,7 +69,7 @@ impl AddColumn {
         }
     }
 
-    pub fn from_pb(pb: &PbAddColumn) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbAddColumn) -> Result<Self> {
         Ok(Self {
             column_name: pb.column_name.clone(),
             data_type_json: pb.data_type_json.clone(),
@@ -86,13 +86,13 @@ pub struct DropColumn {
 }
 
 impl DropColumn {
-    pub fn to_pb(&self) -> PbDropColumn {
+    pub(crate) fn to_pb(&self) -> PbDropColumn {
         PbDropColumn {
             column_name: self.column_name.clone(),
         }
     }
 
-    pub fn from_pb(pb: &PbDropColumn) -> Self {
+    pub(crate) fn from_pb(pb: &PbDropColumn) -> Self {
         Self {
             column_name: pb.column_name.clone(),
         }
@@ -107,14 +107,14 @@ pub struct RenameColumn {
 }
 
 impl RenameColumn {
-    pub fn to_pb(&self) -> PbRenameColumn {
+    pub(crate) fn to_pb(&self) -> PbRenameColumn {
         PbRenameColumn {
             old_column_name: self.old_column_name.clone(),
             new_column_name: self.new_column_name.clone(),
         }
     }
 
-    pub fn from_pb(pb: &PbRenameColumn) -> Self {
+    pub(crate) fn from_pb(pb: &PbRenameColumn) -> Self {
         Self {
             old_column_name: pb.old_column_name.clone(),
             new_column_name: pb.new_column_name.clone(),
@@ -146,7 +146,7 @@ pub struct ModifyColumn {
 }
 
 impl ModifyColumn {
-    pub fn to_pb(&self) -> PbModifyColumn {
+    pub(crate) fn to_pb(&self) -> PbModifyColumn {
         PbModifyColumn {
             column_name: self.column_name.clone(),
             data_type_json: self.data_type_json.clone(),
@@ -155,7 +155,7 @@ impl ModifyColumn {
         }
     }
 
-    pub fn from_pb(pb: &PbModifyColumn) -> Result<Self> {
+    pub(crate) fn from_pb(pb: &PbModifyColumn) -> Result<Self> {
         Ok(Self {
             column_name: pb.column_name.clone(),
             data_type_json: pb.data_type_json.clone(),
diff --git a/crates/fluss/src/metadata/table_stats.rs b/crates/fluss/src/metadata/table_stats.rs
index 53c6f72f..2c94ff68 100644
--- a/crates/fluss/src/metadata/table_stats.rs
+++ b/crates/fluss/src/metadata/table_stats.rs
@@ -34,14 +34,14 @@ impl BucketStatsRequest {
         }
     }
 
-    pub fn to_pb(&self) -> PbTableStatsReqForBucket {
+    pub(crate) fn to_pb(&self) -> PbTableStatsReqForBucket {
         PbTableStatsReqForBucket {
             partition_id: self.partition_id,
             bucket_id: self.bucket_id,
         }
     }
 
-    pub fn from_pb(pb: &PbTableStatsReqForBucket) -> Self {
+    pub(crate) fn from_pb(pb: &PbTableStatsReqForBucket) -> Self {
         Self {
             partition_id: pb.partition_id,
             bucket_id: pb.bucket_id,
@@ -66,7 +66,7 @@ pub struct BucketStatsError {
 }
 
 impl BucketStats {
-    pub fn from_pb(pb: &PbTableStatsRespForBucket) -> Self {
+    pub(crate) fn from_pb(pb: &PbTableStatsRespForBucket) -> Self {
         let error = pb.error_code.map(|code| BucketStatsError {
             code,
             message: pb.error_message.clone(),
@@ -87,7 +87,7 @@ pub struct TableStats {
 }
 
 impl TableStats {
-    pub fn from_pb(pb: &GetTableStatsResponse) -> Self {
+    pub(crate) fn from_pb(pb: &GetTableStatsResponse) -> Self {
         Self {
             buckets: pb.buckets_resp.iter().map(BucketStats::from_pb).collect(),
         }
diff --git a/crates/fluss/src/record/arrow.rs b/crates/fluss/src/record/arrow.rs
index dbbbcc99..e5658b6a 100644
--- a/crates/fluss/src/record/arrow.rs
+++ b/crates/fluss/src/record/arrow.rs
@@ -55,50 +55,50 @@ use std::{
 use crate::error::Error::IllegalArgument;
 use arrow::ipc::writer::IpcWriteOptions;
 /// const for record batch
-pub const BASE_OFFSET_LENGTH: usize = 8;
-pub const LENGTH_LENGTH: usize = 4;
-pub const MAGIC_LENGTH: usize = 1;
-pub const COMMIT_TIMESTAMP_LENGTH: usize = 8;
-pub const CRC_LENGTH: usize = 4;
-pub const SCHEMA_ID_LENGTH: usize = 2;
-pub const ATTRIBUTE_LENGTH: usize = 1;
-pub const LAST_OFFSET_DELTA_LENGTH: usize = 4;
-pub const WRITE_CLIENT_ID_LENGTH: usize = 8;
-pub const BATCH_SEQUENCE_LENGTH: usize = 4;
-pub const RECORDS_COUNT_LENGTH: usize = 4;
-
-pub const BASE_OFFSET_OFFSET: usize = 0;
-pub const LENGTH_OFFSET: usize = BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH;
-pub const MAGIC_OFFSET: usize = LENGTH_OFFSET + LENGTH_LENGTH;
-pub const COMMIT_TIMESTAMP_OFFSET: usize = MAGIC_OFFSET + MAGIC_LENGTH;
-pub const CRC_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
-pub const SCHEMA_ID_OFFSET: usize = CRC_OFFSET + CRC_LENGTH;
-pub const ATTRIBUTES_OFFSET: usize = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH;
-pub const LAST_OFFSET_DELTA_OFFSET: usize = ATTRIBUTES_OFFSET + ATTRIBUTE_LENGTH;
-pub const WRITE_CLIENT_ID_OFFSET: usize = LAST_OFFSET_DELTA_OFFSET + LAST_OFFSET_DELTA_LENGTH;
-pub const BATCH_SEQUENCE_OFFSET: usize = WRITE_CLIENT_ID_OFFSET + WRITE_CLIENT_ID_LENGTH;
-pub const RECORDS_COUNT_OFFSET: usize = BATCH_SEQUENCE_OFFSET + BATCH_SEQUENCE_LENGTH;
-pub const RECORDS_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
-
-pub const RECORD_BATCH_HEADER_SIZE: usize = RECORDS_OFFSET;
-pub const ARROW_CHANGETYPE_OFFSET: usize = RECORD_BATCH_HEADER_SIZE;
-pub const LOG_OVERHEAD: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+pub(crate) const BASE_OFFSET_LENGTH: usize = 8;
+pub(crate) const LENGTH_LENGTH: usize = 4;
+pub(crate) const MAGIC_LENGTH: usize = 1;
+pub(crate) const COMMIT_TIMESTAMP_LENGTH: usize = 8;
+pub(crate) const CRC_LENGTH: usize = 4;
+pub(crate) const SCHEMA_ID_LENGTH: usize = 2;
+pub(crate) const ATTRIBUTE_LENGTH: usize = 1;
+pub(crate) const LAST_OFFSET_DELTA_LENGTH: usize = 4;
+pub(crate) const WRITE_CLIENT_ID_LENGTH: usize = 8;
+pub(crate) const BATCH_SEQUENCE_LENGTH: usize = 4;
+pub(crate) const RECORDS_COUNT_LENGTH: usize = 4;
+
+pub(crate) const BASE_OFFSET_OFFSET: usize = 0;
+pub(crate) const LENGTH_OFFSET: usize = BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH;
+pub(crate) const MAGIC_OFFSET: usize = LENGTH_OFFSET + LENGTH_LENGTH;
+pub(crate) const COMMIT_TIMESTAMP_OFFSET: usize = MAGIC_OFFSET + MAGIC_LENGTH;
+pub(crate) const CRC_OFFSET: usize = COMMIT_TIMESTAMP_OFFSET + COMMIT_TIMESTAMP_LENGTH;
+pub(crate) const SCHEMA_ID_OFFSET: usize = CRC_OFFSET + CRC_LENGTH;
+pub(crate) const ATTRIBUTES_OFFSET: usize = SCHEMA_ID_OFFSET + SCHEMA_ID_LENGTH;
+pub(crate) const LAST_OFFSET_DELTA_OFFSET: usize = ATTRIBUTES_OFFSET + ATTRIBUTE_LENGTH;
+pub(crate) const WRITE_CLIENT_ID_OFFSET: usize = LAST_OFFSET_DELTA_OFFSET + LAST_OFFSET_DELTA_LENGTH;
+pub(crate) const BATCH_SEQUENCE_OFFSET: usize = WRITE_CLIENT_ID_OFFSET + WRITE_CLIENT_ID_LENGTH;
+pub(crate) const RECORDS_COUNT_OFFSET: usize = BATCH_SEQUENCE_OFFSET + BATCH_SEQUENCE_LENGTH;
+pub(crate) const RECORDS_OFFSET: usize = RECORDS_COUNT_OFFSET + RECORDS_COUNT_LENGTH;
+
+pub(crate) const RECORD_BATCH_HEADER_SIZE: usize = RECORDS_OFFSET;
+pub(crate) const ARROW_CHANGETYPE_OFFSET: usize = RECORD_BATCH_HEADER_SIZE;
+pub(crate) const LOG_OVERHEAD: usize = LENGTH_OFFSET + LENGTH_LENGTH;
 
 /// Bit 0 of the attributes byte. When set, the batch is append-only and carries
 /// no change-type vector; when clear, a `record_count`-byte change-type vector
 /// precedes the Arrow IPC payload (the changelog of a primary-key table). Shares
 /// the wire layout of the Java client's `DefaultLogRecordBatch`.
-pub const APPEND_ONLY_FLAG_MASK: u8 = 0x01;
+pub(crate) const APPEND_ONLY_FLAG_MASK: u8 = 0x01;
 
 /// Maximum batch size matches Java's Integer.MAX_VALUE limit.
 /// Java uses int type for batch size, so max value is 2^31 - 1 = 2,147,483,647 bytes (~2GB).
 /// This is the implicit limit in FileLogRecords.java and other Java components.
-pub const MAX_BATCH_SIZE: usize = i32::MAX as usize; // 2,147,483,647 bytes (~2GB)
+pub(crate) const MAX_BATCH_SIZE: usize = i32::MAX as usize; // 2,147,483,647 bytes (~2GB)
 
 /// const for record
 /// The "magic" values.
 #[derive(Debug, Clone, Copy)]
-pub enum LogMagicValue {
+pub(crate) enum LogMagicValue {
     V0 = 0,
 }
 
@@ -145,15 +145,15 @@ fn validate_batch_size(batch_size_bytes: i32) -> Result<usize> {
 
 // NOTE: Rust layout/offsets currently match Java only for V0.
 // TODO: Add V1 layout/offsets to keep parity with Java's V1 format.
-pub const CURRENT_LOG_MAGIC_VALUE: u8 = LogMagicValue::V0 as u8;
+pub(crate) const CURRENT_LOG_MAGIC_VALUE: u8 = LogMagicValue::V0 as u8;
 
 /// Value used if writer ID is not available or non-idempotent.
-pub const NO_WRITER_ID: i64 = -1;
+pub(crate) const NO_WRITER_ID: i64 = -1;
 
 /// Value used if batch sequence is not available.
-pub const NO_BATCH_SEQUENCE: i32 = -1;
+pub(crate) const NO_BATCH_SEQUENCE: i32 = -1;
 
-pub const BUILDER_DEFAULT_OFFSET: i64 = 0;
+pub(crate) const BUILDER_DEFAULT_OFFSET: i64 = 0;
 
 /// Initial capacity for Arrow column vectors (pre-allocation hint, not a record cap).
 /// Matching Java's `ArrowWriter.INITIAL_CAPACITY`.
@@ -163,7 +163,7 @@ const INITIAL_ROW_CAPACITY: usize = 1024;
 /// Matching Java's `ArrowWriter.BUFFER_USAGE_RATIO`.
 const BUFFER_USAGE_RATIO: f32 = 0.95;
 
-pub struct MemoryLogRecordsArrowBuilder {
+pub(crate) struct MemoryLogRecordsArrowBuilder {
     base_log_offset: i64,
     schema_id: i32,
     magic: u8,
@@ -190,7 +190,7 @@ pub struct MemoryLogRecordsArrowBuilder {
     estimated_compression_ratio: f32,
 }
 
-pub trait ArrowRecordBatchInnerBuilder: Send {
+pub(crate) trait ArrowRecordBatchInnerBuilder: Send {
     fn build_arrow_record_batch(&mut self) -> Result<Arc<RecordBatch>>;
 
     fn append(&mut self, row: &dyn InternalRow) -> Result<bool>;
@@ -208,7 +208,7 @@ pub trait ArrowRecordBatchInnerBuilder: Send {
 }
 
 #[derive(Default)]
-pub struct PrebuiltRecordBatchBuilder {
+pub(crate) struct PrebuiltRecordBatchBuilder {
     arrow_record_batch: Option<Arc<RecordBatch>>,
     records_count: i32,
 }
@@ -253,7 +253,7 @@ impl ArrowRecordBatchInnerBuilder for PrebuiltRecordBatchBuilder {
     }
 }
 
-pub struct RowAppendRecordBatchBuilder {
+pub(crate) struct RowAppendRecordBatchBuilder {
     table_schema: SchemaRef,
     column_writers: Vec<ColumnWriter>,
     records_count: i32,
@@ -631,7 +631,7 @@ fn estimate_arrow_ipc_overhead(
     Ok(ipc_message_len.saturating_sub(raw_data))
 }
 
-pub trait ToArrow {
+pub(crate) trait ToArrow {
     fn append_to(&self, builder: &mut dyn ArrayBuilder) -> Result<()>;
 }
 
@@ -834,7 +834,7 @@ impl LogRecordsSource {
     }
 }
 
-pub struct LogRecordsBatches {
+pub(crate) struct LogRecordsBatches {
     source: LogRecordsSource,
     current_pos: usize,
     remaining_bytes: usize,
@@ -910,7 +910,7 @@ impl Iterator for LogRecordsBatches {
     }
 }
 
-pub struct LogRecordBatch {
+pub(crate) struct LogRecordBatch {
     data: Bytes,
 }
 
@@ -1433,7 +1433,7 @@ pub(crate) fn from_arrow_type(arrow_type: &ArrowDataType) -> Result<DataType> {
 }
 
 #[derive(Clone)]
-pub struct ReadContext {
+pub(crate) struct ReadContext {
     target_schema: SchemaRef,
     full_schema: SchemaRef,
     row_type: Arc<RowType>,
@@ -1740,7 +1740,7 @@ fn align_record_batch_to_schema(
     Ok(RecordBatch::try_new(target_schema, columns)?)
 }
 
-pub enum LogRecordIterator {
+pub(crate) enum LogRecordIterator {
     Empty,
     Arrow(ArrowLogRecordIterator),
 }
@@ -1784,7 +1784,7 @@ impl BatchChangeTypes {
     }
 }
 
-pub struct ArrowLogRecordIterator {
+pub(crate) struct ArrowLogRecordIterator {
     reader: ArrowReader,
     base_offset: i64,
     timestamp: i64,
@@ -1842,7 +1842,7 @@ impl Iterator for ArrowLogRecordIterator {
     }
 }
 
-pub struct ArrowReader {
+pub(crate) struct ArrowReader {
     batch: Arc<TypedBatch>,
 }
 
@@ -1871,7 +1871,7 @@ impl ArrowReader {
         ColumnarRow::from_typed_batch(Arc::clone(&self.batch), row_id)
     }
 }
-pub struct MyVec<T>(pub StreamReader<T>);
+pub(crate) struct MyVec<T>(pub(crate) StreamReader<T>);
 
 #[cfg(test)]
 mod tests {
diff --git a/crates/fluss/src/record/mod.rs b/crates/fluss/src/record/mod.rs
index 7e548b2c..c2b06cc3 100644
--- a/crates/fluss/src/record/mod.rs
+++ b/crates/fluss/src/record/mod.rs
@@ -23,9 +23,10 @@ use std::collections::HashMap;
 
 mod arrow;
 mod error;
-pub mod kv;
+pub(crate) mod kv;
 
-pub use arrow::*;
+pub use arrow::{from_arrow_field, to_arrow_schema, to_arrow_type};
+pub(crate) use arrow::*;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum ChangeType {
@@ -88,7 +89,7 @@ impl fmt::Display for ChangeType {
 
 #[derive(Clone)]
 pub struct ScanRecord {
-    pub row: ColumnarRow,
+    row: ColumnarRow,
     offset: i64,
     timestamp: i64,
     change_type: ChangeType,
diff --git a/crates/fluss/src/row/mod.rs b/crates/fluss/src/row/mod.rs
index 7b483b4e..4998e476 100644
--- a/crates/fluss/src/row/mod.rs
+++ b/crates/fluss/src/row/mod.rs
@@ -19,20 +19,20 @@ pub mod binary_array;
 pub mod binary_map;
 mod column;
 pub(crate) mod column_vector;
-pub mod columnar;
+pub(crate) mod columnar;
 pub mod view;
 
 pub(crate) mod datum;
 mod decimal;
 
-pub mod binary;
+pub(crate) mod binary;
 pub(crate) mod column_writer;
-pub mod compacted;
-pub mod encode;
-pub mod field_getter;
+pub(crate) mod compacted;
+pub(crate) mod encode;
+pub(crate) mod field_getter;
 mod fixed_schema_decoder;
 mod lookup_row;
-pub mod paimon;
+pub(crate) mod paimon;
 mod projected_row;
 mod row_decoder;
 
@@ -44,26 +44,26 @@ pub use column::*;
 pub use compacted::CompactedRow;
 pub use datum::*;
 pub use decimal::{Decimal, MAX_COMPACT_PRECISION};
-pub use encode::KeyEncoder;
+pub(crate) use encode::KeyEncoder;
 pub(crate) use fixed_schema_decoder::FixedSchemaDecoder;
 pub use lookup_row::LookupRow;
 pub(crate) use projected_row::ProjectedRow;
-pub use row_decoder::{CompactedRowDecoder, RowDecoder, RowDecoderFactory};
+pub(crate) use row_decoder::{CompactedRowDecoder, RowDecoder, RowDecoderFactory};
 use serde::Serialize;
 pub use view::{ArrayView, MapView, RowView};
 
-pub struct BinaryRow<'a> {
+pub(crate) struct BinaryRow<'a> {
     data: BinaryDataWrapper<'a>,
 }
 
-pub enum BinaryDataWrapper<'a> {
+pub(crate) enum BinaryDataWrapper<'a> {
     Bytes(Bytes),
     Ref(&'a [u8]),
 }
 
 impl<'a> BinaryRow<'a> {
     /// Returns the binary representation of this row as a byte slice.
-    pub fn as_bytes(&'a self) -> &'a [u8] {
+    pub(crate) fn as_bytes(&'a self) -> &'a [u8] {
         match &self.data {
             BinaryDataWrapper::Bytes(bytes) => bytes.as_ref(),
             BinaryDataWrapper::Ref(r) => r,
diff --git a/crates/fluss/src/rpc/message/mod.rs b/crates/fluss/src/rpc/message/mod.rs
index aba690a5..9e8b1833 100644
--- a/crates/fluss/src/rpc/message/mod.rs
+++ b/crates/fluss/src/rpc/message/mod.rs
@@ -75,64 +75,65 @@ mod scan_kv;
 mod table_exists;
 mod update_metadata;
 
-pub use crate::rpc::RpcError;
-pub use acquire_kv_snapshot_lease::*;
-pub use add_server_tag::*;
-pub use alter_cluster_configs::*;
-pub use alter_database::*;
-pub use alter_table::*;
-pub use api_versions::*;
-pub use authenticate::*;
-pub use cancel_rebalance::*;
-pub use create_acls::*;
-pub use create_database::*;
-pub use create_partition::*;
-pub use create_table::*;
-pub use database_exists::*;
-pub use delete_producer_offsets::*;
-pub use describe_cluster_configs::*;
-pub use drop_acls::*;
-pub use drop_database::*;
-pub use drop_kv_snapshot_lease::*;
-pub use drop_partition::*;
-pub use drop_table::*;
-pub use fetch::*;
-pub use get_cluster_health::*;
-pub use get_database_info::*;
-pub use get_kv_snapshot_metadata::*;
-pub use get_lake_snapshot::*;
-pub use get_latest_kv_snapshots::*;
-pub use get_latest_lake_snapshot::*;
-pub use get_producer_offsets::*;
-pub use get_security_token::*;
-pub use get_table::*;
-pub use get_table_schema::*;
-pub use get_table_stats::*;
-pub use header::*;
-pub use init_writer::*;
-pub use limit_scan::*;
-pub use list_acls::*;
-pub use list_database_summaries::*;
-pub use list_databases::*;
-pub use list_kv_snapshots::*;
-pub use list_offsets::*;
-pub use list_partition_infos::*;
-pub use list_rebalance_progress::*;
-pub use list_remote_log_manifests::*;
-pub use list_tables::*;
-pub use lookup::*;
-pub use prefix_lookup::*;
-pub use produce_log::*;
-pub use put_kv::*;
-pub use rebalance::*;
-pub use register_producer_offsets::*;
-pub use release_kv_snapshot_lease::*;
-pub use remove_server_tag::*;
-pub use scan_kv::*;
-pub use table_exists::*;
-pub use update_metadata::*;
+pub(crate) use crate::rpc::RpcError;
+pub(crate) use acquire_kv_snapshot_lease::*;
+pub(crate) use add_server_tag::*;
+pub(crate) use alter_cluster_configs::*;
+pub(crate) use alter_database::*;
+pub(crate) use alter_table::*;
+pub(crate) use api_versions::*;
+pub(crate) use authenticate::*;
+pub(crate) use cancel_rebalance::*;
+pub(crate) use create_acls::*;
+pub(crate) use create_database::*;
+pub(crate) use create_partition::*;
+pub(crate) use create_table::*;
+pub(crate) use database_exists::*;
+pub(crate) use delete_producer_offsets::*;
+pub(crate) use describe_cluster_configs::*;
+pub(crate) use drop_acls::*;
+pub(crate) use drop_database::*;
+pub(crate) use drop_kv_snapshot_lease::*;
+pub(crate) use drop_partition::*;
+pub(crate) use drop_table::*;
+pub(crate) use fetch::*;
+pub(crate) use get_cluster_health::*;
+pub(crate) use get_database_info::*;
+pub(crate) use get_kv_snapshot_metadata::*;
+pub(crate) use get_lake_snapshot::*;
+pub(crate) use get_latest_kv_snapshots::*;
+pub(crate) use get_latest_lake_snapshot::*;
+pub(crate) use get_producer_offsets::*;
+pub(crate) use get_security_token::*;
+pub(crate) use get_table::*;
+pub(crate) use get_table_schema::*;
+pub(crate) use get_table_stats::*;
+pub(crate) use header::*;
+pub(crate) use init_writer::*;
+pub(crate) use limit_scan::*;
+pub(crate) use list_acls::*;
+pub(crate) use list_database_summaries::*;
+pub(crate) use list_databases::*;
+pub(crate) use list_kv_snapshots::*;
+pub use list_offsets::OffsetSpec;
+pub(crate) use list_offsets::*;
+pub(crate) use list_partition_infos::*;
+pub(crate) use list_rebalance_progress::*;
+pub(crate) use list_remote_log_manifests::*;
+pub(crate) use list_tables::*;
+pub(crate) use lookup::*;
+pub(crate) use prefix_lookup::*;
+pub(crate) use produce_log::*;
+pub(crate) use put_kv::*;
+pub(crate) use rebalance::*;
+pub(crate) use register_producer_offsets::*;
+pub(crate) use release_kv_snapshot_lease::*;
+pub(crate) use remove_server_tag::*;
+pub(crate) use scan_kv::*;
+pub(crate) use table_exists::*;
+pub(crate) use update_metadata::*;
 
-pub trait RequestBody {
+pub(crate) trait RequestBody {
     type ResponseBody;
 
     const API_KEY: ApiKey;
@@ -144,14 +145,14 @@ impl<T: RequestBody> RequestBody for &T {
     const API_KEY: ApiKey = T::API_KEY;
 }
 
-pub trait WriteType<W>: Sized
+pub(crate) trait WriteType<W>: Sized
 where
     W: BufMut,
 {
     fn write(&self, writer: &mut W) -> Result<(), WriteError>;
 }
 
-pub trait ReadType<R>: Sized
+pub(crate) trait ReadType<R>: Sized
 where
     R: Buf,
 {
diff --git a/crates/fluss/src/rpc/mod.rs b/crates/fluss/src/rpc/mod.rs
index a4b191d8..1a431a63 100644
--- a/crates/fluss/src/rpc/mod.rs
+++ b/crates/fluss/src/rpc/mod.rs
@@ -25,7 +25,7 @@ mod frame;
 pub mod message;
 pub use error::*;
 mod server_connection;
-pub use server_connection::*;
+pub(crate) use server_connection::*;
 mod convert;
 mod transport;
 
diff --git a/crates/fluss/src/util/mod.rs b/crates/fluss/src/util/mod.rs
index ee8dde4a..d4b6e5cb 100644
--- a/crates/fluss/src/util/mod.rs
+++ b/crates/fluss/src/util/mod.rs
@@ -15,9 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
-pub mod murmur_hash;
-pub mod partition;
-pub mod varint;
+pub(crate) mod murmur_hash;
+pub(crate) mod partition;
+pub(crate) mod varint;
 
 use crate::TableId;
 use crate::metadata::TableBucket;
@@ -27,7 +27,7 @@ use std::hash::Hash;
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-pub fn current_time_ms() -> i64 {
+pub(crate) fn current_time_ms() -> i64 {
     SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .unwrap_or(std::time::Duration::ZERO)
@@ -38,7 +38,7 @@ pub fn current_time_ms() -> i64 {
 // File cleanup is now handled via RAII with FileCleanupGuard in arrow.rs
 // which uses Rust's drop order to ensure files are closed before deletion.
 
-pub struct FairBucketStatusMap<S> {
+pub(crate) struct FairBucketStatusMap<S> {
     map: LinkedHashMap<TableBucket, Arc<S>>,
     size: usize,
 }
diff --git a/crates/fluss/tests/integration/utils.rs b/crates/fluss/tests/integration/utils.rs
index 2ebe31d9..42d90a26 100644
--- a/crates/fluss/tests/integration/utils.rs
+++ b/crates/fluss/tests/integration/utils.rs
@@ -260,12 +260,11 @@ pub async fn wait_for_cluster_ready_with_sasl(cluster: &FlussTestingCluster) {
             let connection = cluster
                 .get_fluss_connection_with_sasl(username, password)
                 .await;
-            if connection
-                .get_metadata()
-                .get_cluster()
-                .get_one_available_server()
-                .is_some()
-            {
+            let nodes = match connection.get_admin() {
+                Ok(admin) => admin.get_server_nodes().await.unwrap_or_default(),
+                Err(_) => Vec::new(),
+            };
+            if !nodes.is_empty() {
                 Ok(())
             } else {
                 Err(