fix(tests): resolve flaky catalog tests by waiting for S3 bucket creation (#1599)

## Which issue does this PR close?
- Closes #1581

## What changes are included in this PR?
This ensures tests only run after the complete test environment is
initialized correctly, eliminating the timing-based failures.
- Wait for both the catalog service and the MinIO service to be ready
- Verify S3 bucket existence before proceeding with tests
- Add proper retry logic with clear logging for bucket readiness

## Are these changes tested?
Yes

Co-authored-by: Renjie Liu <[email protected]>

Author: fvaleye

crates/catalog/glue/tests/glue_catalog_test.rs

@@ -73,6 +73,11 @@ async fn get_catalog() -> GlueCatalog {
         sleep(std::time::Duration::from_millis(1000)).await;
     }
 
+    while !scan_port_addr(minio_socket_addr) {
+        info!("Waiting for 1s minio to ready...");
+        sleep(std::time::Duration::from_millis(1000)).await;
+    }
+
     let props = HashMap::from([
         (AWS_ACCESS_KEY_ID.to_string(), "my_access_id".to_string()),
         (
@@ -89,6 +94,24 @@ async fn get_catalog() -> GlueCatalog {
         (S3_REGION.to_string(), "us-east-1".to_string()),
     ]);
 
+    // Wait for bucket to actually exist
+    let file_io = iceberg::io::FileIO::from_path("s3a://")
+        .unwrap()
+        .with_props(props.clone())
+        .build()
+        .unwrap();
+
+    let mut retries = 0;
+    while retries < 30 {
+        if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+            info!("S3 bucket 'warehouse' is ready");
+            break;
+        }
+        info!("Waiting for bucket creation... (attempt {})", retries + 1);
+        sleep(std::time::Duration::from_millis(1000)).await;
+        retries += 1;
+    }
+
     let config = GlueCatalogConfig::builder()
         .uri(format!("http://{}", glue_socket_addr))
         .warehouse("s3a://warehouse/hive".to_string())

crates/catalog/hms/tests/hms_catalog_test.rs

@@ -73,6 +73,11 @@ async fn get_catalog() -> HmsCatalog {
         sleep(std::time::Duration::from_millis(1000)).await;
     }
 
+    while !scan_port_addr(minio_socket_addr) {
+        info!("Waiting for 1s minio to ready...");
+        sleep(std::time::Duration::from_millis(1000)).await;
+    }
+
     let props = HashMap::from([
         (
             S3_ENDPOINT.to_string(),
@@ -83,6 +88,24 @@ async fn get_catalog() -> HmsCatalog {
         (S3_REGION.to_string(), "us-east-1".to_string()),
     ]);
 
+    // Wait for bucket to actually exist
+    let file_io = iceberg::io::FileIO::from_path("s3a://")
+        .unwrap()
+        .with_props(props.clone())
+        .build()
+        .unwrap();
+
+    let mut retries = 0;
+    while retries < 30 {
+        if file_io.exists("s3a://warehouse/").await.unwrap_or(false) {
+            info!("S3 bucket 'warehouse' is ready");
+            break;
+        }
+        info!("Waiting for bucket creation... (attempt {})", retries + 1);
+        sleep(std::time::Duration::from_millis(1000)).await;
+        retries += 1;
+    }
+
     let config = HmsCatalogConfig::builder()
         .address(hms_socket_addr.to_string())
         .thrift_transport(HmsThriftTransport::Buffered)