[tests] replace fixed tokio::time::sleep workarounds with wait_for_table_ready helper.

crates/fluss/tests/integration/log_table.rs

@@ -22,7 +22,7 @@ mod table_test {
         ColumnPlan, array_dt_basics_columns, as_row_type, create_partitions, create_table,
         dt_array_int, dt_map_string_int, dt_row_seq_label, extract_ids_from_batches,
         get_shared_cluster, make_int_array, make_string_array, map_dt_basics_columns,
-        row_dt_basics_columns, scalar_dt_columns,
+        row_dt_basics_columns, scalar_dt_columns, wait_for_partitions_ready, wait_for_table_ready,
     };
     use arrow::array::record_batch;
     use fluss::client::{EARLIEST_OFFSET, FlussTable, TableScan};
@@ -475,7 +475,7 @@ mod table_test {
             &TableDescriptor::builder().schema(schema).build().unwrap(),
         )
         .await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_ready(&admin, &table_path).await;
 
         let table = connection.get_table(&table_path).await.unwrap();
         let scanner = table.new_scan().create_record_batch_log_scanner().unwrap();
@@ -595,8 +595,8 @@ mod table_test {
         // Create partitions
         create_partitions(&admin, &table_path, "region", &["US", "EU"]).await;
 
-        // Wait for partitions to be available
-        tokio::time::sleep(Duration::from_secs(2)).await;
+        // Wait for partition bucket leaders to be available.
+        wait_for_partitions_ready(&admin, &table_path, &["US", "EU"]).await;
 
         let table = connection
             .get_table(&table_path)

crates/fluss/tests/integration/record_batch_log_reader.rs

@@ -21,6 +21,7 @@
 mod reader_test {
     use crate::integration::utils::{
         create_partitions, create_table, extract_ids_from_batches, get_shared_cluster,
+        wait_for_partitions_ready, wait_for_table_buckets_ready, wait_for_table_ready,
     };
     use arrow::array::record_batch;
     use fluss::client::{EARLIEST_OFFSET, FlussConnection, RecordBatchLogReader};
@@ -48,7 +49,7 @@ mod reader_test {
             .build()
             .expect("Failed to build table");
         create_table(&admin, &table_path, &table_descriptor).await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_ready(&admin, &table_path).await;
 
         let table = connection
             .get_table(&table_path)
@@ -121,7 +122,7 @@ mod reader_test {
             .build()
             .expect("Failed to build table");
         create_table(&admin, &table_path, &table_descriptor).await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_ready(&admin, &table_path).await;
 
         let table = connection
             .get_table(&table_path)
@@ -189,7 +190,7 @@ mod reader_test {
             .build()
             .expect("Failed to build table");
         create_table(&admin, &table_path, &table_descriptor).await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_ready(&admin, &table_path).await;
 
         let table = connection
             .get_table(&table_path)
@@ -284,7 +285,7 @@ mod reader_test {
             .build()
             .expect("Failed to build table");
         create_table(&admin, &table_path, &table_descriptor).await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_buckets_ready(&admin, &table_path, &[0, 1]).await;
 
         let table = connection
             .get_table(&table_path)
@@ -380,7 +381,7 @@ mod reader_test {
             .expect("Failed to build table");
 
         create_table(&admin, &table_path, &table_descriptor).await;
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        wait_for_table_ready(&admin, &table_path).await;
 
         let table = connection
             .get_table(&table_path)
@@ -453,7 +454,7 @@ mod reader_test {
 
         create_table(&admin, &table_path, &table_descriptor).await;
         create_partitions(&admin, &table_path, "region", &["US", "EU"]).await;
-        tokio::time::sleep(Duration::from_secs(2)).await;
+        wait_for_partitions_ready(&admin, &table_path, &["US", "EU"]).await;
 
         let table = connection
             .get_table(&table_path)

crates/fluss/tests/integration/utils.rs

@@ -24,6 +24,7 @@ use fluss::metadata::{
 use fluss::record::ScanBatch;
 use fluss::row::FlussArray;
 use fluss::row::binary_array::FlussArrayWriter;
+use fluss::rpc::message::OffsetSpec;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::LazyLock;
@@ -100,6 +101,94 @@ pub async fn create_table(
         .expect("Failed to create table");
 }
 
+const READINESS_TIMEOUT: Duration = Duration::from_secs(30);
+const READINESS_POLL_INTERVAL: Duration = Duration::from_millis(200);
+
+/// Waits until the default bucket of a non-partitioned table can serve offset requests.
+///
+/// Newly-created tables may not have bucket leaders immediately. Polling list offsets avoids
+/// fixed sleeps that are either flaky on slow CI or waste time when the cluster is ready sooner.
+pub async fn wait_for_table_ready(admin: &FlussAdmin, table_path: &TablePath) {
+    wait_for_table_buckets_ready(admin, table_path, &[0]).await;
+}
+
+/// Waits until the specified buckets of a non-partitioned table can serve offset requests.
+pub async fn wait_for_table_buckets_ready(
+    admin: &FlussAdmin,
+    table_path: &TablePath,
+    buckets: &[i32],
+) {
+    let start = std::time::Instant::now();
+
+    loop {
+        match admin
+            .list_offsets(table_path, buckets, OffsetSpec::Latest)
+            .await
+        {
+            Ok(_) => return,
+            Err(err) => {
+                if start.elapsed() >= READINESS_TIMEOUT {
+                    panic!(
+                        "Timed out waiting for table '{table_path}' buckets {buckets:?} to become ready after {} seconds. Last error: {err:?}",
+                        READINESS_TIMEOUT.as_secs()
+                    );
+                }
+            }
+        }
+
+        tokio::time::sleep(READINESS_POLL_INTERVAL).await;
+    }
+}
+
+/// Waits until all listed partition values can serve offset requests for the default bucket.
+pub async fn wait_for_partitions_ready(
+    admin: &FlussAdmin,
+    table_path: &TablePath,
+    partition_values: &[&str],
+) {
+    for partition_value in partition_values {
+        wait_for_partition_ready(admin, table_path, partition_value).await;
+    }
+}
+
+/// Waits until one partition value can serve offset requests for the default bucket.
+pub async fn wait_for_partition_ready(
+    admin: &FlussAdmin,
+    table_path: &TablePath,
+    partition_value: &str,
+) {
+    wait_for_partition_buckets_ready(admin, table_path, partition_value, &[0]).await;
+}
+
+/// Waits until the specified buckets of a partition can serve offset requests.
+pub async fn wait_for_partition_buckets_ready(
+    admin: &FlussAdmin,
+    table_path: &TablePath,
+    partition_value: &str,
+    buckets: &[i32],
+) {
+    let start = std::time::Instant::now();
+
+    loop {
+        match admin
+            .list_partition_offsets(table_path, partition_value, buckets, OffsetSpec::Latest)
+            .await
+        {
+            Ok(_) => return,
+            Err(err) => {
+                if start.elapsed() >= READINESS_TIMEOUT {
+                    panic!(
+                        "Timed out waiting for table '{table_path}' partition '{partition_value}' buckets {buckets:?} to become ready after {} seconds. Last error: {err:?}",
+                        READINESS_TIMEOUT.as_secs()
+                    );
+                }
+            }
+        }
+
+        tokio::time::sleep(READINESS_POLL_INTERVAL).await;
+    }
+}
+
 pub fn make_string_array(values: &[Option<&str>]) -> FlussArray {
     let mut writer = FlussArrayWriter::new(values.len(), &DataTypes::string());
     for (idx, value) in values.iter().enumerate() {